WHY
解决并发问题。
WHAT
通过一个容器解决生产者和消费者强耦合问题，否则消费者只能排队等生产者串行解决。
生产者和消费者之间不直接通信，而是通过一个阻塞队列来通信。阻塞队列就是一个缓冲区。
HOW
一、阻塞队列实现
阻塞队列是解决生产者消费者的关键。
-- 两种自定义阻塞队列方式
-- jdk1.5版本以后java.util.concurrent提供了阻塞队列类
阻塞队列接口：
package com.bytebeats.concurrent.queue;
/* 阻塞队列接口*/
public interface IBlockingQueue{
public Object take();
public void put(Object data);
}
方式1.推荐使用jdk提供的阻塞队列类
jdk1.5版本后新增的java.util.concurrent包新增了java.util.concurrent.BlockingQueue接口，并且提供了以下几种实现：
- java.util.concurrent.ArrayBlockingQueue
- java.util.concurrent.LinkedBlockingQueue
- java.util.concurrent.SynchronousQueue
- java.util.concurrent.PriorityBlockingQueue
方式2.用Object.wai()/notifyAll() 实现自定义阻塞队列
package blocking.queue;
import java.util.LinkedList;
public class TraditionalBlockingQueue implements IBlockingQueue{
/*这个表示该阻塞队列长度，最多能接受的数据数量*/
private int queueSize;
private final LinkedList<Object> queueList = new LinkedList<Object>();
private final Object lock = new Object();
public TraditionalBlockingQueue(int queueSize) {
if (queueSize < 1) {
throw new IllegalArgumentException("queueSize must be positive number");
}
this.queueSize = queueSize;
}
@Override
public void put(Object data) throws InterruptedException {
synchronized (lock) {
/*这里必须使用while，而不使用if，这是一种自旋锁*/
while (queueList.size() >= queueSize) {
System.out.println("data is full");
lock.wait();
}
queueList.add(data);
lock.notifyAll();
}
}
@Override
public Object take() throws InterruptedException {
synchronized (lock) {
while(queueList.size() < 1) {
System.out.println("data is empty");
lock.wait();
}
Object data = queueList.removeFirst();
/*此处必须使用notifyAll，如果使用notify在某些情况下会导致生产者-消费者同时处于阻塞情况*/
lock.notifyAll();
return data;
}
}
}
方式3.用Lock和Condition实现自定义阻塞队列
暂不考虑
二、生产者-消费者实现
生产者
package blocking.queue;
public class Producer implements Runnable {
private IBlockingQueue queue;
private int consumerNum;
private int dataSize;
public Producer(IBlockingQueue queue, int consumerNum, int dataSize) {
this.queue = queue;
this.dataSize = dataSize;
this.consumerNum = consumerNum;
}
@Override
public void run() {
/*我这个Producer想往BlockingQueue里面放多少数据*/
for (int i = 0; i < dataSize; i++) {
try {
queue.put("data_" + i);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
/*为每个Consumer创建一个结束标志位，是为了保证最后每个Consumer线程都能退出*/
for (int i = 0; i < consumerNum; i++) {
try {
queue.put("#");
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
System.out.println("Producer over");
}
}
消费者
package blocking.queue;
import java.util.concurrent.TimeUnit;
public class Consumer implements Runnable {
private IBlockingQueue queue;
public Consumer(IBlockingQueue queue) {
this.queue = queue;
}
@Override
public void run() {
while (true) {
String data = null;
try {
data = (String) queue.take();
System.out.println("Consumer " + Thread.currentThread().getName() + " consume:" + data);
/*consumer领到#就退出，而#和consumer个数是一致的，所以每个consumer都能领到，然后安全退出。*/
if ("#".equals(data)) {
break;
}
TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
System.out.println("Consumer over");
}
}
测试代码
package blocking.queue;
public class QueueTest {
public static void main(String[] args) {
/*2表示该queue最多能放2个数据*/
new QueueTest().testRun(new TraditionalBlockingQueue(2));
}
private void testRun(TraditionalBlockingQueue queue) {
/*9表示有9个Consumer，是为了告诉Producer要设置多少个结束标志位（这是个比较土的通知Consumer线程结束的方式）
20表示Producer想创建多少个数据（和Queue的区别是这20个是等Queue Not Full才分别放进去）*/
Thread producer = new Thread(new Producer(queue, 9, 20));
producer.start();
Thread consumer1 = new Thread(new Consumer(queue));
consumer1.start();
Thread consumer2 = new Thread(new Consumer(queue));
consumer2.start();
Thread consumer3 = new Thread(new Consumer(queue));
consumer3.start();
Thread consumer4 = new Thread(new Consumer(queue));
consumer4.start();
Thread consumer5 = new Thread(new Consumer(queue));
consumer5.start();
Thread consumer6 = new Thread(new Consumer(queue));
consumer6.start();
Thread consumer7 = new Thread(new Consumer(queue));
consumer7.start();
Thread consumer8 = new Thread(new Consumer(queue));
consumer8.start();
Thread consumer9 = new Thread(new Consumer(queue));
consumer9.start();
}
}
测试结果
data is full
data is full
Consumer Thread-1 consume:data_0
Consumer Thread-2 consume:data_1
data is full
Consumer Thread-3 consume:data_2
data is full
Consumer Thread-4 consume:data_3
data is full
Consumer Thread-5 consume:data_4
data is full
Consumer Thread-6 consume:data_5
data is full
Consumer Thread-7 consume:data_6
data is full
Consumer Thread-8 consume:data_7
data is full
Consumer Thread-9 consume:data_8
data is full
data is full
Consumer Thread-4 consume:data_9
Consumer Thread-5 consume:#
Consumer over
data is full
Consumer Thread-6 consume:#
Consumer over
Consumer Thread-1 consume:#
Consumer over
Consumer Thread-2 consume:#
Consumer over
Consumer Thread-7 consume:#
Consumer over
data is empty
data is empty
data is full
Consumer Thread-3 consume:#
Consumer over
Consumer Thread-9 consume:#
Consumer over
Producer over
Consumer Thread-8 consume:#
Consumer over
Consumer Thread-4 consume:#
Consumer over
测试结果：Producer执行put的时候会校验queue是不是满了，满了要等不满再put，Consumer执行take的时候会校验queue是不是空的，空了要等不空再take。
每个consumer都会在遇到#后结束，并且consumer去抢夺producer产生的data。
- Tips:如果想用jdk自带的阻塞队列类，把IBlockingQueue改成ArrayBlockingQueue之类就可以
网友评论