隊列是一種 FIFO（先進先出）的數(shù)據(jù)結(jié)構(gòu)，本文要講的 BlockingQueue 也是一種隊列，而且強調(diào)了線程安全的特性。
BlockingQueue全稱：java.util.concurrent.BlockingQueue。它是是一個線程安全的隊列接口，多個線程能夠以并發(fā)的方式從隊列中插入數(shù)據(jù)，取出數(shù)據(jù)的同時不會出現(xiàn)線程安全的問題。
生產(chǎn)者和消費者例子
BlockingQueue 通常用于消費者線程向隊列存入數(shù)據(jù)，消費者線程從隊列中取出數(shù)據(jù)，具體如下
- 生產(chǎn)者線程不停的向隊列中插入數(shù)據(jù)，直到隊列滿了，生產(chǎn)者線程被阻塞
- 消費者線程不停的從隊列中取出數(shù)據(jù)，直到隊列為空，消費者線程被阻塞
(推薦教程：Java教程）
BlockingQueue 方法
BlockingQueue 提供 4 種不同類型的方法用于插入數(shù)，取出數(shù)據(jù)以及檢查數(shù)據(jù)，具體如下
- 操作失敗，拋出異常
- 無論成功/失敗，立即返回
true/false - 如果隊列為空/滿，阻塞當前線程
- 如果隊列為空/滿，阻塞當前線程并有超時機制插入
add(o)offer(o)put(o)offer(o, timeout, timeunit)取出remove(o)poll()take()poll(timeout, timeunit)檢查element()peek()
BlockingQueue 的具體實現(xiàn)類
BlockingQueue 只是一個接口，在實際開發(fā)中有如下的類實現(xiàn)了該接口。
- ArrayBlockingQueue
- DelayQueue
- LinkedBlockingQueue
- PriorityBlockingQueue
- SynchronousQueue
ArrayBlockingQueue 的使用
這里以 BlockingQueue 接口的具體實現(xiàn)類 ArrayBlockingQueue 舉例。通過 ArrayBlockingQueue實現(xiàn)一個消費者和生產(chǎn)者多線程模型。
核心內(nèi)容如下：
- 以
ArrayBlockingQueue作為生產(chǎn)者和消費者的數(shù)據(jù)容器 - 通過
ExecutorService啟動 3 個線程，2 兩個生產(chǎn)者，1 個消費者 - 指定數(shù)據(jù)總量
（推薦微課：Java微課）
生產(chǎn)者線程
ArrayBlockingQueueProducer
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
/**
* 生產(chǎn)者線程向容器存入指定總量的 任務
*
*/
public class ArrayBlockingQueueProducer implements Runnable {
private static final Logger logger = LoggerFactory.getLogger(ArrayBlockingQueueProducer.class);
// 容器
private ArrayBlockingQueue<String> queue;
// 生產(chǎn)指定的數(shù)量
private AtomicInteger numberOfElementsToProduce;
public ArrayBlockingQueueProducer(ArrayBlockingQueue<String> queue, AtomicInteger numberOfElementsToProduce) {
this.queue = queue;
this.numberOfElementsToProduce = numberOfElementsToProduce;
}
@Override
public void run() {
try {
while (numberOfElementsToProduce.get() > 0) {
try {
// 向隊列中存入任務
String task = String.format("task_%s", numberOfElementsToProduce.getAndUpdate(x -> x-1));
queue.put(task);
logger.info("thread {}, produce task {}", Thread.currentThread().getName(), task);
// 任務為0，生產(chǎn)者線程退出
if (numberOfElementsToProduce.get() == 0) {
break;
}
} catch (Exception e) {
e.printStackTrace();
}
}
} catch (Exception e) {
logger.error(this.getClass().getName().concat(". has error"), e);
}
}
}消費者線程
ArrayBlockingQueueConsumer
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
/**
* 消費者線程向容器 消費 指定總量的任務
*
*/
public class ArrayBlockingQueueConsumer implements Runnable {
private static final Logger logger = LoggerFactory.getLogger(ArrayBlockingQueueConsumer.class);
private ArrayBlockingQueue<String> queue;
private AtomicInteger numberOfElementsToProduce;
public ArrayBlockingQueueConsumer(ArrayBlockingQueue<String> queue, AtomicInteger numberOfElementsToProduce) {
this.queue = queue;
this.numberOfElementsToProduce = numberOfElementsToProduce;
}
@Override
public void run() {
try {
while (!queue.isEmpty() || numberOfElementsToProduce.get() >= 0) {
// 從隊列中獲取任務，并執(zhí)行任務
String task = queue.take();
logger.info("thread {} consume task {}", Thread.currentThread().getName(),task);
// 隊列中數(shù)據(jù)為空，消費者線程退出
if (queue.isEmpty()) {
break;
}
}
} catch (Exception e) {
logger.error(this.getClass().getName().concat(". has error"), e);
}
}
}測試 TestBlockingQueue
import com.ckjava.synchronizeds.appCache.WaitUtils;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
/**
* 1. 以 ArrayBlockingQueue 作為生產(chǎn)者和消費者的數(shù)據(jù)容器 <br>
* 2. 通過 ExecutorService 啟動 3 個線程，2 兩個生產(chǎn)者，1 個消費者 <br>
* 3. 指定數(shù)據(jù)總量
*/
public class TestBlockingQueue {
public static void main(String[] args) {
ArrayBlockingQueue<String> arrayBlockingQueue = new ArrayBlockingQueue<>(10);
/*BlockingQueue delayQueue = new DelayQueue();
BlockingQueue<String> linkedBlockingQueue = new LinkedBlockingQueue<>(10);
BlockingQueue<String> priorityBlockingQueue = new PriorityBlockingQueue<>(10);
BlockingQueue<String> synchronousQueue = new SynchronousQueue<>();*/
ExecutorService executorService = Executors.newFixedThreadPool(3);
// 最多生產(chǎn) 5 個數(shù)據(jù)
AtomicInteger numberOfElementsToProduce = new AtomicInteger(5);
// 2 個生產(chǎn)者線程
executorService.submit(new ArrayBlockingQueueProducer(arrayBlockingQueue, numberOfElementsToProduce));
executorService.submit(new ArrayBlockingQueueProducer(arrayBlockingQueue, numberOfElementsToProduce));
// 1 個消費者線程
executorService.submit(new ArrayBlockingQueueConsumer(arrayBlockingQueue, numberOfElementsToProduce));
executorService.shutdown();
WaitUtils.waitUntil(() -> executorService.isTerminated(), 1000L);
}
}輸出如下：
13:54:17.884 [pool-1-thread-3] INFO c.c.b.ArrayBlockingQueueConsumer - thread pool-1-thread-3 consume task task_5
13:54:17.884 [pool-1-thread-1] INFO c.c.b.ArrayBlockingQueueProducer - thread pool-1-thread-1, produce task task_5
13:54:17.884 [pool-1-thread-2] INFO c.c.b.ArrayBlockingQueueProducer - thread pool-1-thread-2, produce task task_4
13:54:17.887 [pool-1-thread-3] INFO c.c.b.ArrayBlockingQueueConsumer - thread pool-1-thread-3 consume task task_4
13:54:17.887 [pool-1-thread-2] INFO c.c.b.ArrayBlockingQueueProducer - thread pool-1-thread-2, produce task task_2
13:54:17.887 [pool-1-thread-1] INFO c.c.b.ArrayBlockingQueueProducer - thread pool-1-thread-1, produce task task_3
13:54:17.887 [pool-1-thread-3] INFO c.c.b.ArrayBlockingQueueConsumer - thread pool-1-thread-3 consume task task_3
13:54:17.887 [pool-1-thread-2] INFO c.c.b.ArrayBlockingQueueProducer - thread pool-1-thread-2, produce task task_1
13:54:17.887 [pool-1-thread-3] INFO c.c.b.ArrayBlockingQueueConsumer - thread pool-1-thread-3 consume task task_2
13:54:17.887 [pool-1-thread-3] INFO c.c.b.ArrayBlockingQueueConsumer - thread pool-1-thread-3 consume task task_1（推薦內(nèi)容：Java面試基礎(chǔ)題）
以上就是關(guān)于JUC 之 BlockingQueue 接口以及 ArrayBlockingQueue 實現(xiàn)類的相關(guān)介紹了，希望對大家有所幫助。
