这篇文章主要介绍“BlockingQueue接口及ArrayBlockingQueue实现类的方法”的相关知识，小编通过实际案例向大家展示操作过程，操作方法简单快捷，实用性强，希望这篇“BlockingQueue接口及ArrayBlockingQueue实现类的方法”文章能帮助大家解决问题。
队列是一种 FIFO（先进先出）的数据结构，本文要讲的 BlockingQueue 也是一种队列，而且强调了线程安全的特性。
BlockingQueue全称：java.util.concurrent.BlockingQueue。它是是一个线程安全的队列接口，多个线程能够以并发的方式从队列中插入数据，取出数据的同时不会出现线程安全的问题。
BlockingQueue 通常用于消费者线程向队列存入数据，消费者线程从队列中取出数据，具体如下
生产者线程不停的向队列中插入数据，直到队列满了，生产者线程被阻塞
消费者线程不停的从队列中取出数据，直到队列为空，消费者线程被阻塞
(推荐教程：Java教程）
BlockingQueue 提供 4 种不同类型的方法用于插入数，取出数据以及检查数据，具体如下
操作失败，抛出异常
无论成功/失败，立即返回 true/false
如果队列为空/满，阻塞当前线程
如果队列为空/满，阻塞当前线程并有超时机制插入add(o) offer(o) put(o) offer(o, timeout, timeunit)取出remove(o) poll() take() poll(timeout, timeunit)检查element() peek()
BlockingQueue 只是一个接口，在实际开发中有如下的类实现了该接口。
ArrayBlockingQueue
DelayQueue
LinkedBlockingQueue
PriorityBlockingQueue
SynchronousQueue
这里以 BlockingQueue 接口的具体实现类 ArrayBlockingQueue 举例。通过 ArrayBlockingQueue实现一个消费者和生产者多线程模型。
核心内容如下：
以 ArrayBlockingQueue 作为生产者和消费者的数据容器
通过 ExecutorService 启动 3 个线程，2 两个生产者，1 个消费者
指定数据总量
ArrayBlockingQueueProducer
计算
安全
数据库
网络和加速
企业服务
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
/**
* 生产者线程向容器存入指定总量的 任务
*
*/
public class ArrayBlockingQueueProducer implements Runnable {
private static final Logger logger = LoggerFactory.getLogger(ArrayBlockingQueueProducer.class);
// 容器
private ArrayBlockingQueue<String> queue;
// 生产指定的数量
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，生产者线程退出
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) {
// 从队列中获取任务，并执行任务
String task = queue.take();
logger.info("thread {} consume task {}", Thread.currentThread().getName(),task);
// 队列中数据为空，消费者线程退出
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 作为生产者和消费者的数据容器 <br>
* 2. 通过 ExecutorService 启动 3 个线程，2 两个生产者，1 个消费者 <br>
* 3. 指定数据总量
*/
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);
// 最多生产 5 个数据
AtomicInteger numberOfElementsToProduce = new AtomicInteger(5);
// 2 个生产者线程
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关于“BlockingQueue接口及ArrayBlockingQueue实现类的方法”的内容就介绍到这里了，感谢大家的阅读。如果想了解更多行业相关的知识，可以关注亿速云行业资讯频道，小编每天都会为大家更新不同的知识点。
亿速云「云服务器」，即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘，价格低至29元/月。点击查看>>
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
