直接上代码
class LinkedBlockingDequeDemo {
// 循环是否结束的开关
private static volatile boolean flag1 = true;
private static volatile boolean flag2 = true;
// 生成者生产的产品
private static AtomicInteger atomicInteger = new AtomicInteger(1);
//二个 双端阻塞队列
private LinkedBlockingDeque linkedBlockingDeque1;
private LinkedBlockingDeque linkedBlockingDeque2;
public LinkedBlockingDequeDemo(LinkedBlockingDeque linkedBlockingDeque1, LinkedBlockingDeque linkedBlockingDeque2) {
this.linkedBlockingDeque1 = linkedBlockingDeque1;
this.linkedBlockingDeque2 = linkedBlockingDeque2;
System.out.println(linkedBlockingDeque1.getClass().getName());
System.out.println(linkedBlockingDeque2.getClass().getName());
}
// 生产者
public void producer() throws InterruptedException {
String data = "";
while (flag1) {
data = atomicInteger.getAndIncrement() + "";//这是产品
//TimeUnit.SECONDS.sleep(1);// 1秒生成一个产品
if (Integer.valueOf(data) <= 10) {
//存到 linkedBlockingDeque1 队列中
linkedBlockingDeque1.put(data + "队列1");
System.out.println(Thread.currentThread().getName() + "添加元素到 阻塞队列 linkedBlockingDeque1,成功,元素为: " + data + "队列1");
} else {
//存到 linkedBlockingDeque2 队列中
linkedBlockingDeque2.put(data + "队列2");
System.out.println(Thread.currentThread().getName() + "添加元素到 阻塞队列 linkedBlockingDeque2,成功,元素为: " + data + "队列2");
}
}
}
// 消费者
public void consumer1() throws InterruptedException {
while (flag2) {
try {
TimeUnit.MILLISECONDS.sleep(3);
if (!linkedBlockingDeque1.isEmpty()) {
// 自己的队列不为空, 就从自己的队列中取数据消费, 从头开始消费数据
System.out.println(Thread.currentThread().getName() + "从自己队列 Deque1中 的头部消费了一个产品:" + linkedBlockingDeque1.takeFirst());
} else if (!linkedBlockingDeque2.isEmpty()) {
// 如果 另一个队列不为空, 就从他的尾开始消费数据
System.out.println(Thread.currentThread().getName() + "从别人队列 Deque2中 的尾部消费了一个产品:" + linkedBlockingDeque2.takeLast());
} else {
flag2 = false;//这里结束消费
}
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("------------------------------------------");
}
}
public void consumer2() throws InterruptedException {
while (flag2) {
try {
TimeUnit.MILLISECONDS.sleep(3);
if (!linkedBlockingDeque2.isEmpty()) {
// 自己的队列不为空, 就从自己的队列中取数据消费, 从头开始消费数据
System.out.println(Thread.currentThread().getName() + "从自己队列 Deque2中 的头部消费了一个产品:" + linkedBlockingDeque2.takeFirst());
} else if (!linkedBlockingDeque1.isEmpty()) {
// 如果 另一个队列不为空, 就从他的尾开始消费数据
System.out.println(Thread.currentThread().getName() + "从别人队列 Deque1中 的尾部消费了一个产品:" + linkedBlockingDeque1.takeLast());
} else {
flag2 = false;//这里结束消费
}
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("------------------------------------------");
}
}
// 停止生产的方法
public void stop() {
this.flag1 = false;
}
public static void main(String[] args) throws InterruptedException {
//工作窃取代码演示: 什么是工作窃取? 在生产者和消费者模型中,
// 每个消费者都对应一个阻塞队列(LinkedBlockingDeque),当消费者当前的队列中的任务消费完了后, 不会就此结束,
// 他会从另一个队列中获取对应的任务来进行消费,
// 这就是 LinkedBlockingDeque的好处,因为他是双端队列,可以从头和尾 来获取元素
/**
*
* 看到的效果就是: 消费者线程1, 从自己的队列Deque1 中, 从头部开始消费1-10, 之后,开始从 别人的队列Deque2中消费产品了
*
* 这里让生产者 1毫秒之后停止生产,然后消费者开始消费
*/
LinkedBlockingDeque<String> linkedBlockingDeque1 = new LinkedBlockingDeque<>();
LinkedBlockingDeque<String> linkedBlockingDeque2 = new LinkedBlockingDeque<>();
LinkedBlockingDequeDemo linkedBlockingDequeDemo = new LinkedBlockingDequeDemo(linkedBlockingDeque1, linkedBlockingDeque2);
new Thread(() -> {
try {
linkedBlockingDequeDemo.producer();
} catch (InterruptedException e) {
e.printStackTrace();
}
}, "生产者线程").start();
new Thread(() -> {
try {
linkedBlockingDequeDemo.consumer1();
} catch (InterruptedException e) {
e.printStackTrace();
}
}, "消费者线程1").start();
new Thread(() -> {
try {
linkedBlockingDequeDemo.consumer2();
} catch (InterruptedException e) {
e.printStackTrace();
}
}, "消费者线程2").start();
TimeUnit.MILLISECONDS.sleep(1);
linkedBlockingDequeDemo.stop();
}
}
运行结果 : 可以看到 线程1,在自己的阻塞队列中消费完之后, 并没有结束,从别人的队列中来 获取任务来执行
