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