簡單理解阻塞隊列(BlockingQueue)中的take/put方法以及Condition存在的作用

Condition：可以理解成一把鎖的一個鑰匙，它既可以解鎖(通知放行)，又可以加鎖(阻塞)

notFull：當隊列元素滿了時，阻塞生產，當隊列元素存在元素但是沒有滿時，去通知消費

notEmpty：當隊列中不存在元素時，阻塞消費，當隊列元素存在元素時，去通知生產

while (count >= datas.length) {...}

while (count <= 0) {...}

兩個while循環判斷，而不用if，是因為線程不安全，

導致多線程場景下每個線程獲取到的循環條件count的值存在差異，

導致代碼執行異常，用while可以使當前線程重新刷新判斷條件count的值。

用處：

ThreadPoolExecutor中使用到了阻塞隊列，阻塞隊列中又使用到了ReentrantLock，而ReentrantLock基于AQS。

package com.zhuyz.foundation.juc;

import java.util.concurrent.TimeUnit;

import java.util.concurrent.locks.Condition;

import java.util.concurrent.locks.ReentrantLock;

public class MyArrayBlockingQueue {

// 數組元素

private T[] datas;

// 實際的元素個數(也有索引下標的作用)

private volatile int count;

private final ReentrantLock putLock = new ReentrantLock();

private final ReentrantLock takeLock = new ReentrantLock();

// 通知消費，暫停生產【當數組full時await(put時)，當數組notFull時signal(take時)】

private Condition notFull = putLock.newCondition();

// 通知生產，暫停消費【當數組empty時await(take時)，當數組notEmpty時signal(put時)】

private Condition notEmpty = takeLock.newCondition();

public MyArrayBlockingQueue(int cap) {

this.datas = (T[]) new Object[cap];

}

private void put(T t) {

final ReentrantLock putLock = this.putLock;

putLock.lock();

try {

// 線程不安全，需要循環判斷，插入值之前判斷一下數組長度是否達到最大長度

while (count >= datas.length) {

System.out.println("datas is full, please waiting take");

notFull.await();

}

datas[count++] = t;

System.out.println("put: " + t);

} catch (Exception e) {

System.out.println("異常" + e);

} finally {

putLock.unlock();

}

// 通知獲取元素的線程繼續執行(take_thread)

final ReentrantLock takeLock = this.takeLock;

takeLock.lock();

try {

notEmpty.signal();

} finally {

takeLock.unlock();

}

}

private T take() {

final ReentrantLock takeLock = this.takeLock;

takeLock.lock();

T t = null;

try {

// 線程不安全，需要循環判斷，插入值之前判斷一下數組中元素個數是否為0

while (count <= 0) {

System.out.println("datas is empty, please waiting put");

notEmpty.await();

}

// 獲取元素

t = datas[--count];

System.out.println("take: " + t);

} catch (Exception e) {

System.out.println("異常" + e);

} finally {

takeLock.unlock();

}

final ReentrantLock putLock = this.putLock;

putLock.lock();

try {

// 通知插入元素的線程繼續執行(put_thread)

notFull.signal();

} finally {

putLock.unlock();

}

return t;

}

public static void main(String[] args) throws InterruptedException {

MyArrayBlockingQueue myArrayBlockingQueue = new MyArrayBlockingQueue<>(5);

for (int i = 0; i < 10; i++) {

int finalI = i;

new Thread(() -> myArrayBlockingQueue.put(finalI)).start();

}

TimeUnit.SECONDS.sleep(5L);

for (int i = 0; i < 5; i++) {

new Thread(() -> myArrayBlockingQueue.take()).start();

}

}

}

結果如下：

從結果中可以看出，先put了5個元素，然后另外五個元素被阻塞住了，沒有進去

take消費5個之后，另外五個被阻塞的元素就被put進去了

put: 0

put: 1

put: 2

put: 3

put: 4

datas is full, please waiting take

datas is full, please waiting take

datas is full, please waiting take

datas is full, please waiting take

datas is full, please waiting take

take: 4

put: 5

take: 5

take: 3

put: 6

put: 7

take: 7

put: 8

take: 8

put: 9

本文地址：https://blog.csdn.net/qq_43128724/article/details/110438987

如您對本文有疑問或者有任何想說的，請點擊進行留言回復，萬千網友為您解惑！

總結

以上是生活随笔為你收集整理的java阻塞队列作用_简单理解阻塞队列(BlockingQueue)中的take/put方法以及Condition存在的作用...的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。