队列

阻塞队列

  1. 阻塞队列有没有好的一面？
  2. 不得不阻塞，你如何管理？

定义

阻塞队列，首先是一个队列，而一个阻塞队列在数据结构中所起的作用大概如图所示：
image.png

  • 当阻塞队列是空的时候，从队列中获取元素的操作将会被阻塞
  • 当阻塞队列是满的时候，往队列中添加元素的操作将会被阻塞
  • 试图从空的阻塞队列中获取元素将会被阻塞，直到其他线程往空的队列插入显得元素，同比插入元素。

    什么好处？

    在多线程领域，所谓阻塞，在某些情况下会挂起线程，即阻塞，一旦条件满足，被挂起的线程又会自动被唤醒。
    为什么需要阻塞队列？

    不需要关心什么时候需要阻塞线程，什么时候需要唤醒线程，因为这一切被BlockingQueue一手包办了。也不需要考虑效率和线程安全。

架构梳理和种类分析

  • 红色重点关注 | 分类 | 说明 | | —- | —- | | ArrayBlockingQueue | 是一个基于数组结构的有界阻塞队列，此队列按FIFO原则对元素进行操作。 | | LinkedBlockingQueue | 基于链表结构的有界阻塞队列，按照FIFO排序元素，吞吐量通常高于ArrayBlockingQueue.但是大小是Integer.MAX_VALUE | | PriorityBlockingQueue | 支持优先级排序的无界阻塞队列 | | DelayQueue | 使用优先级队列实现的延迟无界阻塞队列 | | LinkedTransferQueue | 由链表结构组成的无界阻塞队列 | | SynchrobousBlockingQueue | 一种不存储元素的阻塞队列，每个插入操作必须等待另外一个线程调用移除操作，否则插入操作一直处于阻塞状态 | | LinkedBlockingDueue | 由链表结构组成的双向阻塞队列。 |

核心方法

image.png
image.png

element检查队首元素是谁。队列先进先出，队首元素。

  1. package com.interview.demo;
  3. import java.util.concurrent.BlockingQueue;
  4. import java.util.concurrent.SynchronousQueue;
  5. import java.util.concurrent.TimeUnit;
  7. /**
  8.  * @Author leijs
  9.  * @date 2022/3/29
  10.  */
  11. public class SynchronousQueueDemo {
  12.     public static void main(String[] args) {
  13.         BlockingQueue<String> blockingQueue = new SynchronousQueue<>();
  14.         new Thread(() -> {
  15.             try {
  16.                 System.out.println(Thread.currentThread().getName() + "生产a");
  17.                 blockingQueue.put("a");
  18.                 System.out.println(Thread.currentThread().getName() + "生产b");
  19.                 blockingQueue.put("b");
  20.                 System.out.println(Thread.currentThread().getName() + "生产c");
  21.                 blockingQueue.put("c");
  22.             } catch (InterruptedException e) {
  23.                 e.printStackTrace();
  24.             }
  25.         }, "A").start();
  26.         new Thread(() -> {
  27.             try {
  28.                 System.out.println(Thread.currentThread().getName() + "拿走a");
  29.                 blockingQueue.take();
  30.                 TimeUnit.SECONDS.sleep(2);
  31.                 System.out.println(Thread.currentThread().getName() + "拿走b");
  32.                 blockingQueue.take();
  33.                 TimeUnit.SECONDS.sleep(2);
  34.                 System.out.println(Thread.currentThread().getName() + "拿走c");
  35.                 blockingQueue.take();
  36.             } catch (InterruptedException e) {
  37.                 e.printStackTrace();
  38.             }
  39.         }, "B").start();
  40.     }
  41. }

生产者消费者模式

传统版本

  1. package com.interview.demo.producer_consumer;
  3. import java.util.concurrent.locks.Condition;
  4. import java.util.concurrent.locks.Lock;
  5. import java.util.concurrent.locks.ReentrantLock;
  7. /**
  8.  * 一个初始值为0的变量，两个线程交替操作，一个加1，一个减1，来5轮
  9.  *
  10.  * @Author leijs
  11.  * @date 2022/3/29
  12.  */
  13. public class ProdConsumerTraditionDemo {
  15.     public static void main(String[] args) {
  16.         ShareData shareData = new ShareData();
  17.         new Thread(() -> {
  18.             for (int i = 0; i < 5; i++) {
  19.                 shareData.increment();
  20.             }
  21.         }, "t1").start();
  22.         new Thread(() -> {
  23.             for (int i = 0; i < 5; i++) {
  24.                 shareData.decrement();
  25.             }
  26.         }, "t2").start();
  27.     }
  28. }
  30. class ShareData {
  31.     private int num = 0;
  32.     private Lock lock = new ReentrantLock();
  33.     private Condition condition = lock.newCondition();
  35.     public void increment() {
  36.         lock.lock();
  37.         try {
  38.             while (num != 0) {
  39.                 // 等待，不能生产
  40.                 condition.await();
  41.             }
  42.             num++;
  43.             System.out.println(Thread.currentThread().getName() + ":" + num);
  44.             condition.signalAll();
  45.         } catch (InterruptedException e) {
  46.             e.printStackTrace();
  47.         } finally {
  48.             lock.unlock();
  49.         }
  50.     }
  52.     public void decrement() {
  53.         lock.lock();
  54.         try {
  55.             // 注意多线程的判断不能用if只能使用while
  56.             while (num == 0) {
  57.                 // 等待，不能生产
  58.                 condition.await();
  59.             }
  60.             num--;
  61.             System.out.println(Thread.currentThread().getName() + ":" + num);
  62.             condition.signalAll();
  63.         } catch (InterruptedException e) {
  64.             e.printStackTrace();
  65.         } finally {
  66.             lock.unlock();
  67.         }
  68.     }
  69. }

BlockingQueue实现

  1. package com.interview.demo.producer_consumer;
  3. import org.apache.commons.lang3.StringUtils;
  5. import java.util.concurrent.ArrayBlockingQueue;
  6. import java.util.concurrent.BlockingQueue;
  7. import java.util.concurrent.TimeUnit;
  8. import java.util.concurrent.atomic.AtomicInteger;
  10. /**
  11.  * @Author leijs
  12.  * @date 2022/3/29
  13.  */
  14. public class ProdConsumerBlockingQueueDemo {
  16.     public static void main(String[] args) throws InterruptedException {
  17.         MyResource myResource = new MyResource(new ArrayBlockingQueue<>(10));
  18.         new Thread(() -> {
  19.             try {
  20.                 myResource.produce();
  21.             } catch (Exception e) {
  22.                 e.printStackTrace();
  23.             }
  24.         }, "producer").start();
  25.         new Thread(() -> {
  26.             try {
  27.                 myResource.consumer();
  28.             } catch (Exception e) {
  29.                 e.printStackTrace();
  30.             }
  31.         }, "consumer").start();
  32.         TimeUnit.SECONDS.sleep(5);
  33.         System.out.println("主线程叫停了");
  34.         myResource.stop();
  36.     }
  37. }
  39. class MyResource {
  41.     /**
  42.      * 默认开启进行生产+消费
  43.      */
  44.     private volatile boolean FLAG = true;
  45.     private AtomicInteger atomicInteger = new AtomicInteger();
  47.     BlockingQueue<String> blockingQueue = null;
  49.     public MyResource(BlockingQueue<String> blockingQueue) {
  50.         this.blockingQueue = blockingQueue;
  51.         System.out.println(blockingQueue.getClass().getName());
  52.     }
  54.     public void produce() throws Exception {
  55.         String data;
  56.         boolean returnValue;
  57.         while (FLAG) {
  58.             data = atomicInteger.incrementAndGet() + "";
  59.             returnValue = blockingQueue.offer(data, 2L, TimeUnit.SECONDS);
  60.             if (returnValue) {
  61.                 System.out.println(Thread.currentThread().getName() + "插入队列成功" + data);
  62.             } else {
  63.                 System.out.println(Thread.currentThread().getName() + "插入队列失败" + data);
  64.             }
  65.             TimeUnit.SECONDS.sleep(1);
  66.         }
  67.         System.out.println("生产动作结束,flag=false");
  68.     }
  70.     public void consumer() throws Exception {
  71.         String data;
  72.         while (FLAG) {
  73.             data = blockingQueue.poll(2L, TimeUnit.SECONDS);
  74.             if (StringUtils.isEmpty(data)) {
  75.                 FLAG = false;
  76.                 System.out.println(Thread.currentThread().getName() + "超过两秒钟没有获取数据，退出");
  77.                 return;
  78.             }
  79.             System.out.println(Thread.currentThread().getName() + "消费队列成功:" + data);
  80.         }
  81.         System.out.println("消费动作结束");
  82.     }
  84.     public void stop() {
  85.         this.FLAG = false;
  86.     }
  88. }

扩展

Synchronized和Lock的区别

  1. synchronized是关键字，属于JVM层面。

    1. monitorenter(底层是通过Monitor对象来完成的，其实wait/notify)等方法也依赖monitor对象，只有在同步代表快中才能调用wait/notify等方法<br />       monitorexit<br />       Lock是具体类，是API层面的

  2. 使用方法

    1. synchronized 不需要手动改释放锁，代码执行完系统会自动让线程释放对锁的占用。<br />       ReentrantLock则需要用户去手动释放锁，如果没有主动释放锁，就可能出现死锁现象<br />       lock和unlock方法配合try...finally来完成

  3. 等待是否可中断

    1. synchronized 是不可中断的，除非抛出异常或者正常运行完成<br />       lock可中断。可以设置超时方法或者lockInterruptibly

  4. 加锁是否公平 

    1. synchronized 只能是非公平锁<br />       lock两者都可以，构造函数传入true/false

  5. 锁绑定多个对象Condition

    1. Synchronized没有<br />       Reentrant用来实现分组唤醒需要唤醒的线程们，可以精确唤醒，而不是像Synchronized那样随机唤醒一个线程或者唤醒全部线程。

多个线程按照顺序打印

  1. package com.interview.demo;
  3. import java.util.concurrent.locks.Condition;
  4. import java.util.concurrent.locks.Lock;
  5. import java.util.concurrent.locks.ReentrantLock;
  7. /**
  8.  * A 打印1次，B打印2次，C打印3次
  9.  * 循环5次
  10.  *
  11.  * @Author leijs
  12.  * @date 2022/3/29
  13.  */
  14. public class SyncAndReentrantLockDemo {
  15.     public static void main(String[] args) {
  16.         ShareData shareData = new ShareData();
  17.         new Thread(() -> {
  18.             for (int i = 0; i < 5; i++) {
  19.                 shareData.print1();
  20.             }
  21.         }, "A").start();
  22.         new Thread(() -> {
  23.             for (int i = 0; i < 5; i++) {
  24.                 shareData.print2();
  25.             }
  26.         }, "B").start();
  27.         new Thread(() -> {
  28.             for (int i = 0; i < 5; i++) {
  29.                 shareData.print3();
  30.             }
  31.         }, "C").start();
  33.     }
  34. }
  36. class ShareData {
  37.     // A:1, B:2, C:3
  38.     private int num = 1;
  39.     // 判断 -> 干活 -> 通知唤醒
  40.     private Lock lock = new ReentrantLock();
  41.     private Condition c1 = lock.newCondition();
  42.     private Condition c2 = lock.newCondition();
  43.     private Condition c3 = lock.newCondition();
  45.     public void print1() {
  46.         lock.lock();
  47.         try {
  48.             while (num != 1) {
  49.                 c1.await();
  50.             }
  51.             printByTimes(1);
  52.             num = 2;
  53.             c2.signal();
  54.         } catch (InterruptedException e) {
  55.             e.printStackTrace();
  56.         } finally {
  57.             lock.unlock();
  58.         }
  59.     }
  61.     public void print2() {
  62.         lock.lock();
  63.         try {
  64.             while (num != 2) {
  65.                 c2.await();
  66.             }
  67.             printByTimes(2);
  68.             num = 3;
  69.             c3.signal();
  70.         } catch (InterruptedException e) {
  71.             e.printStackTrace();
  72.         } finally {
  73.             lock.unlock();
  74.         }
  75.     }
  77.     public void print3() {
  78.         lock.lock();
  79.         try {
  80.             while (num != 3) {
  81.                 c3.await();
  82.             }
  83.             printByTimes(3);
  84.             num = 1;
  85.             c1.signal();
  86.         } catch (InterruptedException e) {
  87.             e.printStackTrace();
  88.         } finally {
  89.             lock.unlock();
  90.         }
  91.     }
  93.     private void printByTimes(int times) {
  94.         for (int i = 0; i < times; i++) {
  95.             System.out.println(Thread.currentThread().getName() + "打印" + i);
  96.         }
  97.     }
  98. }

callable

  1. package com.interview.demo;
  3. import java.util.concurrent.Callable;
  4. import java.util.concurrent.ExecutionException;
  5. import java.util.concurrent.FutureTask;
  7. /**
  8.  * @Author leijs
  9.  * @date 2022/3/29
  10.  */
  11. public class CallableDemo {
  12.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  13.         FutureTask<Integer> futureTask = new FutureTask<>(new MyThread());
  14.         Thread thread = new Thread(futureTask, "AA");
  15.         thread.start();
  16.         System.out.println("*********** result:" + futureTask.get());
  17.     }
  18. }
  20. class MyThread implements Callable<Integer> {
  22.     @Override
  23.     public Integer call() throws Exception {
  24.         return 1024;
  25.     }
  26. }

futureTask.get()建议放在最后。如果没有计算完成就要去强求，就会导致阻塞，值得计算完成。