8.1 是什么

阻塞队列，顾名思义，首先它是一个队列，而一个阻塞队列在数据结构中所起的作用大致如下图所示：
image.png

  • 线程 1 往阻塞队列中添加元素，而线程 2 从阻塞队列中移除元素，当阻塞队列是空时，从队列中获取元素的操作将会被阻塞。
  • 当阻塞队列是满时，往队列里添加元素的操作将会被阻塞。
  • 试图从空的阻塞队列中获取元素的线程将会被阻塞，直到其他的线程往空的队列插入新的元素。同样试图往已满的阻塞队列中添加新元素的线程同样也会被阻塞，直到其他的线程从列中移除一个或者多个元素或者完全清空队列后使队列重新变得空闲起来并后续新增

    8.2 有什么好处

    在多线程领域：所谓阻塞，在某些情况下会挂起线程(即阻塞)，一旦条件满足，被挂起的线程又会自动被唤醒
    BlockingQueue 好处是我们不需要关心什么时候需要阻塞线程，什么时候需要唤醒线程，因为这一切 BlockingQueue 都给你一手包办了
    在 concurrent 包发布以前，在多线程环境下，我们每个程序员都必须去自己控制这些细节，尤其还要兼顾效率和线程安全，而这会给我们的程序带来不小的复杂度。

    8.3 种类分析

    ArrayBlockingQueue：由数组结构组成的有界阴塞队列。
    LinkedBlockingQueue：由链表结构组成的有界(但大小默认值为 Integer.MAX_VALUE)阻塞队列。
    PriorityBlockingQueue：支持优先级排序的无界阻塞队列。
    DelayQueue：使用优先级队列实现的延迟无界阻塞队列。
    SynchronousQueue：不存储元素的阻塞队列，也即单个元素的队列。
    LinkedTransferQueue：由链表结构组成的无界阻塞队列。
    LinkedBlockingDeque：由链表结构组成的双向阻塞队列。

    8.4 核心方法

    | 方法类型 | 抛出异常 | 特殊值 | 阻塞 | 超时 | | —- | —- | —- | —- | —- | | 插入 | add(e) | offer(e) | put(e) | offer(e,time,unit) | | 移除 | remove() | poll() | take() | poll(time,unit) | | 检查 | element() | peek() | 不可用 | 不可用 |
抛出异常 当阻塞队列满时，再往队列里 add 插入元素会抛 IllegalStateException: Queue full
当阻塞队列空时，再往队列里 remove 移除元素会抛 NoSuchElementException
特殊值 插入方法，成功 ture 失败 false
移除方法，成功返回出队列的元素，队列里面没有就返回 null
一直阻塞 当阻塞队列满时，生产者线程继续往队列里 put 元素，队列会一直阻塞生产线程直到 put 数据 or 响应中断退出。
当阻塞队列空时，消费者线程试图从队列里 take 元素，队列会一直阻塞消费者线程直到队列可用。
超时退出 当阻塞队列满时，队列会阻塞生产者线程一定时间，超过后限时后生产者线程会退出

8.4.1 抛出异常 api

  1. package s02.e08;
  3. import java.util.concurrent.ArrayBlockingQueue;
  4. import java.util.concurrent.BlockingQueue;
  6. public class BlockingQueueDemo {
  7.     public static void main(String[] args) throws Exception {
  8.         BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);
  10.         System.out.println(blockingQueue.add("a"));
  11.         System.out.println(blockingQueue.add("b"));
  12.         System.out.println(blockingQueue.add("c"));
  13.         System.out.println(blockingQueue.add("x"));
  14.     }
  15. }

image.png

  1. package s02.e08;
  3. import java.util.concurrent.ArrayBlockingQueue;
  4. import java.util.concurrent.BlockingQueue;
  6. public class BlockingQueueDemo {
  7.     public static void main(String[] args) throws Exception {
  8.         BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);
  10.         System.out.println(blockingQueue.add("a"));
  11.         System.out.println(blockingQueue.add("b"));
  12.         System.out.println(blockingQueue.add("c"));
  14.         blockingQueue.remove();
  15.         blockingQueue.remove();
  16.         blockingQueue.remove();
  17.         blockingQueue.remove();
  18.     }
  19. }

image.png

  1. package s02.e08;
  3. import java.util.concurrent.ArrayBlockingQueue;
  4. import java.util.concurrent.BlockingQueue;
  6. public class BlockingQueueDemo {
  7.     public static void main(String[] args) throws Exception {
  8.         BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);
  10.         System.out.println(blockingQueue.add("a"));
  11.         System.out.println(blockingQueue.add("b"));
  12.         System.out.println(blockingQueue.add("c"));
  14.         System.out.println(blockingQueue.element());
  16.         blockingQueue.remove();
  17.         blockingQueue.remove();
  18.         blockingQueue.remove();
  19.     }
  20. }

image.png

8.4.2 特殊值 api

  1. package s02.e08;
  3. import java.util.concurrent.ArrayBlockingQueue;
  4. import java.util.concurrent.BlockingQueue;
  6. public class BlockingQueueDemo {
  7.     public static void main(String[] args) throws Exception {
  8.         BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);
  10.         System.out.println(blockingQueue.offer("a"));
  11.         System.out.println(blockingQueue.offer("a"));
  12.         System.out.println(blockingQueue.offer("a"));
  13.         System.out.println(blockingQueue.offer("x"));
  14.     }
  15. }

image.png

  1. package s02.e08;
  3. import java.util.concurrent.ArrayBlockingQueue;
  4. import java.util.concurrent.BlockingQueue;
  6. public class BlockingQueueDemo {
  7.     public static void main(String[] args) throws Exception {
  8.         BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);
  10.         System.out.println(blockingQueue.offer("a"));
  11.         System.out.println(blockingQueue.offer("b"));
  12.         System.out.println(blockingQueue.offer("c"));
  13.         System.out.println(blockingQueue.offer("x"));
  15.         System.out.println(blockingQueue.peek());
  16.     }
  17. }

image.png

  1. package s02.e08;
  3. import java.util.concurrent.ArrayBlockingQueue;
  4. import java.util.concurrent.BlockingQueue;
  6. public class BlockingQueueDemo {
  7.     public static void main(String[] args) throws Exception {
  8.         BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);
  10.         System.out.println(blockingQueue.offer("a"));
  11.         System.out.println(blockingQueue.offer("b"));
  12.         System.out.println(blockingQueue.offer("c"));
  13.         System.out.println(blockingQueue.offer("x"));
  15.         System.out.println(blockingQueue.peek());
  17.         System.out.println(blockingQueue.poll());
  18.         System.out.println(blockingQueue.poll());
  19.         System.out.println(blockingQueue.poll());
  20.         System.out.println(blockingQueue.poll());
  21.     }
  22. }

image.png

8.4.3 阻塞 api

  1. package s02.e08;
  3. import java.util.concurrent.ArrayBlockingQueue;
  4. import java.util.concurrent.BlockingQueue;
  6. public class BlockingQueueDemo {
  7.     public static void main(String[] args) throws Exception {
  8.         BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);
  10.         blockingQueue.put("a");
  11.         blockingQueue.put("a");
  12.         blockingQueue.put("a");
  13.         System.out.println("=========================");
  14.         blockingQueue.put("x");
  15.     }
  16. }

image.png

  1. package s02.e08;
  3. import java.util.concurrent.ArrayBlockingQueue;
  4. import java.util.concurrent.BlockingQueue;
  6. public class BlockingQueueDemo {
  7.     public static void main(String[] args) throws Exception {
  8.         BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);
  10.         blockingQueue.put("a");
  11.         blockingQueue.put("a");
  12.         blockingQueue.put("a");
  13.         System.out.println("=========================");
  15.         blockingQueue.take();
  16.         blockingQueue.take();
  17.         blockingQueue.take();
  18.         blockingQueue.take();
  19.     }
  20. }

image.png

8.4.4 超时 api

  1. package s02.e08;
  3. import java.util.concurrent.ArrayBlockingQueue;
  4. import java.util.concurrent.BlockingQueue;
  5. import java.util.concurrent.TimeUnit;
  7. public class BlockingQueueDemo {
  8.     public static void main(String[] args) throws Exception {
  9.         BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);
  11.         System.out.println(blockingQueue.offer("a", 2L, TimeUnit.SECONDS));
  12.         System.out.println(blockingQueue.offer("a", 2L, TimeUnit.SECONDS));
  13.         System.out.println(blockingQueue.offer("a", 2L, TimeUnit.SECONDS));
  14.         System.out.println(blockingQueue.offer("a", 2L, TimeUnit.SECONDS));
  15.     }
  16. }

image.png
最后一个等了 2 秒才 false

8.5 SynchronousQueue

SynchronousQueue 没有容量。
与其他 BlockingQueue 不同，SynchronousQueue 是一个不存储元素的 BlockingQueue。
每一个 put 操作必须要等待一个 take 操作，否则不能继续添加元素，反之亦然。

  1. package s02.e08;
  3. import java.util.concurrent.BlockingQueue;
  4. import java.util.concurrent.SynchronousQueue;
  5. import java.util.concurrent.TimeUnit;
  7. public class SynchronousQueueDemo {
  8.     public static void main(String[] args) throws InterruptedException {
  9.         BlockingQueue<String> blockingQueue = new SynchronousQueue<>();
  11.         new Thread(() -> {
  12.             try {
  13.                 System.out.println(Thread.currentThread().getName() + "\t put 1");
  14.                 blockingQueue.put("1");
  15.                 System.out.println(Thread.currentThread().getName() + "\t put 2");
  16.                 blockingQueue.put("2");
  17.                 System.out.println(Thread.currentThread().getName() + "\t put 3");
  18.                 blockingQueue.put("3");
  19.             } catch (InterruptedException e) {
  20.                 e.printStackTrace();
  21.             }
  22.         }, "AAA").start();
  24.         new Thread(() -> {
  25.             try {
  26.                 // 暂停一会儿线程
  27.                 try {
  28.                     TimeUnit.SECONDS.sleep(5);
  29.                 } catch (InterruptedException e) {
  30.                     e.printStackTrace();
  31.                 }
  32.                 System.out.println(Thread.currentThread().getName() + "\t" + blockingQueue.take());
  33.                 try {
  34.                     TimeUnit.SECONDS.sleep(5);
  35.                 } catch (InterruptedException e) {
  36.                     e.printStackTrace();
  37.                 }
  38.                 System.out.println(Thread.currentThread().getName() + "\t" + blockingQueue.take());
  39.                 try {
  40.                     TimeUnit.SECONDS.sleep(5);
  41.                 } catch (InterruptedException e) {
  42.                     e.printStackTrace();
  43.                 }
  44.                 System.out.println(Thread.currentThread().getName() + "\t" + blockingQueue.take());
  45.             } catch (InterruptedException e) {
  46.                 e.printStackTrace();
  47.             }
  48.         }, "BBB").start();
  49.     }
  50. }

image.png

8.6 用在哪里

  • 生产者消费者模式
  • 线程池
  • 消息中间件

    8.6.1 生产者消费者模式——传统版

    一个初始值为零的变量，两个线程对其交替操作，一个加 1 一个减 1 ，来 5 轮 ```java package s02.e08;

import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock;

// 资源类 class ShareData { private int number = 0; private Lock lock = new ReentrantLock(); private Condition condition = lock.newCondition();

  1. public void increment() throws Exception {
  2.     lock.lock();
  3.     try {
  4.         // 1 判断
  5.         while (number != 0) {
  6.             // 等待,不能生产
  7.             condition.await();
  8.         }
  9.         // 2 干活
  10.         number++;
  11.         System.out.println(Thread.currentThread().getName() + "\t" + number);
  12.         // 3 通知唤醒
  13.         condition.signalAll();
  14.     } catch (Exception e) {
  15.         e.printStackTrace();
  16.     } finally {
  17.         lock.unlock();
  18.     }
  19. }
  21. public void decrement() throws Exception {
  22.     lock.lock();
  23.     try {
  24.         // 1 判断
  25.         while (number == 0) {
  26.             // 等待,不能生产
  27.             condition.await();
  28.         }
  29.         // 2 干活
  30.         number--;
  31.         System.out.println(Thread.currentThread().getName() + "\t" + number);
  32.         // 3 通知唤醒
  33.         condition.signalAll();
  34.     } catch (Exception e) {
  35.         e.printStackTrace();
  36.     } finally {
  37.         lock.unlock();
  38.     }
  39. }

}

public class ProdConsumer_TraditionDemo { public static void main(String[] args) { ShareData shareData = new ShareData();

  1.     new Thread(() -> {
  2.         for (int i = 1; i <= 5; i++) {
  3.             try {
  4.                 shareData.increment();
  5.             } catch (Exception e) {
  6.                 e.printStackTrace();
  7.             }
  8.         }
  9.     }, "AA").start();
  11.     new Thread(() -> {
  12.         for (int i = 1; i <= 5; i++) {
  13.             try {
  14.                 shareData.decrement();
  15.             } catch (Exception e) {
  16.                 e.printStackTrace();
  17.             }
  18.         }
  19.     }, "BB").start();
  20. }

}

  1. ![image.png](https://cdn.nlark.com/yuque/0/2022/png/390086/1644769968129-3644d405-d7c9-4ee2-af70-a840d903bb03.png#clientId=ube101e58-d1ee-4&crop=0&crop=0&crop=1&crop=1&from=paste&height=216&id=u80484c9d&margin=%5Bobject%20Object%5D&name=image.png&originHeight=216&originWidth=69&originalType=binary&ratio=1&rotation=0&showTitle=false&size=1667&status=done&style=none&taskId=u24fadda7-1d29-48e1-8c75-03990b8396e&title=&width=69)<br />多线程的判断一定要用 while 判断，不能用 if 判断，否则会产生虚假唤醒<br />![image.png](https://cdn.nlark.com/yuque/0/2022/png/390086/1644770372605-773e4676-558f-454a-b0e0-263a86167c5e.png#clientId=ube101e58-d1ee-4&crop=0&crop=0&crop=1&crop=1&from=paste&height=202&id=ucc0bf973&margin=%5Bobject%20Object%5D&name=image.png&originHeight=202&originWidth=753&originalType=binary&ratio=1&rotation=0&showTitle=false&size=79637&status=done&style=none&taskId=ue9c80a50-8059-47d5-9c38-23f81e7d6ef&title=&width=753)
  2. ```java
  3. package s02.e08;
  5. import java.util.concurrent.locks.Condition;
  6. import java.util.concurrent.locks.Lock;
  7. import java.util.concurrent.locks.ReentrantLock;
  9. // 资源类
  10. class ShareData {
  11.     private int number = 0;
  12.     private Lock lock = new ReentrantLock();
  13.     private Condition condition = lock.newCondition();
  15.     public void increment() throws Exception {
  16.         lock.lock();
  17.         try {
  18.             // 1 判断
  19.             if (number != 0) {
  20.                 // 等待,不能生产
  21.                 condition.await();
  22.             }
  23.             // 2 干活
  24.             number++;
  25.             System.out.println(Thread.currentThread().getName() + "\t" + number);
  26.             // 3 通知唤醒
  27.             condition.signalAll();
  28.         } catch (Exception e) {
  29.             e.printStackTrace();
  30.         } finally {
  31.             lock.unlock();
  32.         }
  33.     }
  35.     public void decrement() throws Exception {
  36.         lock.lock();
  37.         try {
  38.             // 1 判断
  39.             if (number == 0) {
  40.                 // 等待,不能生产
  41.                 condition.await();
  42.             }
  43.             // 2 干活
  44.             number--;
  45.             System.out.println(Thread.currentThread().getName() + "\t" + number);
  46.             // 3 通知唤醒
  47.             condition.signalAll();
  48.         } catch (Exception e) {
  49.             e.printStackTrace();
  50.         } finally {
  51.             lock.unlock();
  52.         }
  53.     }
  54. }
  56. public class ProdConsumer_TraditionDemo {
  57.     public static void main(String[] args) {
  58.         ShareData shareData = new ShareData();
  60.         new Thread(() -> {
  61.             for (int i = 1; i <= 5; i++) {
  62.                 try {
  63.                     shareData.increment();
  64.                 } catch (Exception e) {
  65.                     e.printStackTrace();
  66.                 }
  67.             }
  68.         }, "AA").start();
  70.         new Thread(() -> {
  71.             for (int i = 1; i <= 5; i++) {
  72.                 try {
  73.                     shareData.decrement();
  74.                 } catch (Exception e) {
  75.                     e.printStackTrace();
  76.                 }
  77.             }
  78.         }, "BB").start();
  80.         new Thread(() -> {
  81.             for (int i = 1; i <= 5; i++) {
  82.                 try {
  83.                     shareData.increment();
  84.                 } catch (Exception e) {
  85.                     e.printStackTrace();
  86.                 }
  87.             }
  88.         }, "CC").start();
  90.         new Thread(() -> {
  91.             for (int i = 1; i <= 5; i++) {
  92.                 try {
  93.                     shareData.decrement();
  94.                 } catch (Exception e) {
  95.                     e.printStackTrace();
  96.                 }
  97.             }
  98.         }, "DD").start();
  99.     }
  100. }

image.png
这里会有生产 2 个的情况

8.6.2 Synchronized 和 Lock 有什么区别

  1. 原始构成

synchronized 是关键字属于 JVM 层面
monitorenter(底层是通过 monitor 对象来完成，其实 wait/notify 等方法也依赖于 monitor 对象只有在同步块或方法中才能调 wait/notify 等方法)
monitorexit
Lock 是具体类(java.util.concurrent.Locks.Lock)是 api 层面的锁

  1. package s02.e08;
  3. import java.util.concurrent.locks.ReentrantLock;
  5. public class SyncAndReentrantLockDemo {
  6.     public static void main(String[] args) {
  7.         synchronized (new Object()) {
  9.         }
  11.         new ReentrantLock();
  12.     }
  13. }

image.png
出现两个 monitorexit 是为了保证不会死锁

  1. 使用方法

synchronized 不需要用户去手动释放锁，当 synchronized 代码执行完后系统会自动让线程释放对锁的占用
ReentrantLock 则需要用户去手动释放锁若没有主动释放锁，就有可能导致出现死锁现象
需要 lock( ) 和 unlock() 方法配合 try/finally 语句块来完成

  1. 等待是否可中断

synchronized 不可中断，除非抛出异常或者正常运行完成
ReentrantLock 可中断

  • 设置超时方法 tryLock(Long timeout，TimeUnit unit)
  • LockInterruptibly() 放代码块中，调用 interrupt() 方法可中断
  1. 加锁是否公平

synchronized 非公平锁
ReentrantLock 两者都可以，默认非公平锁，构造方法可以传入 boolean 值，true 为公平锁，false 为非公平锁

  1. 锁绑定多个条件 condition

synchronized 没有
ReentrantLock 用来实现分组唤醒需要唤醒的线程们，可以精确唤醒，而不是像 synchronized 要么随机唤醒一个线程要么唤醒全部线程。
题目：多线程之间按顺序调用，实现 A->B->C 三个线程启动，要求如下
AA 打印 5 次，BB 打印 10 次，CC 打印 15 次，紧接着 AA 打印 5 次，BB 打印 10次，CC 打印 15 次，来10轮

  1. package s02.e08;
  3. import java.util.concurrent.locks.Condition;
  4. import java.util.concurrent.locks.Lock;
  5. import java.util.concurrent.locks.ReentrantLock;
  7. class ShareResource {
  8.     private int number = 1; //A:1 B:2 c:3
  9.     private Lock lock = new ReentrantLock();
  10.     private Condition c1 = lock.newCondition();
  11.     private Condition c2 = lock.newCondition();
  12.     private Condition c3 = lock.newCondition();
  15.     public void print5() {
  16.         lock.lock();
  17.         try {
  18.             // 1. 判断
  19.             while (number != 1) {
  20.                 c1.await();
  21.             }
  23.             // 2. 干活
  24.             for (int i = 1; i <= 5; i++) {
  25.                 System.out.println(Thread.currentThread().getName() + "\t" + i);
  26.             }
  28.             // 3. 通知
  29.             number = 2;
  30.             c2.signal();
  31.         } catch (Exception e) {
  32.             e.printStackTrace();
  33.         } finally {
  34.             lock.unlock();
  35.         }
  36.     }
  38.     public void print10() {
  39.         lock.lock();
  40.         try {
  41.             // 1. 判断
  42.             while (number != 2) {
  43.                 c2.await();
  44.             }
  46.             // 2. 干活
  47.             for (int i = 1; i <= 10; i++) {
  48.                 System.out.println(Thread.currentThread().getName() + "\t" + i);
  49.             }
  51.             // 3. 通知
  52.             number = 3;
  53.             c3.signal();
  54.         } catch (Exception e) {
  55.             e.printStackTrace();
  56.         } finally {
  57.             lock.unlock();
  58.         }
  59.     }
  61.     public void print15() {
  62.         lock.lock();
  63.         try {
  64.             // 1. 判断
  65.             while (number != 3) {
  66.                 c3.await();
  67.             }
  69.             // 2. 干活
  70.             for (int i = 1; i <= 15; i++) {
  71.                 System.out.println(Thread.currentThread().getName() + "\t" + i);
  72.             }
  74.             // 3. 通知
  75.             number = 1;
  76.             c1.signal();
  77.         } catch (Exception e) {
  78.             e.printStackTrace();
  79.         } finally {
  80.             lock.unlock();
  81.         }
  82.     }
  83. }
  85. public class SyncAndReentrantLockDemo {
  86.     public static void main(String[] args) {
  87.         ShareResource shareResource = new ShareResource();
  88.         new Thread(() -> {
  89.             for (int i = 1; i <= 10; i++) {
  90.                 shareResource.print5();
  91.             }
  92.         }, "A").start();
  93.         new Thread(() -> {
  94.             for (int i = 1; i <= 10; i++) {
  95.                 shareResource.print10();
  96.             }
  97.         }, "B").start();
  98.         new Thread(() -> {
  99.             for (int i = 1; i <= 10; i++) {
  100.                 shareResource.print15();
  101.             }
  102.         }, "C").start();
  103.     }
  104. }

image.png

8.6.3 线程通信之生产者消费者阻塞队列版

  1. package s02.e08;
  3. import java.util.concurrent.ArrayBlockingQueue;
  4. import java.util.concurrent.BlockingQueue;
  5. import java.util.concurrent.TimeUnit;
  6. import java.util.concurrent.atomic.AtomicInteger;
  8. class MyResource {
  9.     private volatile boolean FLAG = true; // 默认开启，进行生产+消费
  10.     private AtomicInteger atomicInteger = new AtomicInteger();
  12.     BlockingQueue<String> blockingQueue = null;
  14.     public MyResource(BlockingQueue<String> blockingQueue) {
  15.         this.blockingQueue = blockingQueue;
  16.         System.out.println(blockingQueue.getClass().getName());
  17.     }
  19.     public void myprod() throws Exception {
  20.         String data = null;
  21.         boolean retvalue;
  22.         while (FLAG) {
  23.             data = atomicInteger.incrementAndGet() + "";
  24.             retvalue = blockingQueue.offer(data, 2L, TimeUnit.SECONDS);
  25.             if (retvalue) {
  26.                 System.out.println(Thread.currentThread().getName() + "\t插入队列" + data + "成功");
  27.             } else {
  28.                 System.out.println(Thread.currentThread().getName() + "\t插入队列" + data + "失败");
  29.             }
  30.             TimeUnit.SECONDS.sleep(1);
  31.         }
  32.         System.out.println(Thread.currentThread().getName() + "\t大老板叫停了，表示FLAG=false,生产动作结束");
  33.     }
  35.     public void myConsumer() throws Exception {
  36.         String result = null;
  37.         while (FLAG) {
  38.             result = blockingQueue.poll(2L, TimeUnit.SECONDS);
  39.             if (null == result || result.equalsIgnoreCase("")) {
  40.                 FLAG = false;
  41.                 System.out.println(Thread.currentThread().getName() + "\t超过2秒钟没有取到蛋糕，消费退出");
  42.                 System.out.println();
  43.                 System.out.println();
  44.                 return;
  45.             }
  46.             System.out.println(Thread.currentThread().getName() + "\t消费队列蛋糕" + result + "成功");
  47.         }
  48.     }
  50.     public void stop() throws Exception {
  51.         this.FLAG = false;
  52.     }
  53. }
  55. /**
  56.  * volatile/cAs/atomicInteger/BLockQueue/线程交互/原子引用
  57.  */
  58. public class Prodconsumer_BlockQueueDemo {
  59.     public static void main(String[] args) throws Exception {
  60.         MyResource myResource = new MyResource(new ArrayBlockingQueue<>(10));
  62.         new Thread(() -> {
  63.             System.out.println(Thread.currentThread().getName() + "\t生产线程启动");
  64.             try {
  65.                 myResource.myprod();
  66.             } catch (Exception e) {
  67.                 e.printStackTrace();
  68.             }
  69.         }, "Prod").start();
  71.         new Thread(() -> {
  72.             System.out.println(Thread.currentThread().getName() + "\t 消费线程启动");
  73.             System.out.println();
  74.             System.out.println();
  75.             try {
  76.                 myResource.myConsumer();
  77.                 System.out.println();
  78.                 System.out.println();
  79.             } catch (Exception e) {
  80.                 e.printStackTrace();
  81.             }
  82.         }, "Consumer").start();
  84.         // 暂停一会儿线程
  85.         try {
  86.             TimeUnit.SECONDS.sleep(5);
  87.         } catch (InterruptedException e) {
  88.             e.printStackTrace();
  89.         }
  90.         System.out.println();
  91.         System.out.println();
  92.         System.out.println();
  94.         System.out.println("5秒钟时间到，大老板main线程叫停，活动结束");
  96.         myResource.stop();
  97.     }
  98. }

image.png