掌握系列之并发编程-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()