【java并發編程】Lock & Condition 協調同步生產消費-有解無憂

一、協調生產/消費的需求

本文內容主要想向大家介紹一下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并發編程》中介紹了二者的區別，歡迎關注，

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
佇列已經空了，停止消費！

歡迎關注我的博客，更多精品知識合集

本文轉載注明出處（必須帶連接，不能只轉文字）：字母哥博客 - zimug.com

覺得對您有幫助的話，幫我點贊、分享！您的支持是我不竭的創作動力！，另外，筆者最近一段時間輸出了如下的精品內容，期待您的關注，

《kafka修煉之道》
《手摸手教你學Spring Boot2.0》
《Spring Security-JWT-OAuth2一本通》
《實戰前后端分離RBAC權限管理系統》
《實戰SpringCloud微服務從青銅到王者》

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/474625.html

標籤：Java

上一篇：在功能組件打字稿中，從googlemaps將坐標傳遞給父級

下一篇：多執行緒筆記（三）