教你一文读清线程通信之阻塞队列版（JUC编程）

简介

在java语言中实现线程通信经历了三个阶段：
- synchronized
- Lock
- BlockingQueue

可通过下面的链接认识第一、二阶段的线程通信
一文读清synchronized 和 Lock的优缺点
可通过下面的链接认识第三阶段的核心类BlockingQueue
一文读清阻塞队列

线程通信1-3版本 工作图：

下文将介绍第三阶段的线程通信：使用volatile/CAS/atomnicInteger/BlockingQueue/线程交互/原子引用 技术实现生产者消费者线程通信。不用手动加锁，也不用手动阻塞和唤醒线程。

• 问题背景：模拟一个生产者和一个消费者线程，当标志位flag为true时，两个同时工作，否则同时停止。
• 问题分析：如果标志位为true，生产者每隔一秒生产一个产品。如果标志位为true，消费者每隔一秒消费一个产品。初始阻塞队列内的元素为空，在生产和消费的过程中不手动加锁、也不手动阻塞和唤醒线程

资源类和线程工作任务：

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
// 共享资源类
class MyResource1 {
// 利用volatile的内存可见性，保证工作标志flag改变时，能够第一时间通知生产者、消费者线程
private volatile boolean flag = true;// 工作标志默认开启，进行生产+消费
// 多线程环境下尽量不使用++、--再进行手动控制，而改用原子变量替代（CAS自旋锁）
private AtomicInteger atomicInteger = new AtomicInteger();

// 面向抽象，可通用BlockingQueue的7种实现子类型阻塞队列
BlockingQueue<String> blockingQueue = null;
// 采用set注入和构造注入中的构造注入阻塞队列
public MyResource1(BlockingQueue<String> blockingQueue) {
this.blockingQueue = blockingQueue;
System.out.println(blockingQueue.getClass().getName());
}
// 生产者线程工作任务
public void produce() throws Exception {
String queueElement = null;
boolean retValue;
while (flag) {
// 获取生产者生产的产品，放到阻塞队列中
queueElement = atomicInteger.incrementAndGet() + "";
retValue = blockingQueue.offer(queueElement, 2L, TimeUnit.SECONDS);
if (retValue) {
System.out.println(Thread.currentThread().getName() + " 产品" + queueElement + "加入队列成功");
} else
System.out.println(Thread.currentThread().getName() + " 产品加入队列失败");
// 每隔一秒生产一个产品
TimeUnit.SECONDS.sleep(1L);
}
System.out.println(Thread.currentThread().getName() + " 被叫停了，不再生产");
}
// 消费者线程工作任务
public void consumer() throws Exception {
String retVal = null;
while (flag) {
// 超时等待2秒，获取产品
retVal = blockingQueue.poll(2L, TimeUnit.SECONDS);
if (retVal != null) {
System.out.println(Thread.currentThread().getName() + " 消费产品" + retVal + "成功");
} else {
flag = false;
System.out.println(Thread.currentThread().getName() + " 超过2秒没有获取到物品，停止消费");
return ;
}
}
}
public void stop() {
this.flag = false;
}
}

开始多个线程和控制线程结束：

public class ProdConsumer_BlockingQueueDemo {
public static void main(String[] args) {
// 创建可放置5个元素的阻塞队列
MyResource1 myResource = new MyResource1(new ArrayBlockingQueue<>(5));

new Thread(() -> {
try {
System.out.println(Thread.currentThread().getName() + " 开始启动");
myResource.produce();
} catch (Exception e) {e.printStackTrace();}
}, "生产者线程").start();

new Thread(() -> {
try {
System.out.println(Thread.currentThread().getName() + " 开始启动");
myResource.consumer();
} catch (Exception e) { e.printStackTrace();}
}, "消费者线程").start();

// 10秒钟之后，结束所有线程的任务
try { TimeUnit.SECONDS.sleep(5L); } catch (InterruptedException e) {e.printStackTrace();}
System.out.println();
System.out.println();
System.out.println();
System.out.println("10秒之后 开始叫停所有线程的工作");
myResource.stop();
}
}

运行如下：

从结果可知，用阻塞队列来达到线程之间的通信 不用通过手动加锁、手动阻塞和唤醒，阻塞队列自己就能够控制线程的执行。

• 点赞
• 收藏
• 分享
• 文章举报