java并发包（第四天）

àì夳堔傛蜴生んèń 2022-05-12 05:04 211阅读 0赞

*  同步容器类
    
     *  Vector与ArrayList区别
        
         *  Vector源码
         *  ArrayList源码
     *  HasTable与HasMap的区别
     *  Collections工具类
     *  ConcurrentHashMap
     *  CountDownLatch
     *  CyclicBarrier
     *  Semaphore
 *  并发队列
    
     *  ConcurrentLinkedDeque（非阻塞）
     *  BlockingQueue（阻塞）
        
         *  ArrayBlockingQueue
         *  LinkedBlockingQueue
         *  PriorityBlockingQueue
         *  SynchronousQueue
     *  使用BlockingQueue模拟生产者与消费者

# 同步容器类 #

## Vector与ArrayList区别 ##

1.  ArrayList是最常用的List实现类，内部是通过数组实现的，它允许对元素进行快速随机访问。数组的缺点是每个元素之间不能有间隔，当数组大小不满足时需要增加存储能力，就要讲已经有数组的数据复制到新的存储空间中。当从ArrayList的中间位置插入或者删除元素时，需要对数组进行复制、移动、代价比较高。因此，它适合随机查找和遍历，不适合插入和删除。
2.  Vector与ArrayList一样，**也是通过数组实现的，不同的是它支持线程的同步**，即某一时刻只有一个线程能够写Vector，避免多线程同时写而引起的不一致性，但实现同步需要很高的花费，因此，访问它比访问ArrayList慢。

### Vector源码 ###

public synchronized boolean add(E e) {
        modCount++;
        ensureCapacityHelper(elementCount + 1);
        elementData[elementCount++] = e;
        return true;
    }

### ArrayList源码 ###

public boolean add(E e) {
        ensureCapacityInternal(size + 1);  // Increments modCount!!
        elementData[size++] = e;
        return true;
    }

## HasTable与HasMap的区别 ##

1.  HashMap不是线程安全的，HastMap是一个接口 是map接口的子接口，是将键映射到值的对象，其中键和值都是对象，并且不能包含重复键，但可以包含重复值。HashMap允许null key和null value，而hashtable不允许。
2.  HashTable是线程安全的一个Collection。
3.  HashMap是Hashtable的轻量级实现（非线程安全的实现），他们都完成了Map接口，主要区别在于HashMap允许空（null）键值（key）,由于非线程安全，效率上可能高于Hashtable。
4.  HashMap允许将null作为一个entry的key或者value，而Hashtable不允许。
5.  HashMap把Hashtable的contains方法去掉了，改成containsvalue和containsKey。

## Collections工具类 ##

1.  使用Collections.synchronizedMap(map);等方法将Collection里面的非线程安全的类转换为线程安全的。里面其实是使用synchronized包装起来。
2.  源代码

public int size() {
                synchronized (mutex) {
       return m.size();}
            }
            public boolean isEmpty() {
                synchronized (mutex) {
       return m.isEmpty();}
            }
            public boolean containsKey(Object key) {
                synchronized (mutex) {
       return m.containsKey(key);}
            }
            public boolean containsValue(Object value) {
                synchronized (mutex) {
       return m.containsValue(value);}
            }
            public V get(Object key) {
                synchronized (mutex) {
       return m.get(key);}
            }
    
            public V put(K key, V value) {
                synchronized (mutex) {
       return m.put(key, value);}
            }
            public V remove(Object key) {
                synchronized (mutex) {
       return m.remove(key);}
            }
            public void putAll(Map<? extends K, ? extends V> map) {
                synchronized (mutex) {m.putAll(map);}
            }
            public void clear() {
                synchronized (mutex) {m.clear();}
            }

## ConcurrentHashMap ##

ConcurrentHashMap内部使用段(Segment)来表示这些不同的部分，每个段其实就是一个小的HashTable,它们有自己的锁。只要多个修改操作发生在不同的段上，它们就可以并发进行。把一个整体分成了16个段(Segment)也就是最高支持16个线程的并发修改操作。  
这也是在多线程场景时减小锁的粒度从而降低锁竞争的一种方案。并且代码中大多共享变量使用volatile关键字声明，目的是第一时间获取修改的内容，性能非常好。

## CountDownLatch ##

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

package top.nightliar.study.day04;
    
    import java.util.concurrent.CountDownLatch;
    
    /** * Created by Nightliar * 2018-09-17 13:49 */
    public class ThreadDemo02 { 
    
        public static void main(String[] args) throws InterruptedException {
            CountDownLatch countDownLatch = new CountDownLatch(2);
            System.out.println("主线程开始");
            new Thread(new Runnable() {
                @Override
                public void run() {
                    System.out.println("子线程1开始");
                    countDownLatch.countDown();  // 每次减去1
                    System.out.println("子线程1结束");
                }
            }).start();
            new Thread(new Runnable() {
                @Override
                public void run() {
                    System.out.println("子线程2开始");
                    countDownLatch.countDown();  // 每次减去1
                    System.out.println("子线程2结束");
                }
            }).start();
            countDownLatch.await();  // 调用当前方法主线程阻塞 countDown结果为0, 阻塞变为运行状态
            System.out.println("两个子线程执行完毕");
            System.out.println("主线程结束");
        }
    
    }

结果：
    主线程开始
    子线程2开始
    子线程1开始
    子线程2结束
    子线程1结束
    两个子线程执行完毕
    主线程结束

## CyclicBarrier ##

CyclicBarrier初始化时规定一个数目，然后计算调用了CyclicBarrier.await()进入等待的线程数。当线程数达到了这个数目时，所有进入等待状态的线程被唤醒并继续。  
CyclicBarrier就象它名字的意思一样，可看成是个障碍， 所有的线程必须到齐后才能一起通过这个障碍。  
CyclicBarrier初始时还可带一个Runnable的参数， 此Runnable任务在CyclicBarrier的数目达到后，所有其它线程被唤醒前被执行。

package top.nightliar.study.day04;
    
    import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    
    /** * Created by Nightliar * 2018-09-17 13:59 */
    public class ThreadDemo03 { 
    
        public static void main(String[] args) {
            CyclicBarrier cyclicBarrier = new CyclicBarrier(4);
            for (int i = 0; i < 4; i++) {
                new CreateThread(cyclicBarrier).start();
            }
        }
    
    }
    
    class CreateThread extends Thread{
    
        CyclicBarrier cyclicBarrier;
    
        public CreateThread(CyclicBarrier cyclicBarrier) {
            this.cyclicBarrier = cyclicBarrier;
        }
    
        @Override
        public void run() {
            System.out.println(getName() + "写入数据开始...");
            try {
                sleep(40);
                System.out.println(getName() + "写入数据完毕");
                cyclicBarrier.await();
                System.out.println(getName() + "所有线程一起执行完毕");
            } catch (InterruptedException | BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    
    }

结果：
    Thread-0写入数据开始...
    Thread-3写入数据开始...
    Thread-2写入数据开始...
    Thread-1写入数据开始...
    Thread-2写入数据完毕
    Thread-3写入数据完毕
    Thread-0写入数据完毕
    Thread-1写入数据完毕
    Thread-2所有线程一起执行完毕
    Thread-3所有线程一起执行完毕
    Thread-0所有线程一起执行完毕
    Thread-1所有线程一起执行完毕

## Semaphore ##

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

package top.nightliar.study.day04;
    
    import java.util.Random;
    import java.util.concurrent.Semaphore;
    
    /** * 10个线程同时运行，但只能同时运行其中三个。 * Created by Nightliar * 2018-09-17 14:32 */
    public class ThreadDemo04 { 
    
        public static void main(String[] args) {
            // 设置3个并发量
            Semaphore semaphore = new Semaphore(3);
            for (int i = 0; i < 11; i++) {
                new CreateThread02("第" + i + "个人，", semaphore).start();
            }
        }
    
    }
    
    class CreateThread02 extends Thread {
    
        Semaphore semaphore;
        String name;
    
        public CreateThread02(String name, Semaphore semaphore) {
            this.semaphore = semaphore;
            this.name = name;
        }
    
        @Override
        public void run() {
            // 获取当前可用信号量
            int availablePermits = semaphore.availablePermits();
            System.out.println("当前可用信号量为：" + availablePermits);
            try {
                // 申请获取资源 如果资源达到3次，就等待
                semaphore.acquire();
                System.out.println(name + "终于抢到资源了！！");
                Thread.sleep(new Random().nextInt(1000)); // 模拟操作资源所需的时间。
                System.out.println(name + "操作资源结束！！");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                // 释放资源
                semaphore.release();
            }
        }
    }

结果：
    当前可用信号量为：3
    当前可用信号量为：3
    第0个人，终于抢到资源了！！
    第1个人，终于抢到资源了！！
    当前可用信号量为：1
    当前可用信号量为：1
    第5个人，终于抢到资源了！！
    当前可用信号量为：0
    当前可用信号量为：0
    当前可用信号量为：0
    当前可用信号量为：0
    当前可用信号量为：0
    当前可用信号量为：0
    当前可用信号量为：0
    第0个人，操作资源结束！！
    第4个人，终于抢到资源了！！
    第1个人，操作资源结束！！
    第7个人，终于抢到资源了！！
    第7个人，操作资源结束！！
    第3个人，终于抢到资源了！！
    第5个人，操作资源结束！！
    第2个人，终于抢到资源了！！
    第2个人，操作资源结束！！
    第8个人，终于抢到资源了！！
    第3个人，操作资源结束！！
    第10个人，终于抢到资源了！！
    第4个人，操作资源结束！！
    第9个人，终于抢到资源了！！
    第8个人，操作资源结束！！
    第6个人，终于抢到资源了！！
    第10个人，操作资源结束！！
    第6个人，操作资源结束！！
    第9个人，操作资源结束！！

# 并发队列 #

## ConcurrentLinkedDeque（非阻塞） ##

ConcurrentLinkedQueue : 是一个适用于高并发场景下的队列，通过无锁的方式，实现了高并发状态下的高性能，通常ConcurrentLinkedQueue性能好BlockingQueue。它是一个**非阻塞，无锁，无界，线程安全双端操作的队列 。**该队列的元素遵循先进先出的原则。头是最先加入的，尾是最近加入的，该队列不允许null元素。  
简单说就是ConcurrentLinkedQueue的升级版，在JDK7之后才提供。该队列也不允许空元素，而且size方法并不是常量时间，其需要遍历链表，此时并发修改链表会造成统计size不正确。同样，bulk操作和equal以及toArray方法不保证原子性。

ConcurrentLinkedDeque<String> deque = new ConcurrentLinkedDeque<>();
    deque.offer("张三");
    deque.offer("李四");
    deque.offer("王五");
    //从头获取元素,删除该元素
    System.out.println(deque.poll());
    //从头获取元素,不刪除该元素
    System.out.println(deque.peek());
    //获取总长度
    System.out.println(deque.size());

## BlockingQueue（阻塞） ##

阻塞队列（BlockingQueue）是一个支持两个附加操作的队列。这两个附加的操作是：  
1. 在队列为空时，获取元素的线程会等待队列变为非空。  
2. 当队列满时，存储元素的线程会等待队列可用。  
java.util.concurrent包提供了几种不同形式的阻塞队列，如数组阻塞队列ArrayBlockingQueue、链表阻塞队列LinkedBlockingQueue、优先级阻塞队列PriorityBlockingQueue和延时队列DelayQueue等。

### ArrayBlockingQueue ###

ArrayBlockingQueue是一个**有边界的阻塞队列，它的内部实现是一个数组。**有边界的意思是它的容量是有限的，我们必须在其初始化的时候指定它的容量大小，容量大小一旦指定就不可改变。  
ArrayBlockingQueue是以**先进先出**的方式存储数据，最新插入的对象是尾部，最新移出的对象是头部。

ArrayBlockingQueue<String> arrays = new ArrayBlockingQueue<String>(3);
    arrays.add("李四");
    arrays.add("张军");
    arrays.add("张军");
    // 添加阻塞队列
    arrays.offer("张三", 1, TimeUnit.SECONDS);

### LinkedBlockingQueue ###

LinkedBlockingQueue阻塞队列大小的配置是可选的，如果我们初始化时指定一个大小，它就是有边界的，如果不指定，它就是无边界的。说是无边界，其实是采用了默认大小为Integer.MAX\_VALUE的容量 。它的内部实现是一个链表。  
和ArrayBlockingQueue一样，LinkedBlockingQueue 也是以先进先出的方式存储数据，最新插入的对象是尾部，最新移出的对象是头部。

LinkedBlockingQueue linkedBlockingQueue = new LinkedBlockingQueue(3);
    linkedBlockingQueue.add("张三");
    linkedBlockingQueue.add("李四");
    linkedBlockingQueue.add("李四");
    System.out.println(linkedBlockingQueue.size());

### PriorityBlockingQueue ###

PriorityBlockingQueue是一个没有边界的队列，排序规则和java.util.PriorityQueue一样。需要注意，PriorityBlockingQueue中允许插入null对象。所有插入PriorityBlockingQueue的对象必须实现 java.lang.Comparable接口，队列优先级的排序规则就 是按照我们对这个接口的实现来定义的。  
另外，我们可以从PriorityBlockingQueue获得一个迭代器Iterator，但这个迭代器并不保证按照优先级顺 序进行迭代。

### SynchronousQueue ###

SynchronousQueue是一个没有数据缓冲的BlockingQueue，生产者线程对其的插入操作put必须等待消费者的移除操作take，反过来也一样。

package top.nightliar.study.day04;
    
    import java.util.concurrent.ConcurrentLinkedDeque;
    import java.util.concurrent.SynchronousQueue;
    
    /** * Created by Nightliar * 2018-09-17 15:45 */
    public class ThreadDemo05 { 
    
        public static void main(String[] args) throws InterruptedException {
            final SynchronousQueue<Integer> queue = new SynchronousQueue<Integer>();
    
            Thread product = new Thread(new Runnable() {
                @Override
                public void run() {
                    System.out.println("生产者线程start");
                    try {
                        queue.put(1);
                    } catch (InterruptedException e) {
                    }
                    System.out.println("生产者线程end");
                }
            });
    
            Thread customer = new Thread(new Runnable() {
                @Override
                public void run() {
                    System.out.println("消费者线程start");
                    try {
                        System.out.println("消费的数据: " + queue.take());
                    } catch (InterruptedException e) {
                    }
                    System.out.println("消费者线程end");
                }
            });
            // 先启动消费者，消费者会等待生产者生产数据才会去消费
            customer.start();
            Thread.sleep(1000);
            product.start();
        }
    }

## 使用BlockingQueue模拟生产者与消费者 ##

package top.nightliar.study.day04;
    
    import java.util.concurrent.BlockingDeque;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.LinkedBlockingDeque;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.atomic.AtomicInteger;
    
    /** * 使用BlockingQueue模拟生产者与消费者 * Created by Nightliar * 2018-09-17 16:26 */
    public class ThreadDemo06 { 
    
        public static void main(String[] args) throws InterruptedException {
            BlockingDeque<String> deque = new LinkedBlockingDeque<>(10);
            Product p1 = new Product(deque);
            Product p2 = new Product(deque);
            Consumer c1 = new Consumer(deque);
            c1.start();
            p1.start();
            p2.start();
            // 执行10s
            Thread.sleep(10 * 1000);
            p1.interrupt();
            p2.interrupt();
    
        }
    
    }
    
    /** * 生产者 */
    class Product extends Thread {
    
        private BlockingDeque<String> deque;
        private volatile boolean flag = true;
        private static AtomicInteger count = new AtomicInteger();
    
        public Product(BlockingDeque<String> deque) {
            this.deque = deque;
        }
    
        @Override
        public void run() {
            System.out.println("生成者开始工作");
            try {
                while (true){
                    String data = count.incrementAndGet()+"";
                    // 将数据存入队列中
                    boolean offer = deque.offer(data, 2, TimeUnit.SECONDS);
                    if (offer) {
                        System.out.println("生产者,存入" + data + "到队列中,成功.");
                    } else {
                        System.out.println("生产者,存入" + data + "到队列中,失败.");
                    }
                    Thread.sleep(1000);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                this.flag = false;
            }
        }
    }
    
    /** * 消费者 */
    class Consumer extends Thread {
    
        private BlockingDeque<String> deque;
        private volatile boolean flag = true;
        private static AtomicInteger count = new AtomicInteger();
    
        public Consumer(BlockingDeque<String> deque) {
            this.deque = deque;
        }
    
        @Override
        public void run() {
            System.out.println("消费线程启动...");
            try {
                while (flag) {
                    String data = deque.poll(2, TimeUnit.SECONDS);
                    if (data != null) {
                        System.out.println("消费者,拿到队列中的数据data:" + data);
                        Thread.sleep(1000);
                    } else {
                        System.out.println("消费者,超过2秒未获取到数据..");
                        flag = false;
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                System.out.println("消费者退出线程...");
            }
    
        }
    }