Java并发容器 & 并发队列

墨蓝 2024-03-16 18:03 74阅读 0赞

### 并发容器概览 ###

ConcurrentHashMap : 线程安全的HashMap

CopyOnWriteArrayList: 线程安全的List

BlockingQueue:这是一个接口，表示阻塞队列，非常适合用于作为数据共享的通道

ConcurrentLinkedQueue : 高效的非阻塞并发队列，使用链表实现。可以看做一个线程安全的LinkedList

ConcurrentSkipListMap : 是一个Map，使用跳表的数据结构进行快速查找

### 一、集合类的迭代历史 ###

#### 1、Vector 和 Hashtable ####

Vector 和 Hashtable 出现较早，但是并发性能差，主要是内部方法由sychronized修饰，目前使用较少。

##### （1）Vector #####

Vector 可以理解为一个线程安全的ArrayList；

使用演示如下，用法和 List 类似：

/**
     *      演示Vector
     */
    public class VectorDemo {
        
        public static void main(String[] args) {
        
            Vector<String> vector = new Vector<>();
            vector.add("李白");
            System.out.println(vector.get(0));
        }
    }

##### （2）Hashtable #####

Hashtable 可以理解为一个线程安全的HashMap；

用法演示如下，用法和 HashMap 类似：

/**
     *      演示Hashtable
     */
    public class HashtableDemo {
        
        public static void main(String[] args) {
        
            Hashtable<String, Object> hashtable = new Hashtable<>();
            hashtable.put("三月","杜甫");
            System.out.println(hashtable.get("三月"));
        }
    }

#### 2. ArrayList 和 HashMap 的改造 ####

虽然这两不是线程安全的，但是可以通过改造，使他们`变成线程安全的`：

*  **Collections.sychronizedList**(new ArrayList< E >())
 *  **Collections.sychronizedMap**(new HashMap< K,V >())

用法演示：

/**
     *      演示Collections.sychronizedList(new ArrayList<E>())
     */
    public class SycList {
        
        public static void main(String[] args) {
        
            List<Integer> list = Collections.synchronizedList(new ArrayList<>());
            list.add(5);
            System.out.println(list.get(0));
        }
    }

Collections.sychronizedList() 方法是如何实现线程安全的？

##### （1）Collections.sychronizedList() 方法源码 #####

![9b5c4a09384c1a6210b2974defbb84fa.png][]

![7a3a8f64c2970188e9d434da83227862.png][]

RandomAccess 是随机访问接口，表示这个集合能够跳着被随机访问。SynchronizedRandomAccessList 继承了 SynchronizedList类：

![5ece76e0022ef7c0aa634773efc25a5d.png][]

进入SynchronizedList 中，发现他里面方法的实现都是通过`sychronized同步代码块`实现

![3cc188c48176756ae3c24cd24363f221.png][]

**总结**：通过Collections.sychronizedList 等方式实现将ArrayList、HashMap转为线程安全的方法，并不比上面的Vector、Hashtable高明多少，只是从同步方法 演变成 同步代码块 的方式

#### 3、ConcurrentHashMap 和 CopyOnWriteArrayList ####

*  取代上面两种的并发安全实现
 *  `绝大多数`并发情况下，ConcurrentHashMap 和 CopyOnWriteArrayList的性能都很好
 *  CopyOnWriteArrayList 更适合`读多写少`的场景，如果经常有写操作， Collections.sychronizedList 比 CopyOnWriteArrayList 性能好
 *  无论是读还是写操作，ConcurrentHashMap 都比 Hashtable、Collections.synchronizedMap() 方法的性能好。

### 二、ConcurrentHashMap ###

#### 1. Map接口 ####

Map的接口和实现类：

![552e405d6f24121ba8305039a40ad716.png][]

#### 2. HashMap的线程不安全 ####

*  死循环造成CPU使用率100%
    
    在多个线程同时扩容的时候，会造成链表的死循环（你指向我，我指向你）
 *  同时put`碰撞`导致数据丢失
    
    如果两个数据计算出hash值后算出的对应集合的位置重复，那肯定会丢失1个数据
 *  同时put`扩容`导致数据丢失
    
    多个线程同时put，如果多个线程都需要执行扩容操作，那只会保留一个扩容后的数组

#### 3、HashMap 1.7 & 1.8 的结构特点 ####

##### （1）HashMap 1.7 #####

HashMap是在bucket中储存键对象和值对象，作为Map.Entry。执行put()方法传递键和值时，会先对键调用hashCode()方法，返回的hashCode用于找到bucket位置来储存Entry对象。如果它们的bucket位置相同，‘碰撞’会发生，这时会使用链表的方式一直往下链，这个Entry(包含有键值对的Map.Entry对象)会存储在链表中。俗称拉链法。

下面图中的一个个绿色方格就对应一个个 Entry。

![c5faecb277b796191e5ad46e0ac6928a.png][]

##### （2）HashMap 1.8 #####

1.8在原来的拉链法的基础上，增加了红黑树结构，即这个某个bucket 位置的 链表结构元素超过8个（并且HashMap总容量大于某个阈值）时，就会将链表转为红黑树结构。

![6a26d8ce985ac3739a5691e1482febaf.png][]

##### （3）**HashMap在并发场景下的特点** #####

*  非线程安全
 *  迭代时不允许修改内容
 *  只读的并发是安全的
 *  如果一定要使用HashMap在并发环境，那请用`Collections.sychronizedMap()`;

#### 4、ConcurrentHashMap 1.7 & 1.8 结构 ####

##### （1）ConcurrentHashMap 1.7 #####

由 segment 组成（类似于块结构），每个Segment里面对应一个类似HashMap的数组+链表的数据结构，默认有16个，这个默认值可以在初始化时自己设置，设置好之后就固定了，不允许扩容。每个segment 独立 上ReentrantLock锁，每个segment之间互不影响，提高了并发效率。

![ffbabae8ec139e87638ff8e818d0fbf5.png][]

##### （2）ConcurrentHashMap 1.8 #####

舍弃了 segment 结构，采用 node 结构，每一个节点就是一个node，通过 CAS + synchronized 实现并发安全，其整体结构与 HashMap 1.8 的结构类似，也是在遇到hash冲突时采用链表+红黑树来存储Entry，如下所示：

![d32c69d699958f0f009aea2a1d957838.png][]

**为什么链表的长度超过8要转为红黑树结构？**

*  红黑树的占用空间是链表的两倍，所以在hashCode冲突个数不多时，优先使用链表存储
 *  当链表过长时，为了查找的时候更快，使用红黑树，链表的查找时间复杂度O(n)，红黑树的查找时间复杂度O(logn)
 *  之所以选择8作为阈值，是因为hashCode冲突个数达到8的概率极小，hashmap源码的注释有说：依据泊松分布，达到8的概率小于千万分之一。但是为了保证在这种极端情况出现时，依然能有较高的查询效率，就转成红黑树结构。

#### 5. ConcurrentHashMap 1.8 的 get/put() 方法分析 ####

##### (1) put() 方法 #####

put() 方法中直接调用 putVal() 方法，下面看一下 putVal() 源码：

/** Implementation for put and putIfAbsent */
    final V putVal(K key, V value, boolean onlyIfAbsent) {
        
        //这里和hashmap不一样，hashmap允许一个元素的key为null，但是这里就不允许了
        // 如果这个槽点没有值
        if (key == null || value == null) throw new NullPointerException();
        //计算出自己的hash值
        int hash = spread(key.hashCode());
        int binCount = 0;
        //在这个for循环中，完成对 值的插入工作
        for (Node<K,V>[] tab = table;;) {
        
            Node<K,V> f; int n, i, fh;
            //判断tab是否没有被初始化，或长度等于0，他就进行初始化
            if (tab == null || (n = tab.length) == 0)
                tab = initTable();
            //如果他已经被初始化，且这个位置是空的，那就直接放入赋值
            else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
        
                //CAS操作
                if (casTabAt(tab, i, null,
                             new Node<K,V>(hash, key, value, null)))
                    break;                   // no lock when adding to empty bin
            }
            //判断当前的Hash值是不是MOVED
            //MOVED代表一种特殊的节点，一种转移节点，说明这个槽点正在扩容
            else if ((fh = f.hash) == MOVED)
                //帮助进行扩容和转移工作
                tab = helpTransfer(tab, f);
            else {
        
                //如果这个槽点有值
                V oldVal = null;
                //保证线程安全
                synchronized (f) {
        
                    if (tabAt(tab, i) == f) {
        
                        if (fh >= 0) {
        
                            binCount = 1;
                            //进行链表操作，根据当前hash值，找到这个hash该放的对应链表位置
                            for (Node<K,V> e = f;; ++binCount) {
        
                                K ek;
                                //判断当前存在不存在这个hash对应的key
                                if (e.hash == hash &&
                                    ((ek = e.key) == key ||
                                     (ek != null && key.equals(ek)))) {
        
                                    //把原来的oldVal赋成新值，并在后面返回oldVal
                                    oldVal = e.val;
                                    if (!onlyIfAbsent)
                                        e.val = value;
                                    break;
                                }
                                //到了这个就说明，这个是一个新的
                                Node<K,V> pred = e;
                                if ((e = e.next) == null) {
        
                                    //就在链表的最后创建一个新的节点
                                    //并把值初始化赋上
                                    pred.next = new Node<K,V>(hash, key,
                                                              value, null);
                                    break;
                                }
                            }
                        }
                        //判断他是否是一个红黑树
                        else if (f instanceof TreeBin) {
        
                            Node<K,V> p;
                            binCount = 2;
                            //putTreeVal()把值放到红黑树中
                            if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
                                                                  value)) != null) {
        
                                oldVal = p.val;
                                if (!onlyIfAbsent)
                                    p.val = value;
                            }
                        }
                    }
                }
                //走到这，代表已经完成添加操作了
                if (binCount != 0) {
        
                    //判断是否要将链表转成红黑树
                    //TREEIFY_THRESHOLD默认值为8，代表链表节点最少为8个才会尝试转成红黑树
                    if (binCount >= TREEIFY_THRESHOLD)
                        //treeifyBin()转换红黑树方法
                        //这里方法会要求数组的长度要大于默认的64；
                        //且链表节点长度要大于等于8个节点才会转红黑树
                        treeifyBin(tab, i);
                    if (oldVal != null)
                        //最后这里就是上面说的返回oldVal值
                        return oldVal;
                    break;
                }
            }
        }
        addCount(1L, binCount);
        return null;
    }

结合 put() 方法的源码，可以知道 put 的大致工作流程

*  判断key value不为空
 *  计算hash值
 *  根据对应位置节点的类型，来赋值：①直接存储；②helpTransfer（扩容时的转移）；③增长链表；④给红黑树增加节点检查，满足阈值就“红黑树”化
 *  返回oldValue

##### (1) get() 方法 #####

get() 方法源码：

public V get(Object key) {
        
        Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
        //获取到这个key的hash值，并用h来表示
        int h = spread(key.hashCode());
        //判断当前的这个数组长度不能等于null，且长度大于0，否则就直接返回null
        //代表这个map都没被建立初始化完毕
        if ((tab = table) != null && (n = tab.length) > 0 &&
            (e = tabAt(tab, (n - 1) & h)) != null) {
        
            //这个key对应的hash赋值并和这个槽点的hash值作比较
            if ((eh = e.hash) == h) {
        
                if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                    //就返回val，说明找到了
                    return e.val;
            }
            //如果为负数，说明他是一个红黑树节点或者转移节点
            else if (eh < 0)
                //那就用find()方法去找到这个红黑树对应的位置
                return (p = e.find(h, key)) != null ? p.val : null;
            //到了这里就说明，这个节点不是数组，又不是红黑树
            //那他这里就是一个链表数据结构
            //那就用while循环遍历这个链表
            while ((e = e.next) != null) {
        
                //找到对应的值
                if (e.hash == h &&
                    ((ek = e.key) == key || (ek != null && key.equals(ek))))
                    //返回
                    return e.val;
            }
        }
        return null;
    }

结合 get() 方法的源码，可以知道其大致工作流程：

*  计算hash值
 *  找到对应的位置，根据实际情况进行：
    
     *  直接取值
     *  红黑树里找值
     *  遍历链表取值
 *  返回找到的结果

#### 6. ConcurrentHashMap 1.7结构和1.8结构的对比 ####

*  数据结构不同
    
     *  1.7采用Segment块的结构，默认16个块，也就是16个线程数并发
     *  1.8采用和hashmap 类似的数组+链表+红黑树结构，不限制线程数
     *  并发度的改变：从 16个（因为1.7默认16个segment，不过可以自己设置）------>不限（1.8不限）
 *  Hash碰撞
    
     *  1.7采用拉链法，链表的形式往下
     *  1.8采用拉链法，链表形式往下，然后在根据链表长度，和Map总容量超过阈值时会转成红黑树
 *  保证并发安全
    
     *  1.7采用分段锁，通过Segment块保证线程安全，Segment块继承ReentrantLock
     *  1.8采用unsafe工具类的CAS操作 + sychronized修饰符
 *  查询复杂度
    
     *  1.7链表查询复杂度为：O(n)
     *  1.8当hash冲突的Entry个数超过8时，会转成红黑树，红黑树的时间复杂度为O(logn)

#### 7. ConcurrentHashMap 错误的使用案例 ####

##### （1）错误使用 #####

代码演示：

/**
     *      ConcurrentHashMap 组合操作并不保证线程安全
     */
    public class OptionsNotSafe implements Runnable {
        
        private static ConcurrentHashMap<String,Integer> scores = new ConcurrentHashMap<>();
    
        public static void main(String[] args) throws InterruptedException {
        
            scores.put("范仲淹",0);
            OptionsNotSafe r = new OptionsNotSafe();
            Thread thread1 = new Thread(r);
            Thread thread2 = new Thread(r);
            thread1.start();
            thread2.start();
            thread1.join();
            thread2.join();
    
            System.out.println(scores);
        }
    
        @Override
        public void run() {
        
            for (int i = 0; i < 1000; i++) {
        
                Integer score = scores.get("范仲淹");
                Integer newScore = score+1;
                scores.put("范仲淹",newScore);
            }
        }
    
    }

![f1abcf09a324c49fa9935d3eae8095b0.png][]

乍一看，以为 ConcurrentHashMap 并发不安全，其实 ConcurrentHashMap 线程安全体现在每一个单独的操作中，他只能保证一个get()或一个put()操作是具有线程安全的，但不能保证多个操作的组合是线程安全的。

##### （1）正确使用： #####

对于上面的修改`范仲淹`的分数操作，其实可以使用 replace() 方法实现线程安全，对 `run()`方法修改成如下：

@Override
        public void run() {
        
            for (int i = 0; i < 1000; i++) {
        
                    while (true) {
        
                        Integer score = scores.get("范仲淹");
                        Integer newScore = score + 1;
                        // 使用 ConcurrentHashMap 提供的方法replace，通过返回值的boolean来判断是否修改成功，不然就一直尝试修改,replace是基于 CAS的原理
                        boolean flag = scores.replace("范仲淹", score, newScore);
                        // System.out.println(flag);
                        if (flag){
        
                        	break;
                        }else {
        
                            System.out.println(score+":"+flag);
                        }
                    }
            }
        }

![42cd9dfed89ea72ef44f84f34d8c19cc.png][]

### 四、CopyOnWriteArrayList ###

#### 1. 诞生原因 ####

*  代替Vector和Collections.synchronizedList()，就和ConcurrentHashMap代替Collections.synchronizedMap()的原因一样
 *  Vector和SynchronizedList的锁的粒度太大，并发效率相对比较低，并且迭代（遍历）时无法编辑
 *  Copy-On-Write并发容器还包括CopyOnWriteArraySet，用来替代同步Set

#### 2、使用场景 ####

读多写少，且要求读操作很快，写操作可以慢一点的场景，比如：

*  网站的黑名单、白名单等，读操作比较多，写操作比较少，而且写的慢也没关系；
 *  监听器，不轻易增加或减少，大部分时间都是监听某事件并进行通知或报警

#### 3、读写规则 ####

CopyOnWriteArrayList 读取是完全不用加锁的，并且更厉害的是写入也不会阻塞读取操作。只有写入和写入之间需要进行同步等待，即只有写写互斥，读写不互斥

##### （1）ArrayList不支持在迭代过程中修改数据： #####

代码演示：

public class CopyOnWriteArrayListDemo1 {
        
        public static void main(String[] args) {
        
            ArrayList<String> list = new ArrayList<>();
            list.add("1");
            list.add("2");
            list.add("3");
            list.add("4");
            list.add("5");
    
            Iterator<String> iterator = list.iterator();
    
            while (iterator.hasNext()){
        
                System.out.println("list is" + list);
                String next = iterator.next();
                System.out.println(next);
    
                if (next.equals("3")){
        
                    list.remove("5");
                }
    
                if (next.equals("3")){
        
                    list.add("3 found");
                }
            }
        }
    }

![ce13c514efb0379afcf5f4ed5c4af3a9.png][]

在对list迭代过程中，修改list数据，会报错，说明ArrayList不允许在迭代时修改。

##### (2) CopyOnWriteArrayList支持在迭代过程中修改数据： #####

public class CopyOnWriteArrayListDemo2 {
        
        public static void main(String[] args) {
        
            CopyOnWriteArrayList<String> list = new CopyOnWriteArrayList<>();
            list.add("1");
            list.add("2");
            list.add("3");
            list.add("4");
            list.add("5");
    
            Iterator<String> iterator = list.iterator();
    
            while (iterator.hasNext()){
        
                System.out.println(list);
                String next = iterator.next();
                System.out.println(next);
    
                if (next.equals("3")){
        
                    list.remove("5");
                }
    
                if (next.equals("4")){
        
                    list.add("4 found");
                }
            }
        }
    
    }

![c034a64c9c68852d6441472882c65c82.png][]

CopyOnWriteArrayList 在迭代过程中，修改数据，并没有报错，说明CopyOnWriteArrayList允许在迭代时修改。

#### 4、实现原理 ####

##### (1) CopyOnWrite含义 #####

在写操作时，在原来的数据A基础上复制一份B，然后把B放在一块新的内存中，通过对B进行修改写入，之后再把指向原来A内存地址的指针，指向新复制出来的B所在的内存地址，而原来的内存由于没有指针绑定，便会被回收。

有三个特点：

*  创建新副本，读写分离
    
    对整个原来数据复制一份副本，把修改的内容写入新的副本中，最后再替换掉原来的数据
 *  不可变原理
    
    对于原来的数据，是不可变的，只是会被回收
 *  迭代遍历的时候
    
    在遍历的过程中修改了数据，不是直接修改原来的数据，而是修改副本数据，所以迭代器中的数据还是旧数据

代码演示：

/**
     * 描述：     对比两个迭代器
     */
    public class CopyOnWriteArrayListDemo2 {
        
    
        public static void main(String[] args) throws InterruptedException {
        
            CopyOnWriteArrayList<Integer> list = new CopyOnWriteArrayList<>(new Integer[]{
        1, 2, 3});
    
            System.out.println(list);
    
            Iterator<Integer> itr1 = list.iterator();
    
            list.remove(2);
            Thread.sleep(1000);
            System.out.println(list);
    
            Iterator<Integer> itr2 = list.iterator();
    
            itr1.forEachRemaining(System.out::println);
            itr2.forEachRemaining(System.out::println);
    
        }
    }

![539f14a070321693999eb2cbb6fb8466.png][]

CopyOnWriteArrayList在迭代时，迭代器中拿到什么数据，取决于他的`创建时间`，不取决于他执行`迭代循环的时间`

#### 5、缺点 ####

*  数据一致性问题：CopyOnWrite容器只能保证数据的最终一致性，不能保证数据的实时一致性。所以如果你希望写入的数据，马上能读到，请不要使用CopyOnWrite容器
 *  内存占用问题：因为CopyOnWrite的写是复制机制，所以在进行写操作的时候，内存里会同时驻扎两个对象的内存，内存消耗大

#### 6、源码分析 ####

##### （1）使用了 ReentrantLock() #####

![214268dc0de562fbfffa6b5262a00029.png][]

##### (3) add()方法 #####

![403c630c6f8227c570409eb4154f77b0.png][]

public boolean add(E e) {
        
        final ReentrantLock lock = this.lock;
        //上锁
        lock.lock();
        try {
        
            //拿到原来的数组
            Object[] elements = getArray();
            //获取原来的数组长度
            int len = elements.length;
            //复制一份原来的数组内容
            Object[] newElements = Arrays.copyOf(elements, len + 1);
            //把要添加的元素添加到新的数组中的最后一个位置
            newElements[len] = e;
            //将 array 存储的数组引用，改成这个新创建的数组引用
            setArray(newElements);
            //返回添加成功boolean
            return true;
        } finally {
        
            //解锁
            lock.unlock();
        }
    }

##### (3) get()方法 #####

get() 方法中没有加锁：

![7f824404c1d30a4678d12bfb2b08e632.png][]

### 五、并发队列 ###

#### 1、为什么要使用队列 ####

*  用队列可以存储数据、可以在线程间传递数据、交换数据等。
 *  队列如果是线程安全的，多线程操作时就可以直接使用队列传数据而不用考虑线程安全问题

#### 2. 队列关系图 ####

![dcab24b76b7784677b725d91c955765e.png][]

#### 4、什么是阻塞队列 ####

*  阻塞队列是具有阻塞功能的队列，所以它首先是一个队列，其次是具有阻塞功能
 *  通常，阻塞队列的一端是给生产者放数据用，另一端给消费者拿数据用。阻塞队列是线程安全的，所以生产者和消费者都可以是多线程的
 *  阻塞队列的分为有界和无界，有界是指队列存储容量有限，无界队列是指存储容量无限大，如 LinkedBlockingQueue 的存储容量为 Integer.MAX\_VALUE，约为2的31次方，可近似认为是无界队列。
 *  阻塞队列是线程池的重要组成部分，比如 Executors.newCachedThreadPool() 就是使用的 SynchonizedQueue，这是一个容量为0的队列。

![image-20210614200904916][]

下面看看几个常用的队列：

### 六、ArrayBlockingQueue阻塞有界队列 ###

#### 1. 特点 ####

*  一个对象数组+一把锁+两个条件（非空和非满两个条件）
 *  入队与出队都用同一把锁
 *  在只有入队高并发或出队高并发的情况下，因为操作数组，且不需要扩容，性能很高
 *  采用了数组，必须**指定大小**，即容量有限

#### 2、主要方法 ####

##### （1）put & take #####

put & take 是最有特色的两个带有阻塞功能的方法

*  take() 方法：获取并移除队列的头结点元素，一旦执行take的时候，如果队列里无数据，则阻塞，直到队列里有数据
 *  put() 方法： 插入元素。但是如果队列已满，那么就无法继续插入，并阻塞，直到队列里有了空闲空间

##### （2）add、remove & element #####

*  add：增加一个元素，满了报错
 *  remove：移除一个元素，空了报错
 *  element：获取头结点元素，并删除，空了报错

上面这三个方法，如果遇到队列满了或者空了，就会抛异常。

##### （3）offer 、poll & peek #####

*  offer：增加一个元素，满了则放入失败，并返回false
 *  poll：取出一个元素，并从队列中删除该元素，空了返回null
 *  peek：取出一个元素，不删除该元素，空了返回null

#### 3. 使用案例 ####

案例：有10个面试者，一共只有1个面试官，大厅里有3个位子休息，每个人面试时间1s。代码如下：

/**
      *  有10个面试者，一共只有1个面试官，大厅里有3个位子休息，每个人面试时间1s
     */
    public class ArrayBlockingQueueDemo {
        
        //主函数
        public static void main(String[] args) throws InterruptedException {
        
            ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<String>(3);//3个位置
            Interviewer r1 = new Interviewer(queue);
            Consumer r2 = new Consumer(queue);
    
            Thread interviewer = new Thread(r1);
            Thread consumer = new Thread(r2);
    
            interviewer.start();
            consumer.start();
        }
    
    }
    
    //面试官类
    class Interviewer implements Runnable {
        
        BlockingQueue<String> queue;
    
        public Interviewer(BlockingQueue<String> queue) {
        
            this.queue = queue;
        }
    
        @Override
        public void run() {
        
            System.out.println("10个候选人都来了");
            for (int i = 0; i < 10; i++) {
        
                String candidate = "Candidate：" + i;
                try {
        
                    queue.put(candidate);
                    System.out.println("安排好了！" + candidate);
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            }
            try {
        
                queue.put("stop");
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
        }
    }
    
    //面试者类
    class Consumer implements Runnable {
        
        BlockingQueue<String> queue;
    
        public Consumer(BlockingQueue<String> queue) {
        
            this.queue = queue;
        }
    
        @Override
        public void run() {
        
            try {
        
                Thread.sleep(1000);
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
            String msg;
            try {
        
                while (!"stop".equals(msg = queue.take())) {
        
                    System.out.println(msg + "，轮到了");
    
                }
                System.out.println("面试结束");
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
        }
    }

![e9a70129bfbd8f4aa47283371c96a4fe.png][]

队列容量为3，所以在放了三个元素之后就会阻塞，消费者每取出一个元素，生产者才能继续放进去一个元素，达到一种平衡。

#### 4. put() 方法源码分析 ####

![c151e55860930e6aaabc6ed0f34dabc9.png][]

### 七、LinkedBlockingQueue阻塞无界队列 ###

#### 1. 特点 ####

*  一个单向链表+两把锁+两个条件（非空和非满两个条件）
 *  两把锁，一把用于入队，一把用于出队，有效的避免了入队与出队时使用一把锁带来的竞争。
 *  在入队与出队都高并发的情况下，性能比ArrayBlockingQueue高很多
 *  采用了链表结构，最大容量为整数最大值（Integer.MAX\_VALUE），可看做容量无限

#### 2. 源码 ####

##### （1）他有take和put锁两把锁 #####

![c9df283c0851e904713e81e0d0684609.png][]

##### （2）put() 方法源码 #####

public void put(E e) throws InterruptedException {
        
        //判断传入的是否为空
        if (e == null) throw new NullPointerException();
        int c = -1;
        //将传入的内容进行包装
        Node<E> node = new Node<E>(e);
        //获取put锁
        final ReentrantLock putLock = this.putLock;
        //获取当前队列元素数
        final AtomicInteger count = this.count;
        //加锁
        putLock.lockInterruptibly();
        try {
        
            //如果当前队列数等于最大容量
            while (count.get() == capacity) {
        
                //阻塞
                notFull.await();
            }
            //放入队列
            enqueue(node);
            //c为自增后的旧值
            c = count.getAndIncrement();
            //c+1就是当前的元素数
            //代表当前元素数小于最大容量，还有空间
            if (c + 1 < capacity)
                //唤醒一个线程，让之前等待的进入工作状态
                notFull.signal();
        } finally {
        
            //解锁
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
    }

### 八、PriorityBlockingQueue阻塞优先级队列 ###

put操作时永远不会阻塞，因为 PriorityBlockingQueue 也是无界队列；但是take操作时，如果队列是空的，就有可能会阻塞

#### 1. 特点 ####

*  支持优先级：可以自己设置排序规则，使队列中的元素可以按照我们设置的顺序取出来，而不是先进先出
 *  使用 ReentrantLock
 *  无界队列，无界的原理是他可以在容量不够用时扩容
 *  是 PriorityQueue 的线程安全版本

### 九、SychronousQueue阻塞直接传递队列 ###

*  无空间容量，容量为0
 *  无锁
 *  只是在线程间直接交换，不持有元素，所以效率很高，是一个极好的`直接传递`的并发数据结构
 *  SynchronousQueue没有peek等函数，因为peek的含义是取出头结点，但是SynchronousQueue的容量是0，所以连头结点都没有，也就没有peek方法。同理，没有iterate相关方法
 *  SynchronousQueue 是线程池 Executors.newCachedThreadPool() 使用的阻塞队列

### 十、DelayQueue阻塞延时队列 ###

DelayQueue也是一个无界队列，所以在put() 时，不会被阻塞.

*  延迟队列，根据延迟时间排序
 *  队列元素需要实现Delayed接口，规定排序规则

### 十一、ConcurrentLinkedQueue非阻塞并发队列 ###

*  并发包中的非阻塞队列只有ConcurrentLinkedQueue这一种。
 *  顾名思义ConcurrentLinkedQueue是使用链表作为其数据结构的
 *  使用CAS非阻塞算法来实现线程安全(不具备阻塞功能)，适合用在对性能要求较高的并发场景，不常用

看源码的offer方法的CAS思想，内有p.casNext方法，使用了unsafe中的UNSAFE.compareAndSwapObject() 方法来实现CAS操作的原子性保证

![1b7958f2337d98b6708594fef92b5dbe.png][]

p.casNext() 使用了unsafe工具包来实现CAS操作的原子性保证, compareAndSwapObject 是一个native 方法

![7c3599ed1349ffd46b77d97d9e6184a6.png][]

### 十二、如何选择适合自己的队列？？ ###

可以从**边界**、**空间**、**吞吐量**三个方面来选择合适自己的队列，下面列举了几个队列的对比：

![97196f0599203582bff5a2a2991f8568.png][]

#### [点我扫码关注微信公众号][Link 1] ####

#### 文章来源：[Java并发容器 & 并发队列][Java_ _] ####

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

### 个人微信：CaiBaoDeCai ###

### 微信公众号名称：Java知者 ###

### 微信公众号 ID： JavaZhiZhe ###

### 谢谢关注！ ###

[9b5c4a09384c1a6210b2974defbb84fa.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/9f22e4526dea48cb9d2f232dc5cf8130.png
[7a3a8f64c2970188e9d434da83227862.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/3d79db7cb17f490a8d8279e53c8b4119.png
[5ece76e0022ef7c0aa634773efc25a5d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/90571c3f4cae4410b7fdd53dd15df4f3.png
[3cc188c48176756ae3c24cd24363f221.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/52e634f7e5d34556879fa8a5bc62f4a9.png
[552e405d6f24121ba8305039a40ad716.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/b31dfbbc50714bde8a567397b72ac082.png
[c5faecb277b796191e5ad46e0ac6928a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/18e70a668ea646a39313ec2b926637e0.png
[6a26d8ce985ac3739a5691e1482febaf.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/ef2e3d9e0e744b72a7002feccb1b597a.png
[ffbabae8ec139e87638ff8e818d0fbf5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/74a28355aea34665bbd6df181eeca48d.png
[d32c69d699958f0f009aea2a1d957838.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/3f2920a10ff24884989dd0318802e42f.png
[f1abcf09a324c49fa9935d3eae8095b0.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/7023ec210ef54bc89ac1e35dd6828651.png
[42cd9dfed89ea72ef44f84f34d8c19cc.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/22a454e1cecd41d4bee4dd3b2f7b0630.png
[ce13c514efb0379afcf5f4ed5c4af3a9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/3785b4194d964b16b150883441e47e3c.png
[c034a64c9c68852d6441472882c65c82.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/627dac7db2c14464a8920263e2ae09c3.png
[539f14a070321693999eb2cbb6fb8466.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/4a90cd63eaf74021bf78783904faa985.png
[214268dc0de562fbfffa6b5262a00029.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/704138c021c34609a6b4ae3ea6c24bb0.png
[403c630c6f8227c570409eb4154f77b0.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/877d033bfff34594a3b7e2597d0ebed9.png
[7f824404c1d30a4678d12bfb2b08e632.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/c43988553b2e4ed2ab432121eb20ddbd.png
[dcab24b76b7784677b725d91c955765e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/cc0c0df56e724c039dc13d1ee4b2c7fe.png
[image-20210614200904916]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/12ef8941721e41719402615c1bf7c7cc.png
[e9a70129bfbd8f4aa47283371c96a4fe.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/68683bb78a594db68ef67b1c6b8773df.png
[c151e55860930e6aaabc6ed0f34dabc9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/c52bed0110234511a683a79b24ca0c06.png
[c9df283c0851e904713e81e0d0684609.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/9d00844e039c4def9fc897944f5bf3d4.png
[1b7958f2337d98b6708594fef92b5dbe.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/b091205a5e88438b8a4dd6f8e46258e1.png
[7c3599ed1349ffd46b77d97d9e6184a6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/5f4a3d30c9cb4e7abcbcf479fe852465.png
[97196f0599203582bff5a2a2991f8568.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/95b6a1bf244640e69ada3d7b71fd2c15.png
[Link 1]: https://docs.qq.com/doc/DSVhXa2JraFpVVE9V
[Java_ _]: https://mp.weixin.qq.com/s?__biz=MzU3NDgyMTIxMw==&mid=2247484941&idx=1&sn=50708763a195a8d762676c20e2c11c7d&chksm=fd2dc03fca5a4929739ea79657acc15d8a7b67831c9b806ce66eb4b91123d50a6f84b33e6d4d#rd