掌握系列之并发编程-4.线程通信

掌握高并发、高可用架构

第二课 并发编程

从本课开始学习并发编程的内容。主要介绍并发编程的基础知识、锁、内存模型、线程池、各种并发容器的使用。

第四节 线程通信

并发编程 线程通信 AQS Condition Lock

本节学习线程间的通信，并手写缓存队列。

线程通信的实现方式

有两种：

1. 关键字synchronized结合wait()、notify()、notifyAll()来实现
2. 使用Lock并且结合Condition来实现

本节内容主要讲解Condition。

Condition

是个接口。实现类是ConditionObject，AQS的一个内部类

public interface Condition {    void await() throws InterruptedException;    void awaitUnInterruptibly();    long awaitNanos(long nanosTimeout) throws InterruptedException;    boolean await(long time, TimeUnit unit) throws InterruptedException;    boolean awaitUntil(Date deadline) throws InterruptedException;    void signal();    void signalAll();}

• await()，会使当前线程等待，并且释放锁；当其他线程执行signal()或signalAll()时，线程会重新获取锁并继续执行；或者当线程被中断时，会使线程跳出等待。该方法和Object.wait()功能相似
• awaitUnInterruptibly()，与await类似，但是不会响应中断，即使是在等待状态
• signal()，用于唤醒一个等待的线程。相对的signalAll()方法可以唤醒所有等待的线程。和Object.notify()功能类似

condition.await()必须在lock和unlock之间使用

使用lock.newCondition()来获取Condition

虚假等待和虚假唤醒

当执行await()或signal()时，线程不一定立即响应，此时会出现虚假等待和虚假唤醒。这是对基础平台语义的让步。若使用"if (!条件)"来做判断的话会有问题，所以一般使用 "while(!条件)"来防止这种情况

不用IF，使用WHILE

if (!条件) {    condition.await();}

while (!条件) {    condition.await();}

缓冲队列的实现

上代码（生产者消费者模式）

import java.util.concurrent.locks.Condition;import java.util.concurrent.locks.Lock;import java.util.concurrent.locks.ReentrantLock;/** * @Description: 缓冲队列 * @Author: lsw * @Version: 1.0 */public class BoundedBuffer {    final Lock lock = new ReentrantLock(); // 锁对象    final Condition notFull = lock.newCondition(); // 写条件    final Condition notEmpty = lock.newCondition(); // 读条件    final Object[] items = new Object[100]; // 容器    int putIdx, // 写索引        takeIdx, // 读索引        count; // 当前数量    public void put(Object it) throws InterruptedException {        lock.lock();        try {            while (count == items.length) {                notFull.await(); // 当容器存满时，使写线程等待            }            // 正常情况            items[putIdx] = it;            putIdx++;            // 存到尾部，则再从头开始            if (putIdx == items.length) {                putIdx = 0;            }            count++;            // 存入对象，就通知读线程进行读取            notEmpty.signal();        } finally {            lock.unlock();        }    }    public Object take() throws InterruptedException {        lock.lock();        try {            while (count == 0) {                notEmpty.await(); // 当容器空，则使读线程等待            }            // 正常情况            Object it = items[takeIdx];            takeIdx++;            // 如果读到尾部，则从头开始            if (takeIdx == items.length) {                takeIdx = 0;            }            count--;            // 唤醒写线程            notFull.signal();            return it;        } finally {            lock.unlock();        }    }}

通过针对同一个Lock创建多个Condition，可以非常灵活的控制各个线程执行或者等待。这就是Condition的强大之处

LockSupport工具类

使用Lock实现加锁解锁以及Condition对线程进行状态操作时，底层都会用到LockSupport.part()或者LockSupport.unpark()。下面我们来研究下这个工具类。

public class LockSupport {    static void park() {}    static void park(Object blocker) {}    static void parkNanos(long nanos) {}    static void parkNanos(Object blocker, long nanos) {}    static void parkUntil(long deadline) {}    static void park(Object blocker, long deadline) {}    static void unpark(Thread t) {}}

park()方法的作用是使当前线程进入等待WAITING队列，直到调用unpark()或者响应中断

parkNanos()方法是指使当前线程进入等待队列，且等待时间不可超过指定的时长

parkUntil()方法是指使当前线程进入等待队列，直到某个截止时间退出等待

参数blocker是可用于记录导致线程等待的对象，方便排查问题

unpark()用于唤醒指定的线程

这些功能的底层是调用的Unsafe本地类库的UNSAFE.park()和UNSAFE.unpark()