[TOC] ## 一、協調生產/消費的需求 本文內容主要想向大家介紹一下Lock結合Condition的使用方法，**為了更好的理解Lock鎖與Condition鎖信號，我們來手寫一個ArrayBlockingQueue**。 JDK實際上已經有這個類，基于Lock鎖與Condition鎖信號實現的，當然JDK實現代碼很復雜包含了更嚴謹的邏輯校驗，以及從性能優化的角度做了更多的工作。本文中我們只是來簡單實現一下其核心邏輯： * ArrayBlockingQueue初始化構造時指定容量上限最大值 * 提供put方法，當達到Queue隊列容量上限最大值，阻塞生產數據的線程。 * put方法生產數據之后，隊列肯定是不為空，通知消費者線程進行消費。 * 提供take方法，當Queue隊列容量為0時候，阻塞消費數據的線程。 * take方法執行之后，隊列肯定不是滿的，通知生產者線程進行生產。 * 一條數據只能被take一次，take之后數據從queue中刪除 相信實現完成上面的邏輯之后，java并發編程之Lock鎖與Condition鎖信號，你肯定是掌握了！**其實這個邏輯基本上就是kafka生產者客戶端緩沖隊列，批量進行數據發送的實現邏輯**。區別是take方法一次取出緩沖區所有數據，本文take方法一次取出一條數據。 ## 二、構造方法 構造隊列的方法很簡單，使用一個List作為數據存儲隊列，并指定其容量。到此我們還沒有實現容量判斷，以及阻塞線程的功能。 ~~~ //類成員變量-存儲數據的隊列 private List<Object> queue; //類成員變量-存儲隊列的容量上限 private int queueSize; public MyBlockingQueue(int queueSize) { this.queueSize = queueSize; queue = new ArrayList<>(queueSize);//存儲消息的集合 } ~~~ ## 三、Lock& Condition邏輯設計 首先我們要有一把鎖，保證數據put與take操作的同步性，即：一條數據只能被take一次，take之后數據從queue中刪除；以及創建Condition邏輯都需要Lock鎖。學過java基礎并發編程的同學，可以把Lock鎖理解為Synchronized 同步代碼塊功能是一樣的。我寫過一個專欄[《java并發編程》](https://mp.weixin.qq.com/mp/appmsgalbum?__biz=MzU0NDU5MTk1MQ==&action=getalbum&album_id=1576334194996232194#wechat_redirect)中介紹了二者的區別，歡迎關注。 ~~~ private Lock lock = new ReentrantLock();//鎖 ~~~ Condition邏輯大家可以理解為傳統JDK多線程編程中的wait與notify，但是Condition的語義更容易被理解。如下文代碼所示： ~~~ private Condition notFull = lock.newCondition(); //隊列不為滿 notFull.signal(); //通知生產者隊列不為滿，可以繼續生產數據(通常在消費者拿走數據之后，調用) notFull.await(); //隊列已滿，阻塞生產線程（await對condition邏輯取反） ~~~ ~~~ private Condition notEmpty = lock.newCondition(); //隊列不為空 notEmpty.signal(); //通知消費者線程隊列不為空，可以繼續消費數據(通常在生產者生產數據之后，調用) notEmpty.await(); //隊列已經空了，阻塞消費線程（await對condition邏輯取反） ~~~ 大家在使用Lock& Condition進行線程同步協調的時候，一定像我一樣先把condition的邏輯語義設計好 * 將當xxxx時候的表達，設計為condition。 * 當情況滿足condition的時候發出信號signal()通知其他線程； * 當情況與condtion正好相反的的時候，使用await阻塞當前線程。 ## 四、put放入數據 其實最重要的就是完成Lock& Condition邏輯設計，剩下的就是填空了，模板如下  通過while循環判斷隊列當前容量是否達到容量上限，如果達到上限就表示隊列滿了。隊列滿了（**notFull取反使用await**），await阻塞生產線程向隊列中繼續放入數據。在這里，有小伙伴曾經問過我一個奇葩的問題：**多線程持有同一個lock鎖，你怎么知道阻塞的是生產線程，而不是消費線程呢？** 答：一個線程是生產線程還是消費線程，取決于它的動作（調用什么方法），并沒有一個標簽給它定義死，調用put方法放入數據的就是生產數據的線程。**while/await組合是標準寫法，請不要隨意創新改成if，否則你會遇到很多詭異的bug。** ~~~ //隊列滿了，await阻塞生產線程 while (queue.size() >= queueSize) { System.out.println(Thread.currentThread().getName() + "等待,因為隊列滿了" ); notFull.await(); } ~~~ 向隊列中添加一條數據，此時我們可以確定隊列是notEmpty，所以使用notEmpty.signal()向生產者發送信號。這里問題又來了：**多線程持有同一個lock鎖，你怎么知道通知的是消費者線程，而不是生產者線程呢？** **答案是我確實不知道**，所以在上文中的`while (queue.size() >= queueSize)`采用的是while，而不是if。即使生產者線程被喚醒了，while判斷也會把它await攔住。 ~~~ //向隊列添加一條消息，同時通知消費者有新消息了 queue.add(message); System.out.println(Thread.currentThread().getName() + "生產" + message ); notEmpty.signal();//通知消費者線程 ~~~ ## 五、take消費數據 take從隊列中取出數據，取出數據之后，隊列肯定是notFull ，所以發出notFull.signal信號。當隊列空了（notEmpty使用await取反)，await同時阻塞消費者線程。 ~~~ public Object take() throws InterruptedException { Object retVal = null; lock.lock();//操作隊列先加鎖 try { //隊列空了，通知生產線程，消費線程阻塞 while (queue.size() == 0) { System.out.println("隊列已經空了，停止消費！"); notEmpty.await(); } //隊列刪除一條消息，同時通知生產者隊列有位置了 retVal = queue.get(0); queue.remove(0); notFull.signal(); //同時通知生產者隊列 } finally { lock.unlock(); } return retVal; } ~~~ 我相信有了上面put方法的基礎，理解take方法中的代碼，就非常容易了，這里我就不做過多的說明了。 ## 六、生產消費測試 ~~~ public static void main(String[] args) { //為了方便查看測試結果，我們的隊列容量設置小一些 MyBlockingQueue queue = new MyBlockingQueue(2); //生產者線程 new Thread(()->{ for(int i = 0;i < 5;i++){ try { queue.put("msg" + i); //放入5條數據 } catch (InterruptedException e) { e.printStackTrace(); } } }).start(); //消費者線程 new Thread(()->{ while(true){ //一直消費 try { System.out.println(Thread.currentThread().getName() + "消費數據" + queue.take()); } catch (InterruptedException e) { e.printStackTrace(); } } }).start(); } ~~~ 輸出結果如下，滿足我們的需求。隊列滿了，生產者線程Thread-0等待；生產消費互相協調通知，最終數據消費完成，隊列空了，消費者線程阻塞。 ~~~ Thread-0生產msg0 Thread-0生產msg1 Thread-0等待,因為隊列滿了 Thread-1消費數據msg0 Thread-0生產msg2 Thread-0等待,因為隊列滿了 Thread-1消費數據msg1 Thread-0生產msg3 Thread-0等待,因為隊列滿了 Thread-1消費數據msg2 Thread-0生產msg4 Thread-1消費數據msg3 Thread-1消費數據msg4 隊列已經空了，停止消費！ ~~~