阻塞队列水太深了，你把握不住的

小灰灰 2024-04-07 12:50 100阅读 0赞

随便在网上搜了一下，全是“深入剖析阻塞队列”、“架构师带你手写阻塞队列”、“阻塞队列竟然有8种”这一类的文章。恕我直言，关注点偏了，你越关注阻塞队列本身，越学不好阻塞队列。

提到阻塞队列，大家脑海中就会冒出：

*  BlockingQueue
 *  ArrayBlockingQueue
 *  LinkedBlockingQueue
 *  SynchronousQueue

但JDK阻塞队列本身是非常简单的，难的是阻塞队列内部的AQS**。**

如果你之前对阻塞队列一无所知又恰好想要学习，希望能耐心看完下面的内容。还是那句话，学习阻塞队列的重点不是阻塞队列本身...我自己都可以手写阻塞队列。

为了打破大家对阻塞队列“难”、“晦涩”、“神秘”的印象，我会从新的角度切入，重构大家对阻塞队列的认识。

主要内容：

*  什么是线程间通信
 *  实现线程间通信
    
     *  轮询
     *  等待唤醒机制：wait/notify
     *  等待唤醒机制：condition
 *  山寨版BlockingQueue
 *  JDK BlockingQueue简介
 *  展望AQS

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

### 什么是线程间通信 ###

定义：

> 针对同一个资源的操作有不同种类的线程。

说人话就是：共享资源+多线程，最典型的例子就是锁和生产者消费者（本文以生产者-消费者为例子讲解）。

以现实生活为例。消费者和生产者就像两个线程，原本做着各自的事情，厂家管自己生产，消费者管自己买，一般情况下彼此互不影响。![1704cacdf968469f8ae6223a59609e67.png][]

但当物资到达某个临界点时，就需要根据供需关系适当作出调整。

![27f5b88c5450404385c4229f3149fdcf.png][]

当厂家做了一大堆东西，产能过剩时，应该暂停生产，扩大宣传，让消费者过来消费

![3927fe15ecaa4229bb8276f58e58acdb.png][]

当消费者发现某个热销商品售罄，应该提醒厂家尽快生产

在上面的案例中，生产者和消费者是不同种类的线程，一个负责存入，另一个负责取出，且它们操作的是同一个资源。但最难的部分在于：

*  资源到达上限时，生产者等待，消费者消费
 *  资源达到下限时，生产者生产，消费者等待

你会发现，原本互不打扰的两个线程之间开始“沟通”了：

*  生产者：喂，我这边做的太多了，先休息会儿，你赶紧消费
 *  消费者：喂，货快没了，我休息会儿，你赶紧生产

这种线程间的相互调度，也就是线程间通信。

> 看到这，你心里暗暗想道：我擦，我只会new Thread().start()，怎么让A线程去喊B线程工作呢？

### 实现线程间通信 ###

还是以上面的生产者-消费者为例，有很多种方式可以实现线程间通信。

#### 轮询 ####

设计理念：生产者和消费者线程各自使用while循环，每隔片刻就去判断Queue的状态，队列为空时生产者才可插入数据，队列不为空时消费者才能取出数据，否则一律sleep等待。![a591fc5f29d84bba82470929a41e54f9.png][]

/**
     * 轮询版本
     */
    public class WhileQueue<T> {
        // 容器，用来装东西
        private final LinkedList<T> queue = new LinkedList<>();
    
        public void put(T resource) throws InterruptedException {
            while (queue.size() >= 1) {
                // 队列满了，不能再塞东西了，轮询等待消费者取出数据
                System.out.println("生产者：队列已满，无法插入...");
                TimeUnit.MILLISECONDS.sleep(1000);
            }
            System.out.println("生产者：插入" + resource + "!!!");
            queue.addFirst(resource);
        }
    
        public void take() throws InterruptedException {
            while (queue.size() <= 0) {
                // 队列空了，不能再取东西，轮询等待生产者插入数据
                System.out.println("消费者：队列为空，无法取出...");
                TimeUnit.MILLISECONDS.sleep(1000);
            }
            System.out.println("消费者：取出消息!!!");
            queue.removeLast();
            TimeUnit.MILLISECONDS.sleep(5000);
        }
    
    }

测试

public class Test {
        public static void main(String[] args) {
            // 队列
            WhileQueue<String> queue = new WhileQueue<>();
    
            // 生产者
            new Thread(new Runnable() {
                @Override
                public void run() {
                    for (int i = 0; i < 100; i++) {
                        try {
                            queue.put("消息" + i);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }).start();
    
            // 消费者
            new Thread(new Runnable() {
                @Override
                public void run() {
                    for (int i = 0; i < 100; i++) {
                        try {
                            queue.take();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }).start();
        }
    }

![c78652f7b21e46b1b5d88c30e1b04a45.jpeg][]由于设定了队列最多只能存1个消息，所以只有当队列为空时，生产者才能插入数据。这是最简单的线程间通信：

> 多个线程不断轮询共享资源，通过共享资源的状态判断自己下一步该做什么。

看到这，你发现自己被骗了：哦，原来要实现线程间通信，并非真的需要A线程直接去叫B线程干什么，只要能按实际情况完成线程切换即可！

但上面的实现方式存在一些缺点：

*  轮询的方式太耗费CPU资源，如果线程过多，比如几百上千个线程同时在那轮询，会给CPU带来较大负担
 *  无法保证原子性（代码里没有演示，但理论上确实如此，如果生产者的操作非原子性，消费者极可能获取到脏数据）

#### 等待唤醒机制：wait/notify ####

相对而言，等待唤醒机制则要优雅得多，底层通过维护线程队列的方式，避免了过多线程同时自旋造成的CPU资源浪费，颇有点“用空间换时间”的味道。当一个生产者线程无法插入数据时，就让它在队列里休眠（阻塞），此时生产者线程会释放CPU资源，等到消费者抢到CPU执行权并取出数据后，再由消费者唤醒生产者继续生产。

举个例子，原本生产者和消费者都要时不时去店里看一下：

*  生产者：货卖完了没有，卖完了我要继续生产（每分钟来店里看一下）
 *  消费者：补货了没，补货了我就可以买了（每分钟来店里看一下）

而现在，生产者去店里看了下，发现还有货，就管自己去后厨睡觉了，等店里货都卖完了，自然会有消费者过来喊他补货，不需要付出额外的精力在店里盯着。

Java有多种方式可以实现等待唤醒机制，最经典的就是wait和notify。

/**
     * wait/notify版本
     */
    public class WaitNotifyQueue<T> {
        // 容器，用来装东西
        private final LinkedList<T> queue = new LinkedList<>();
    
        public synchronized void put(T resource) throws InterruptedException {
            while (queue.size() >= 1) {
                // 队列满了，不能再塞东西了，轮询等待消费者取出数据
                System.out.println("生产者：队列已满，无法插入...");
                this.wait();
            }
            System.out.println("生产者：插入" + resource + "!!!");
            queue.addFirst(resource);
            this.notify();
        }
    
        public synchronized void take() throws InterruptedException {
            while (queue.size() <= 0) {
                // 队列空了，不能再取东西，轮询等待生产者插入数据
                System.out.println("消费者：队列为空，无法取出...");
                this.wait();
            }
            System.out.println("消费者：取出消息!!!");
            queue.removeLast();
            this.notify();
        }
    }

![23af506660384b8a95404ca055d83545.jpeg][]

对比WhileQueue做了哪些改进：

*  用synchronized保证原子性
 *  wait和notify实现等待唤醒![de9579ce0c6f4bb38f3e208bd00a46cd.jpeg][]

但一般推荐使用notifyAll（为什么？）。我们给测试程序再加一个生产者线程就知道了：

![30a3480f555e41cdbbca1ab2c071901f.jpeg][]

开始不久后，整个程序所有线程都阻塞了

原因是：在synchronized机制下，所有等待的线程都在同一个队列里，而notify又恰巧是**随机**唤醒线程（也就是说，有可能生产者唤醒生产者）。

![ddf0158196e94315b5cafaf2751d616f.jpeg][]

最终结果是：所有线程都睡觉了...表现在程序上，就是卡住了。

解决办法是改用notifyAll，把所有线程都唤醒，然后大家一起参与执行权的竞争。你是否有疑问：如果和上面一样，生产者1还是唤醒生产者2呢？

其实这个假设不成立...使用notifyAll以后就不再是随机唤醒某一个线程了，而是唤醒所有线程并重新抢夺执行权。也就是说，每一个线程在进入阻塞之前，都会叫醒其他所有线程！

#### 等待唤醒机制：condition ####

wait/notify版本的缺点是随机唤醒容易出现“己方唤醒己方，最终导致全部线程阻塞”的乌龙事件，虽然wait/notifyAll能解决这个问题，但唤醒全部线程又不够精确，会造成无谓的线程竞争（实际只需要唤醒敌方线程即可）。

作为改进版，可以使用ReentrantLock的Condition替代synchronized的wait/notify：

/**
     * Condition版本
     */
    public class ConditionQueue<T> {
        // 容器，用来装东西
        private final LinkedList<T> queue = new LinkedList<>();
    
        // 显式锁（相对地，synchronized锁被称为隐式锁）
        private final ReentrantLock lock = new ReentrantLock();
        private final Condition producerCondition = lock.newCondition();
        private final Condition consumerCondition = lock.newCondition();
    
        public void put(T resource) throws InterruptedException {
            lock.lock();
            try {
                while (queue.size() >= 1) {
                    // 队列满了，不能再塞东西了，轮询等待消费者取出数据
                    System.out.println("生产者：队列已满，无法插入...");
                    // 生产者阻塞
                    producerCondition.await();
                }
                System.out.println("生产者：插入" + resource + "!!!");
                queue.addFirst(resource);
                // 生产完毕，唤醒消费者
                consumerCondition.signal();
            } finally {
                lock.unlock();
            }
        }
    
        public void take() throws InterruptedException {
            lock.lock();
            try {
                while (queue.size() <= 0) {
                    // 队列空了，不能再取东西，轮询等待生产者插入数据
                    System.out.println("消费者：队列为空，无法取出...");
                    // 消费者阻塞
                    consumerCondition.await();
                }
                System.out.println("消费者：取出消息!!!");
                queue.removeLast();
                // 消费完毕，唤醒生产者
                producerCondition.signal();
            } finally {
                lock.unlock();
            }
        }
    }

如何理解Condition呢？你可以认为lock.newCondition()创建了一个队列，调用producerCondition.await()会把生产者线程放入**生产者的等待队列**中，当消费者调用producerCondition.signal()时会唤醒从**生产者的等待队列**中唤醒一个生产者线程出来工作。![88869d0c93cf439ebc32e28e824c6b21.png][]

也就是说，ReentrantLock的Condition通过拆分线程等待队列，让线程的等待唤醒更加精确了，想唤醒哪一方就唤醒哪一方。

### 山寨版BlockingQueue ###

至此，大家应该对线程间通信有了大致了解。如果你仔细观察，会发现上面其实都采用了阻塞队列实现。我们都是先构造一个Queue，然后生产者和消费者直接操作Queue，至于是否阻塞，由Queue内部判断。这样封装的好处是，将生产者和消费者解耦的同时，不暴露过多细节，使用起来更简单。

大家应该都听过JDK的阻塞队列吧？基于上面的案例，我们改进一下，抽取出一个自定义的阻塞队列（使用wait/nofityAll实现）：

public class BlockingQueue<T> {
    
        // 模拟队列
        private final LinkedList<T> queue = new LinkedList<>();
    
        private int MAX_SIZE = 1;
        private int remainCount = 0;
    
        public BlockingQueue(int capacity) {
            if (capacity <= 0) {
                throw new IllegalArgumentException("size最小为1");
            }
            this.MAX_SIZE = capacity;
        }
    
        public synchronized void put(T resource) throws InterruptedException {
            while (queue.size() >= MAX_SIZE) {
                // 队列满了，不能再塞东西了，阻塞生产者
                System.out.println("插入阻塞...");
                this.wait();
            }
            queue.addFirst(resource);
            remainCount++;
            printMsg(resource, "被插入");
            this.notifyAll();
        }
    
        public synchronized T take() throws InterruptedException {
            while (queue.size() <= 0) {
                // 队列空了，不能再取东西了，阻塞消费者
                System.out.println("取出阻塞...");
                this.wait();
            }
            T resource = queue.removeLast();
            remainCount--;
            printMsg(resource, "被取出");
            this.notifyAll();
            return resource;
        }
    
        private void printMsg(T resource, String operation) throws InterruptedException {
            System.out.println(resource + operation);
            System.out.println("队列容量：" + remainCount);
        }
    }

![6f3d6894c3c24ea787b1ca8dd1aee813.jpeg][]

### JDK BlockingQueue简介 ###

虽然很多人开口闭口“阻塞队列”，但“阻塞队列”在他脑中只是个很模糊的概念。连“阻塞队列”的来龙去脉都不甚清楚，又怎么能说了解呢？

实际上，和List、Set一样，“阻塞队列”也有自己的一脉。在JDK的util包下有一个Queue接口：

![54baff90168743a18b3d0963b3781e68.jpeg][]

如果你继续往下扒，就会发现**Queue和List其实很像，也是集合的一个分支罢了：**

![510485d02664406f9f13dbb281a95129.jpeg][]

为什么很多人会觉得阻塞队列（比如ArrayBlockingQueue）高大上，听起来比ArrayList牛逼呢？主要在于“阻塞”二字！因为大家不了解阻塞，自己也不知道怎么实现阻塞，所以会觉得阻塞队列很神秘，很牛逼。但仔细观察上面的继承关系你会发现，如果ArrayBlockingQueue没有实现BlockingQueue接口，那么它本应该是个普普通通的队列，而不是阻塞队列，也就没有那么惊艳了。

那么BlockingQueue做了啥呢？其实啥也没做，毕竟BlockingQueue只是个接口，而接口只能定义方法...就好比一栋摩天大厦建成了，楼顶有个空中泳池，你觉得很牛逼。那么，你觉得是当初说“我要楼顶有个大花园”的老板牛逼还是把这个方案实现的设计师牛逼呢？

扯远了，其实BlockingQueue继承Queue接口后，就定义了几个方法：![6473c53218854dd0a75c3004d94d2929.png][]

BlockingQueue金口一开，后面的小弟只能满足，所以几个阻塞队列的实现类都有上面的几个方法。![c3844207e4a445ecb5ffbc1b6b79160b.jpeg][]

那么阻塞队列的“阻塞”是怎么实现的呢？以ArrayBlockingQueue为例，通过上面的继承关系分析，Queue和BlockingQueue是接口，里面只有方法定义没有具体实现，有可能实现“阻塞”功能的要么在AbstractQueue，要么就是ArrayBlockingQueue自身。我们查看AbstractQueue发现这家伙几乎啥都没写...![679f33d126f74e13830f14440fa1825c.jpeg][]

也就是说，当初老板发话“我希望这个队列能阻塞”，经理微笑着满口答应，结果转手就交给3个小弟自己整了。好在3个小弟争气，还真给他们搞出来了...

常用的3个阻塞队列：

*  ArrayBlockingQueue
 *  LinkedBlockingQueue
 *  SynchronousQueue

还是以ArrayBlockingQueue为例，它是怎么实现阻塞的呢？

![e9f01b99d7c14a61a14f21650f44aafa.jpeg][]

![ff9a3939741d42d4947cf75aab716e23.jpeg][]

好家伙...竟然用了ReentrantLock，这和我们上面案例中写的ConditionQueue好像啊！

![1a9a050051b54e1fbf8d0317408d0b93.jpeg][]

但是ArrayBlockingQueue只有notFull.await()，没看到signal()，不合理。仔细找找，唯一的可能是ArrayBlockingQueue把signal()藏在enqueue(e)方法里了：![63873d3a5afa400aa134e9c9f41f98ab.jpeg][]

其他两个阻塞队列LinkedBlockingQueue和SynchronousQueue同理，也是用ReentrantLock实现阻塞的。

### 展望AQS ###

看到这里，相信阻塞队列在大家心中已经不再那么神圣了，有什么了不起啊，我们自己也能写啊，还用了好几种方式实现呢！但是扪心自问，阻塞队列总共也就：

*  阻塞
 *  队列

而我们所谓的手写阻塞队列，其实是这样的：队列直接用了LinkedList，阻塞也是借用wait/notify和ReentrantLock实现的。也就是说，我们其实只是做了组装工作，拿现成的队列+阻塞功能拼出了一个阻塞队列。

世间路千万条，总有人不走寻常路。按理说现成的List+wait/notifyAll已经可以造出阻塞队列了，但就是有大佬不满足。

Doug Lea老爷子震惊的说：What?! Why you don't say earlly ya! I have already finished the AQS le...

是的，又是这个男人，他整出了一个AQS，再把AQS塞到ReentrantLock中，最后用ReentrantLock+数组、ReentrantLock+链表、ReentrantLock+Transfer搞出了ArrayBlockingQueue、LinkedBlockingQueue和SynchronousQueue...阻塞队列只能算顺便的，他的初衷其实是利用AQS统一并简化锁的实现，屏蔽同步状态管理、阻塞线程的排队和通知、唤醒机制等，让后续的二次开发更简便。

换句话说：

如果你纠结于阻塞队列怎么实现，那你的格局就太小了...JDK的阻塞队列依赖于ReentrantLock，而ReentrantLock只是对AQS的浅封装，真正需要我们花功夫学习的其实有且只有AQS。

[1704cacdf968469f8ae6223a59609e67.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/72e07b4c4a584599a7e34a152b4dee46.png
[27f5b88c5450404385c4229f3149fdcf.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/f5cd12d9ae8c40acb4660a659d7da067.png
[3927fe15ecaa4229bb8276f58e58acdb.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/ad4eab532b8b4be68953999dc8472a68.png
[a591fc5f29d84bba82470929a41e54f9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/b1bfc503419b42f38d4abcb2553d3eb6.png
[c78652f7b21e46b1b5d88c30e1b04a45.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/8d19c036087b49b3bddbde34fc10537d.jpeg
[23af506660384b8a95404ca055d83545.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/bcd27d4a04c54103969827307093ba47.jpeg
[de9579ce0c6f4bb38f3e208bd00a46cd.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/aee5dfc7b4f24a78add67cc0fa88fe08.jpeg
[30a3480f555e41cdbbca1ab2c071901f.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/4364b5067ff44c898c41b2a0a38a2310.jpeg
[ddf0158196e94315b5cafaf2751d616f.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/dc2c77f9890e43d79a91d755ed28e432.jpeg
[88869d0c93cf439ebc32e28e824c6b21.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/0612457fd25a4143adcad54e2e282995.png
[6f3d6894c3c24ea787b1ca8dd1aee813.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/a2360f158314409dad34fae88b22af32.jpeg
[54baff90168743a18b3d0963b3781e68.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/f41c295717844e6193b78e69007de0ab.jpeg
[510485d02664406f9f13dbb281a95129.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/099f2212202043b8ada65e61831bef16.jpeg
[6473c53218854dd0a75c3004d94d2929.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/53f39e1676d14a2db8876ad1ca89db0c.png
[c3844207e4a445ecb5ffbc1b6b79160b.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/377eddd9cff7479582136d158c5b0c36.jpeg
[679f33d126f74e13830f14440fa1825c.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/6ddcefd923324add9c9322524e87328d.jpeg
[e9f01b99d7c14a61a14f21650f44aafa.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/9e69a3bd53e54a44b3a1e4d9c14756c6.jpeg
[ff9a3939741d42d4947cf75aab716e23.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/19ca10b5e67f4ce38188f178f88ba406.jpeg
[1a9a050051b54e1fbf8d0317408d0b93.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/57781fe604154cd98546c400f22e2ec3.jpeg
[63873d3a5afa400aa134e9c9f41f98ab.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/56f84fafe9ff4888baecc88a157f522b.jpeg