Java多线程基础-9：代码案例之阻塞队列（一）+ https://developer.aliyun.com/article/1520531?spm=a2c6h.13148508.setting.14.75194f0eoT1Sgd
三、代码实现 BlockingQueue
此处介绍如何采用数组实现阻塞队列（不带泛型）。
阻塞队列就是“带有阻塞特性的队列”，实现一个阻塞队列主要分为3步：
实现一个普通队列。
加上线程安全。
加上阻塞功能。
1、实现一个普通队列
结合数据结构的知识，可以通过数组构造出一个循环队列：
class MyBlockingQueue { // 创建数组空间 private int[] elem = new int[1000]; private int front = 0; private int back = 0; private int size = 0; public void put(int value) { // 数组满 if(elem.length == size) { return; } elem[back] = value; back++; // 判断 if(back == elem.length) { back = 0; } size++; } public Integer take() { // 数组空 if(size == 0) { return null; } int value = elem[front]; front++; // 判断是否到数组末尾 if(front == elem.length) { front = 0; } size--; return value; } }
讲解一下这里为什么要加：
if(back == elem.length) { back = 0; } 或 if(front == elem.length) { front = 0; }
在tail或head到达数组的末端时，需要让这两个指针回到数组的开头，以此“循环”。很多同学可能接触过取余数的写法：back += (back+1)%数组长度。 这样的方式也是可行的，但是效率不高。
在写代码时，一般会追求开发效率和执行效率。开发效率是针对程序员开发程序而言的，而执行效率是针对代码运行而言的。写成上面取余数的方式，既没有提高程序的开发效率（这样写代码的可读性并不高），也没有保证执行效率，因此不是一种好的写法，尽管它也能达成目的。
而写成if的方式能够提高这两个效率。在执行效率方面，表面上看if语句有两行，但实际上，每次只有指针到达数组末尾了，才会进入if判断；而到达数组末尾的次数是远远小于操作总次数的。比如上述的elem数组，总长度为1000，只有执行了1000次操作后，才会进入一次if判断让back回到开头，而剩下的999次操作都不会进入if判断。
2、加上线程安全
即加锁，用synchronized即可。直接把synchronized加在成员方法take()和put()上，锁对象是this，即MyBlockingQueue类的实例。
如下面代码中，对象myBlockingQueue被创建出来，那么当它调用take和put时，该对象的实例就是锁对象，即this。找到谁是锁对象，对后面实现阻塞功能有重要影响。
public class Test { public static void main(String[] args) { MyBlockingQueue myBlockingQueue = new MyBlockingQueue(); } } class MyBlockingQueue { // 创建数组空间 private int[] elem = new int[10]; private int front = 0; private int back = 0; private int size = 0; synchronized public void put(int value) { // 数组满 if(elem.length == size) { return; } elem[back] = value; back++; // 判断 if(back == elem.length) { back = 0; } size++; } synchronized public Integer take() { // 数组空 if(size == 0) { return null; } int value = elem[front]; front++; // 判断是否到数组末尾 if(front == elem.length) { front = 0; } size--; return value; } }
3、加上阻塞功能
根据阻塞队列的特性：队列满时继续put()会阻塞，直到有线程take()；队列空时继续从队列中take()也会阻塞。，直到有线程put()。通过wait()和notify()配合来实现。调用 wait() 和 notify() 方法的对象必须是共享同一个锁对象的线程，否则会抛出 IllegalMonitorStateException 异常。由于synchronized直接加在了成员方法上，因此锁对象就是this，那么调用wait()和notify(）的对象也是this。
public class Test { public static void main(String[] args) { MyBlockingQueue myBlockingQueue = new MyBlockingQueue(); } } class MyBlockingQueue { // 创建数组空间 private int[] elem = new int[10]; private int front = 0; private int back = 0; private int size = 0; synchronized public void put(int value) throws InterruptedException { // 数组满 if(elem.length == size) { this.wait(); //队列满再入队列->阻塞 } elem[back] = value; back++; // 判断 if(back == elem.length) { back = 0; } size++; this.notify(); // 唤醒因队列空而阻塞的情况 } synchronized public Integer take() throws InterruptedException { // 数组空 if(size == 0) { this.wait(); // 队列空而出队列->阻塞 } int value = elem[front]; front++; // 判断是否到数组末尾 if(front == elem.length) { front = 0; } size--; this.notify(); // 唤醒因队列满而阻塞的情况 return value; } }
注意，上面两个队列不可能同时阻塞，也就是说，一个队列不可能既是空，又是满。
4、用while替代if进行条件判断
还有一点要强调，其实Java官方并不是非常推荐这样使用wait：
查看Java官方的说明，可以发现：
也就是说，wait()方法可能会被其它其它方法中断，如interrupt方法。此时，wait的等待条件其实还没成熟，就被提前唤醒了，这样代码的执行可能就不符合预期了。
明明队列非空的条件还没满足，但wait被唤醒后就继续往下走了，没有进行条件判断
解决这个问题很简单：更稳妥的方法，是在wait被唤醒之后，再进行一次条件判断。wait之前，发现继续执行的条件不满足，开始wait，然后等到wait被唤醒之后，再确认一下条件是不是满足。如果不满足，还可以继续wait。
wait应总是在while中使用。如这个例子：https://blog.csdn.net/wyd_333/article/details/130575605
当中，涉及的“虚假唤醒”问题就与wait是否在while中使用直接相关。
5、完整代码：MyBlockingQueue
class MyBlockingQueue { // 创建数组空间 private int[] elem = new int[10]; private int front = 0; private int back = 0; private int size = 0; synchronized public void put(int value) throws InterruptedException { // 数组满 // 用while判断是否满足条件 while(elem.length == size) { this.wait(); } elem[back] = value; back++; // 判断 if(back == elem.length) { back = 0; } size++; this.notify(); } synchronized public Integer take() throws InterruptedException { // 数组空 // 用while判断是否满足条件 while(size == 0) { this.wait(); } int value = elem[front]; front++; // 判断是否到数组末尾 if(front == elem.length) { front = 0; } size--; this.notify(); return value; } }
6、实现生产者-消费者模型
public class Test { public static void main(String[] args) { MyBlockingQueue myBlockingQueue = new MyBlockingQueue(); // 生产者 put Thread t1 = new Thread(() -> { try { int value = 0; while(true) { System.out.println("生产：" + value); myBlockingQueue.put(value); Thread.sleep(1000); value++; } } catch (InterruptedException e) { e.printStackTrace(); } }); t1.start(); // 消费者 take Thread t2 = new Thread(() -> { while(true) { try { int value = myBlockingQueue.take(); System.out.println("消费：" + value); } catch (InterruptedException e) { e.printStackTrace(); } } }); t2.start(); } }
在生产者中添加Thread.sleep()，此时是生产得快，消费得慢。运行后可以直观感受到，元素是生产一个、消费一个的。每生产完一个，就立即消费掉，等待再次生产。
当然，也可以在消费者中添加Thread.sleep()。此时是生产得慢，而消费得快。在消费者sleep的时候，生产者生产迅速，很可能会一口气生成完整个队列中的所有元素，令队列满而生产者进入阻塞。等待消费者结束sleep，消费1个元素，此时生产者才会再次生产。
Thread t2 = new Thread(() -> { while(true) { try { // 令消费者sleep Thread.sleep(3000); int value = myBlockingQueue.take(); System.out.println("消费：" + value); } catch (InterruptedException e) { e.printStackTrace(); } } });
上面之所以一口气只生产到10，是因为在MyBlockingQueue中，把队列容量设定为了10.
// 创建数组空间 private int[] elem = new int[10];
四、**补充：SynchronousQueue
在 Java 中，SynchronousQueue 是一个实现了 BlockingQueue 接口的类。与普通的 BlockingQueue 不同的是，SynchronousQueue没有容量capacity，具有一些特殊的行为。
1、SynchronousQueue的使用
***有一个疑问：SynchronousQueue没有capacity，那么它怎么存放put进来的元素呢？又从哪里take元素呢？
确实，SynchronousQueue 并不是像其他阻塞队列那样有固定的容量，它的实现实际上是基于一种传输机制，即：一个线程试图将元素添加到队列中时，必须等待另一个线程从队列中取走该元素，反之亦然。换句话说，SynchronousQueue 可以看作是一个零容量的阻塞队列，其中元素的存储和移除，是通过线程之间的相互配合完成的。
具体来说：当一个线程调用 put 方法时，它会将要添加的元素暂时保存在本地变量中，然后阻塞自己，等待另一个线程调用 take 方法取走该元素。当另一个线程调用 take 方法时，它也会阻塞自己，等待另一个线程调用 put 方法并将要传递的元素交给它。在这个过程中，元素的实际存储并不需要使用队列，因为队列并没有容量，而是仅用于实现线程之间的协作机制。
因此，SynchronousQueue 可以用于在两个线程之间传递元素：一个线程生产元素并调用 put 方法，另一个线程消费元素并调用 take 方法。这种交换操作在两个线程之间同步进行（因此也称为同步队列）。
下面展示了如何在两个线程之间使用 SynchronousQueue 进行元素交换：
import java.util.concurrent.SynchronousQueue; public class SynchronousQueueExample { public static void main(String[] args) { final SynchronousQueue<Integer> queue = new SynchronousQueue<Integer>(); Thread producer = new Thread(new Runnable() { public void run() { try { int value = 42; System.out.println("Producer is putting: " + value); queue.put(value); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } }); Thread consumer = new Thread(new Runnable() { public void run() { try { int value = queue.take(); System.out.println("Consumer is taking: " + value); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } }); producer.start(); consumer.start(); } }
在上面程序中，我们创建了一个 SynchronousQueue 实例，然后创建了一个生产者线程和一个消费者线程。生产者线程调用 put 方法将值 42 添加到队列中，然后阻塞等待消费者线程取走这个元素。消费者线程调用 take 方法从队列中取出一个元素，并在控制台输出它。由于这个队列是同步的，因此生产者线程和消费者线程将会同步进行，直到元素被成功交换。
2、注意
（1）对于 SynchronousQueue 来说，put 和 take 操作必须是交替执行的
如果一个线程调用了 put 方法并被阻塞，那么它只有等待另一个线程调用 take 方法才能继续执行。反之，如果一个线程调用了 take 方法并被阻塞，那么它只有等待另一个线程调用 put 方法才能继续执行。
这种交替执行的特性使得 SynchronousQueue 在某些情况下非常有用。
例如，当一个线程需要向另一个线程传递数据时，可以使用 SynchronousQueue 来保证线程之间的同步和可靠性。在这种情况下，一个线程负责生产数据并将其添加到队列中，而另一个线程负责消费数据并从队列中取出它。由于队列中始终只有一个元素，因此这种操作是非常高效和可靠的。
但要注意，如果一个线程调用了 put 方法但是没有其他线程调用 take 方法取走元素，那么这个线程就会一直被阻塞下去。同样地，如果一个线程调用了 take 方法但是没有其他线程调用 put 方法添加元素，那么这个线程也会一直被阻塞下去。
（2）SynchronousQueue 中的元素不可重复使用。
一旦一个线程调用 put 方法将元素添加到队列中，并被另一个线程调用 take 方法取走，该元素就会从队列中消失，不再可用于后续的操作了。这也是 SynchronousQueue 的一个特殊之处，使得它非常适用于需要高效、安全地将数据传递给另一个线程的场景。
因此，在使用 SynchronousQueue 时，需要特别注意线程的调度和同步问题，以免造成死锁或其他问题。
