wait - notify

Java 为每个对象提供了 wait() 方法使线程阻塞等待，notify() 方法来唤醒由于 wait() 方法阻塞等待的线程。使用这两个方法的前提是当前线程持有该对象的监视器锁（隐式锁），而给对象添加隐式锁的方式就是通过 synchronized 关键字声明。

  • wait()，当前线程释放该对象的监视器锁，等待被唤醒或者被中断退出等待
  • wait(long timeoutMillis)，同 wait() 方法，同时如果超过指定时间没被唤醒，可以退出等待
  • notify()，唤醒一个正在阻塞等待的线程
  • notifyAll()，唤醒所有的正在阻塞等待的线程

wait() 方法会释放对对象的监视器锁，notify() 方法只是唤醒阻塞等待的线程，和其它线程去竞争获取对象的监视器锁，具体能不能获取到监视器锁，要看 CPU 的具体调度。 代码示例：两个线程交替打印数字隐式锁锁住同一个对象，一个线程打印完后，唤醒另一个线程，然后阻塞等待另一个线程唤醒。

  1. public class Main {
  2.     static class Counter {
  3.         private int count = 0;
  5.         public synchronized void printX() {
  6.             while (count < 100 && count % 2 == 0) {
  7.                 count++;
  8.                 System.out.println(Thread.currentThread().getName() + count);
  9.                 notifyAll();
  10.             }
  12.             try {
  13.                 wait();
  14.             } catch (InterruptedException e) {
  15.                 e.printStackTrace();
  16.                 return;
  17.             }
  18.         }
  20.         public synchronized void printY() {
  21.             while (count < 100 && count % 2 == 1) {
  22.                 count++;
  23.                 System.out.println(Thread.currentThread().getName() + count);
  24.                 notifyAll();
  25.             }
  27.             try {
  28.                 wait();
  29.             } catch (InterruptedException e) {
  30.                 e.printStackTrace();
  31.                 return;
  32.             }
  33.         }
  34.     }
  36.     public static void main(String[] args) {
  37.         final Counter counter = new Counter();
  38.         new Thread(() -> {
  39.             while (counter.count < 100) {
  40.                 counter.printX();
  41.             }
  42.             Thread.currentThread().interrupt();
  43.         }, "线程A: ").start();
  45.         new Thread(() -> {
  46.             while (counter.count < 100) {
  47.                 counter.printY();
  48.             }
  49.             Thread.currentThread().interrupt();
  50.         }, "线程B: ").start();
  52.         try {
  53.             Thread.sleep(10000);
  54.         } catch (InterruptedException e) {
  55.             e.printStackTrace();
  56.         }
  57.     }
  58. }

await - signal

如果使用显式锁替代了隐式锁使用，则应该使用 Condition 类提供的 await() 和 signal() 替代 wait() 和 notify()。
显示锁提供了 newCondition() 方法来声明一个条件，其实就是一个普通对象，它提供了如下方法：

  • await()，自动释放锁，进入阻塞等待状态，等待被唤醒或者被中断退出等待
  • awaitUninterruptibly()，释放锁，进入阻塞等待状态，无法通过中断唤醒
  • signal()，唤醒一个阻塞等待的线程
  • signalAll()，唤醒所有的正在阻塞等待的线程

和隐式锁的使用一样，调用以上方法需要保证当前线程持有该 Condition 条件对象的锁，否则会报 IllegalMonitorStateException 异常。

  1. import java.util.ArrayList;
  2. import java.util.List;
  3. import java.util.concurrent.ThreadLocalRandom;
  4. import java.util.concurrent.atomic.AtomicBoolean;
  5. import java.util.concurrent.locks.Condition;
  6. import java.util.concurrent.locks.ReentrantLock;
  8. public class Main {
  9.     static class Factory {
  10.         private volatile AtomicBoolean RUNNING_STATUS = new AtomicBoolean(true);
  12.         private final int MIN_SIZE = 0;
  13.         private final int MAX_SIZE = 10;
  14.         private List<Integer> blockedQueue = new ArrayList<>(MAX_SIZE);
  16.         ReentrantLock lock = new ReentrantLock();
  17.         Condition full = lock.newCondition();
  18.         Condition empty = lock.newCondition();
  20.         public void produce() {
  21.             lock.lock();
  22.             while (blockedQueue.size() >= MAX_SIZE) {
  23.                 System.out.println("阻塞队列满，" + Thread.currentThread().getName() + " 等待生产");
  24.                 try {
  25.                     full.await();
  26.                 } catch (InterruptedException e) {
  27.                     e.printStackTrace();
  28.                     return;
  29.                 }
  30.             }
  32.             try {
  33.                 final int num = ThreadLocalRandom.current().nextInt();
  34.                 blockedQueue.add(num);
  35.                 System.out.println(Thread.currentThread().getName() + " 生产：" + num);
  37.                 Thread.sleep(100);
  38.                 empty.signalAll();
  39.             } catch (InterruptedException e) {
  40.                 e.printStackTrace();
  41.             } finally {
  42.                 lock.unlock();
  43.             }
  44.         }
  46.         public void consume() {
  47.             lock.lock();
  48.             while (blockedQueue.size() <= MIN_SIZE) {
  49.                 System.out.println("阻塞队列空，" + Thread.currentThread().getName() + " 等待消费");
  50.                 try {
  51.                     empty.await();
  52.                 } catch (InterruptedException e) {
  53.                     e.printStackTrace();
  54.                 }
  55.             }
  57.             try {
  58.                 System.out.println(Thread.currentThread().getName() + " 消费：" + blockedQueue.remove(0));
  60.                 Thread.sleep(100);
  61.                 full.signalAll();
  62.             } catch (InterruptedException e) {
  63.                 e.printStackTrace();
  64.             } finally {
  65.                 lock.unlock();
  66.             }
  67.         }
  70.         public boolean running() {
  71.             return RUNNING_STATUS.get();
  72.         }
  74.         public void stop() {
  75.             RUNNING_STATUS.set(false);
  76.         }
  77.     }
  79.     public static void main(String[] args) {
  80.         Factory factory = new Factory();
  82.         new Thread(() -> {
  83.             while (factory.running()) {
  84.                 factory.produce();
  85.             }
  86.             Thread.currentThread().interrupt();
  87.         }, "线程A").start();
  89.         new Thread(() -> {
  90.             while (factory.running()) {
  91.                 factory.produce();
  92.             }
  93.             Thread.currentThread().interrupt();
  94.         }, "线程B").start();
  96.         new Thread(() -> {
  97.             while (factory.running()) {
  98.                 factory.consume();
  99.             }
  100.             Thread.currentThread().interrupt();
  101.         }, "线程C").start();
  103.         new Thread(() -> {
  104.             while (factory.running()) {
  105.                 factory.consume();
  106.             }
  107.             Thread.currentThread().interrupt();
  108.         }, "线程D").start();
  110.         try {
  111.             // 两秒后停止生产和消费
  112.             Thread.sleep(2000);
  113.             factory.stop();
  114.         } catch (InterruptedException e) {
  115.             e.printStackTrace();
  116.         }
  117.     }
  118. }

在线程协调的条件块中，使用 while 而不是 if 通过wait() 或者 await() 使线程等待。因为当线程重新获取到锁后，会从 wait() 或 await() 处重新开始执行，此时竞态条件可能已经发生变化，所以需要再次进入循环判断。

阻塞队列

JDK 5 新增了队列这种集合，并实现了很多常用的阻塞队列，阻塞队列通过在内部使用锁或其它并发控制来实现原子操作，它是线程安全的，我们不需要手动去实现加锁解锁阻塞等待等操作了，可以像使用普通集合类那样方便。阻塞队列 BlockingQueue 接口提供了如下方法：

  • 插入元素
    • put(E e)，直接插入元素，如果没有可用空间则阻塞等待
    • add(E e)，容量足够则直接插入并返回 true，超出容量限制则抛 IllegalStateException 异常
    • offer(E e)，容量足够则直接插入并返回 true，超出容量限制则返回 false
    • offer(E e, long timeout, TimeUnit unit)，容量足够则直接插入并返回 true，否则等待指定时间再尝试插入
  • 删除元素
    • take()，检索并删除第一个元素，如果没有元素则阻塞等待
    • poll(long timeout, TimeUnit unit)，检索并删除第一个元素，如果没有元素则阻塞等待指定时间再尝试，超时没有则返回 null
    • remove(Object o)，删除指定元素的单个实例，有则返回 true；contains(Object o) 判断是否含有某个元素
    • remainingCapacity()，获取队列剩余可用容量，没有限制则返回 Integer.MAX_VALUE

阻塞队列不能插入 null 值，因为 null 用作标识 poll 操作失败。阻塞队列通常用于特定场景，尽量不要当作普通队列使用。

ArrayBlockingQueue

基于数组的、有界（必须指定 capacity）、先进先出（FIFO）的阻塞队列。
内部使用可重入锁 ReentrantLock，可选择公平策略（默认非公平）。

  1. import java.util.concurrent.ArrayBlockingQueue;
  2. import java.util.concurrent.BlockingQueue;
  3. import java.util.concurrent.ThreadLocalRandom;
  5. public class Main {
  7.     static class Producer implements Runnable {
  8.         private final BlockingQueue<Integer> queue;
  10.         public Producer(BlockingQueue<Integer> queue) {
  11.             this.queue = queue;
  12.         }
  14.         @Override
  15.         public void run() {
  16.             while (true) {
  17.                 try {
  18.                     Thread.sleep(50);
  19.                     queue.put(produce());
  20.                 } catch (InterruptedException e) {
  21.                     e.printStackTrace();
  22.                     return;
  23.                 }
  24.             }
  25.         }
  27.         private Integer produce() {
  28.             final int num = ThreadLocalRandom.current().nextInt();
  29.             System.out.println(Thread.currentThread().getName() + " 生产: " + num);
  30.             return num;
  31.         }
  32.     }
  34.     static class Consumer implements Runnable {
  35.         private final BlockingQueue<Integer> queue;
  37.         public Consumer(BlockingQueue<Integer> queue) {
  38.             this.queue = queue;
  39.         }
  41.         @Override
  42.         public void run() {
  43.             while (true) {
  44.                 try {
  45.                     Thread.sleep(80);
  46.                     consume(queue.take());
  47.                 } catch (InterruptedException e) {
  48.                     e.printStackTrace();
  49.                     return;
  50.                 }
  51.             }
  52.         }
  54.         private void consume(Integer take) {
  55.             System.out.println(Thread.currentThread().getName() + " 消费: " + take);
  56.         }
  57.     }
  60.     public static void main(String[] args) {
  61.         final ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);
  63.         final Thread a = new Thread(new Producer(queue), "A");
  64.         final Thread b = new Thread(new Producer(queue), "B");
  65.         final Thread c = new Thread(new Consumer(queue), "C");
  66.         final Thread d = new Thread(new Consumer(queue), "D");
  67.         a.start();
  68.         b.start();
  69.         c.start();
  70.         d.start();
  72.         try {
  73.             Thread.sleep(1000);
  74.             a.interrupt();
  75.             Thread.sleep(1000);
  76.             b.interrupt();
  77.             Thread.sleep(1000);
  78.             c.interrupt();
  79.             Thread.sleep(1000);
  80.             d.interrupt();
  81.             Thread.sleep(1000);
  82.         } catch (InterruptedException e) {
  83.             e.printStackTrace();
  84.         }
  85.     }
  86. }

LinkedBlockingQueue

基于链表的、理论有界（Integer.MAX_VALUE）、先进先出（FIFO）的阻塞队列。
相比于 ArrayBlockingQueue 有更高的吞吐量，但性能较差。

LinkedBlockingDequeJDK 6

基于链表的、有界（默认 Integer.MAX_VALUE，可选 capacity）的阻塞双端队列。

PriorityBlockingQueue

基于优先级队列 PriorityQueue、无界的阻塞队列。
因为是无界的，插入元素不会阻塞，获取元素可以阻塞。可以通过 Comparator 参数指定如何比较优先级并提供实现 Comparable 接口的元素，不指定默认以元素插入顺序。

DelayQueue

包含具有延迟指定时间才可用的元素、无界的阻塞队列。
元素需要实现 Delayed 接口声明延迟时间。由于无界，插入元素永不阻塞；如果队列为空获取元素可以阻塞等待。
内部使用可重入锁 ReentrantLock 保证安全，使用优先级队列 PriorityQueue 实现延迟。该延迟阻塞队列可用于以下几种情况：

  • 缓存过期处理

  • 事件需要延迟处理

    SynchronousQueue

    同步阻塞队列，它不存储元素，用于线程和线程间的一对一通信。插入和获取元素必须成对出现才能同时操作，否则都一直等待对应的线程到来。

  • 公平模式下是先进先出的队列

  • 非公平模式下是后进先出栈 ```java AtomicInteger atomicNUm = new AtomicInteger(0); final SynchronousQueue queue = new SynchronousQueue<>();

new Thread(() -> { while (atomicNUm.get() < 100) { try { queue.put(atomicNUm.incrementAndGet()); System.out.println(“A run: “ + queue.take()); } catch (InterruptedException e) { e.printStackTrace(); } } }).start();

new Thread(() -> { while (atomicNUm.get() < 100) { try { System.out.println(“B run: “ + queue.take()); queue.put(atomicNUm.incrementAndGet()); } catch (InterruptedException e) { e.printStackTrace(); } } }).start(); ```

LinkedTransferQueueJDK 7

基于链表的、理论有界（Integer.MAX_VALUE）、先进先出（FIFO）的阻塞队列。（双同步队列）