014.多线程-并发包

蔚落 2021-10-06 15:34 317阅读 0赞

在日常开发中，我们最常用的集合和字典莫过于ArrayList和HashMap了。

ArrayList，内部是通过数组实现的，它允许对元素进行快速随机访问。
    	但是，当进行增加存储功能，以及插入、删除元素时，需要对数组进行复制以及移动，代价比较高。
    	它适合随机查找和遍历，不适合插入和删除。

--------------------

HashMap，由数组+链表组成的，数组是HashMap的主体，链表则是主要为了解决哈希冲突而存在的。
    	如果定位到的数组位置不含链表,那么对于查找，添加等操作很快，仅需一次寻址即可；
    	如果定位到的数组包含链表，对于添加操作，其时间复杂度为O(n)，首先遍历链表，存在即覆盖，否则新增；
    	对于查找操作来讲，仍需遍历链表，然后通过key对象的equals方法逐一比对查找。
    	所以，性能考虑，HashMap中的链表出现越少，性能才会越好。

但是，ArrayList和HashMap都是线程不安全的。  
与此同时，相对应的 Vector和HashTable却是线程安全的，但是，效率太低了。

Vector与ArrayList一样，也是通过数组实现的，不同的是它支持线程的同步。
    	Vector的add和get均被synchronized修饰，
    	即：同一时刻，只有一个线程可以读写Vector。

--------------------

HashTable类似于HashMap，只不过HashMap允许null key 和 null value，HashTable不允许。
    	不过它的put和get方法均被synchronized修饰，
    	即：同一时刻，只有一个线程可以读写HashTable。

--------------------

因此为提升效率的并发包应运而生。

### Collections.synchronizedList ###

![idea64_2018-11-19_09-51-44.png][]  
![idea64_2018-11-19_10-05-20.png][]

List<String> list1 = new CopyOnWriteArrayList<String>();
            List<String> list2 = Collections.synchronizedList(new ArrayList<String>());

#### CopyOnWriteArrayList与Collections.synchronizedList性能比较 ####

*  读操作时，CopyOnWriteArrayList性能相对较好些。
 *  写操作时，Collections.synchronizedList性能相对较好些。

--------------------

### ConcurrentHashMap ###

ConcurrentHashMap内部使用段（Segment）来表示不同的部分，  
每个部分都相当于一个HashTable，都拥有自己的锁。  
只要多个操作不同时发生在一个段上，就可以并发进行。  
其中Segment数组长度为16，不可以扩容。  
ConcurrentHashMap的初始化容量会平均分配给每个Segment。  
其中的负载因子是针对每个Segment的，  
当实际容量大于阈值（负载因子乘以Segment容量）时，会扩容当前Segment。

/** * The default load factor for this table, used when not * otherwise specified in a constructor. */
        static final float DEFAULT_LOAD_FACTOR = 0.75f;
    
        /** * The default concurrency level for this table, used when not * otherwise specified in a constructor. */
        static final int DEFAULT_CONCURRENCY_LEVEL = 16;

--------------------

### CountDownLatch ###

利用它可以实现类似计数器的功能。  
比如有一个任务A，它要等待其他4个任务执行完毕之后才能执行，  
此时就可以利用CountDownLatch来实现这种功能了。

package cn.qbz.thread;
    
    import java.util.concurrent.CountDownLatch;
    
    public class Test111901 { 
        public static void main(String[] args) throws InterruptedException { 
            final CountDownLatch countDownLatch = new CountDownLatch(3);
            for (int i = 0; i < 3; i++) { 
                new Thread(new Runnable() { 
                    public void run() { 
                        System.out.println("子线程" + Thread.currentThread().getName() + "开始运行...");
                        for (int i = 0; i < 10; i++) { 
                            try { 
                                Thread.sleep(100);
                            } catch (InterruptedException e) { 
                                e.printStackTrace();
                            }
                        }
                        System.out.println("子线程" + Thread.currentThread().getName() + "结束运行...");
                        countDownLatch.countDown();
                    }
                }).start();
            }
    
            System.out.println("主线程等待子线程执行结束....");
            countDownLatch.await();
            System.out.println("主线程执行结束......");
        }
    }

![idea64_2018-11-19_10-56-35.png][]

如果 `final CountDownLatch countDownLatch = new CountDownLatch(4);`  
主线程将会一直阻塞在 `countDownLatch.await();`

--------------------

### CyclicBarrier ###

CyclicBarrier初始化时规定一个数目，  
然后计算调用了CyclicBarrier.await()进入等待的线程数。  
当线程数达到了这个数目时，所有进入等待状态的线程被唤醒并继续。

package cn.qbz.thread;
    
    import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    
    public class Test111902 { 
        public static void main(String[] args) { 
            CyclicBarrier cyclicBarrier = new CyclicBarrier(4);
            for (int i = 0; i < 4; i++) { 
                new ThreadTest111902(cyclicBarrier).start();
            }
            System.out.println("主线程执行结束......");
        }
    }
    
    class ThreadTest111902 extends Thread { 
        private CyclicBarrier cyclicBarrier;
    
        public ThreadTest111902(CyclicBarrier cyclicBarrier) { 
            this.cyclicBarrier = cyclicBarrier;
        }
    
        @Override
        public void run() { 
            System.out.println("子线程" + getName() + "：开始运行");
            try { 
                Thread.sleep(3000);
    
                System.out.println("子线程" + getName() + "：结束运行");
                cyclicBarrier.await();
    
                System.out.println("子线程" + getName() + "：所有线程执行结束....");
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } catch (BrokenBarrierException e) { 
                e.printStackTrace();
            }
        }
    }

![idea64_2018-11-19_11-13-47.png][]

--------------------

### Semaphore ###

Semaphore是一种基于计数的信号量。  
它可以设定一个阈值，多个线程竞争获取许可信号，  
超过阈值后，线程申请许可信号将会被阻塞。  
Semaphore可以用来构建一些对象池，资源池之类的，比如数据库连接池。  
我们也可以创建计数为1的Semaphore，将其作为一种类似互斥锁的机制，这也叫二元信号量，表示两种互斥状态。  
它的用法如下：  
availablePermits函数用来获取当前可用的资源数量  
wc.acquire(); //申请资源  
wc.release();// 释放资源

// 创建一个计数阈值为5的信号量对象 
       	// 只能5个线程同时访问 
       	Semaphore semp = new Semaphore(5);  
       	  
       	try {   
       	    // 申请许可 
       	    semp.acquire();  
       	    try {   
       	        // 业务逻辑 
       	    } catch (Exception e) {   
       	  
       	    } finally {   
       	        // 释放许可 
       	        semp.release();  
       	    }  
       	} catch (InterruptedException e) {   
       	  
       	}

#### code of semaphore demo: ####

有4个业务办理窗口，等候区共有30人，模拟。

package cn.qbz.thread;
    
    import java.util.concurrent.Semaphore;
    
    public class Test111903 { 
        public static void main(String[] args) { 
            Semaphore semaphore = new Semaphore(4);
            for (int i = 0; i < 30; i++) { 
                new ThreadTest111903(semaphore, "第" + i + "个").start();
            }
            System.out.println("主线程执行结束......");
        }
    }
    
    class ThreadTest111903 extends Thread { 
        private Semaphore semaphore;
        private String name;
    
        public ThreadTest111903(Semaphore semaphore, String name) { 
            this.semaphore = semaphore;
            this.name = name;
        }
    
        @Override
        public void run() { 
            //判断是否可以申请许可
            int availablePermits = semaphore.availablePermits();
            if (availablePermits > 0) { 
                System.out.println(name + "...抢到了窗口");
            } else { 
                System.out.println(name + "...没抢到了窗口");
            }
    
            try { 
                //申请资源，如果没有可用的资源，则阻塞在这里
                //有了资源后，再从这里继续执行
                semaphore.acquire();
                System.out.println(name + "。。。开始办理业务");
                Thread.sleep(Math.round(1000));
                System.out.println(name + "。。。结束办理业务");
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } finally { 
                //释放资源
                semaphore.release();
            }
        }
    }

[idea64_2018-11-19_09-51-44.png]: http://upyun.qubianzhong.cn/picture/idea64_2018-11-19_09-51-44.png
[idea64_2018-11-19_10-05-20.png]: http://upyun.qubianzhong.cn/picture/idea64_2018-11-19_10-05-20.png
[idea64_2018-11-19_10-56-35.png]: http://upyun.qubianzhong.cn/picture/idea64_2018-11-19_10-56-35.png
[idea64_2018-11-19_11-13-47.png]: http://upyun.qubianzhong.cn/picture/idea64_2018-11-19_11-13-47.png