生产者和消费者模式的来源
生产者指的是生产数据的线程，消费者是消费数据的线程。在多线程开发中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这种生产能力和消费能力不均衡的问题，便有了生产者和消费者模式。
什么是生产者和消费者模式
生产者和消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通信，而是通过阻塞队列来进行通信，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列相当于一个缓冲区，平衡了生产者和消费者的处理能力。
使用场景
线程池
如何实现生产者和消费者模式
场景
生产者和消费者共用同一个存储空间，生产者Producer生产消息，放到存储空间中，消费者Consumer消费消息。如果存储空间已经空了，那么消费者阻塞，如果存储空间已经满了，那么生产者阻塞。
关键点
- 1、多个线程操作共享数据区域，为了避免并发问题，需要在共享数据区域上锁。
- 2、当一个线程已经在操作共享数据区域的时候，那么其他线程应该阻塞，这样的话，当前线程操作完毕后，需要通知其他的线程可以竞争锁了。
- 3、对于队列任务，我们为了保证公平性，采用LinkedList实现后进先出的链。
实现方式1，wait/notify实现生产者和消费者模式
- 测试
public class DemoTest {
public static void main(String[] args) throws InterruptedException {
LinkedList queue = new LinkedList();
String lock = "lock";
Integer maxQueueSize = 20;
Producer producer0 = new Producer(queue, maxQueueSize, lock);
producer0.produceMsg("消息" + 0);
Thread.sleep(3000);
Consumer consumer0 = new Consumer(queue, lock);
consumer0.consumeMsg();
for (int i = 1 ; i < 10 ; i++) {
Thread.sleep(3000);
Producer producer = new Producer(queue, maxQueueSize, lock);
producer.produceMsg("消息" + i);
}
for (int i = 1 ; i < 10 ; i++) {
Thread.sleep(3000);
Consumer consumer = new Consumer(queue, lock);
consumer.consumeMsg();
}
}
}
- 生产者
public class Producer {
private String lock;
private LinkedList<String> taskQueue;
private Integer maxQueueSize;
public Producer(LinkedList<String> taskQueue, Integer maxQueueSize, String lock) {
this.taskQueue = taskQueue;
this.maxQueueSize = maxQueueSize;
this.lock = lock;
}
public void produceMsg(String msg) {
InnerProducerWorker worker = new InnerProducerWorker(msg);
worker.start();
}
private void addTask(String msg, String state) {
while (taskQueue.size() >= maxQueueSize) {
try {
lock.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
if (taskQueue.size() < maxQueueSize) {
taskQueue.addLast(msg);
System.out.println("================= producer : " + msg);
state = ThreadState.STOPING.name();
}
lock.notifyAll();
}
private final class InnerProducerWorker {
private String state;
private Thread t;
public InnerProducerWorker(String msg) {
this.t = new Thread(() -> {
this.state = ThreadState.RUNNING.name();
synchronized (lock) {
addTask(msg, state);
if (this.state.equals(ThreadState.STOPING.name())
&& !t.isInterrupted()) {
t.interrupt();
}
}
});
}
public void start() {
t.start();
this.state = ThreadState.STARTED.name();
}
}
}
- 消费者
public class Consumer {
private String lock;
private LinkedList<String> taskQueue;
public Consumer(LinkedList<String> taskQueue, String lock) {
this.taskQueue = taskQueue;
this.lock = lock;
}
public void consumeMsg() {
InnerConsumerWorker worker = new InnerConsumerWorker();
worker.start();
}
public String getMsg(String state) {
while (taskQueue.size() == 0) {
try {
lock.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
if (taskQueue.size() > 0) {
String msg = taskQueue.pollFirst();
state = ThreadState.STOPING.name();
lock.notifyAll();
return msg;
}
return null;
}
private final class InnerConsumerWorker {
private String state;
private Thread t;
public InnerConsumerWorker() {
this.t = new Thread(() -> {
this.state = ThreadState.RUNNING.name();
synchronized (lock) {
String msg = getMsg(state);
System.out.println("================= consumer : " + msg);
if (this.state.equals(ThreadState.STOPING.name())
&& !t.isInterrupted()) {
t.interrupt();
}
}
});
}
public void start() {
t.start();
this.state = ThreadState.STARTED.name();
}
}
}
实现方式2，基于ReentrantLock实现生产者和消费者模式
- 测试
public class DemoTest {
public static void main(String[] args) throws InterruptedException {
LinkedList<String> queue = new LinkedList<>();
ReentrantLock reentrantLock = new ReentrantLock();
Condition condition = reentrantLock.newCondition();
Integer maxQueueSize = 8;
Producer producer0 = new Producer(queue, maxQueueSize, reentrantLock, condition);
producer0.produceMsg("消息" + 0);
Thread.sleep(3000);
Consumer consumer0 = new Consumer(queue, reentrantLock, condition);
consumer0.consumeMsg();
for (int i = 1 ; i < 10 ; i++) {
Thread.sleep(3000);
Producer producer = new Producer(queue, maxQueueSize, reentrantLock, condition);
producer.produceMsg("消息" + i);
}
for (int i = 1 ; i < 10 ; i++) {
Thread.sleep(3000);
Consumer consumer = new Consumer(queue, reentrantLock, condition);
consumer.consumeMsg();
}
}
}
/**
* 线程状态
*/
public enum ThreadState {
STARTED("0", "STARTED"),
RUNNING("1", "RUNNING"),
STOPPING("2", "STOPPING");
private String code;
private String name;
ThreadState(String code, String name) {
this.code = code;
this.name = name;
}
}
- 生产者
public class Producer {
private LinkedList<String> taskQueue;
private ReentrantLock reentrantLock;
private Integer maxQueueSize;
private Condition condition;
public Producer(LinkedList<String> taskQueue, Integer maxQueueSize,
ReentrantLock reentrantLock, Condition condition) {
this.taskQueue = taskQueue;
this.maxQueueSize = maxQueueSize;
this.reentrantLock = reentrantLock;
this.condition = condition;
}
public void produceMsg(String msg) {
InnerProducerWorker worker = new InnerProducerWorker(msg);
worker.start();
}
private void addTask(String msg, String state) {
while (taskQueue.size() >= maxQueueSize) {
try {
condition.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
taskQueue.addLast(msg);
System.out.println("================= producer : " + msg);
state = ThreadState.STOPPING.name();
condition.signalAll();
}
private final class InnerProducerWorker {
private String state;
private Thread t;
public InnerProducerWorker(String msg) {
this.t = new Thread(() -> {
reentrantLock.lock();
this.state = ThreadState.RUNNING.name();
addTask(msg, state);
if (this.state.equals(ThreadState.STOPPING.name())
&& !t.isInterrupted()) {
t.interrupt();
}
reentrantLock.unlock();
});
}
public void start() {
t.start();
this.state = ThreadState.STARTED.name();
}
}
}
- 消费者
public class Consumer {
private final BlockingQueue<String> blockingQueue;
public Consumer(BlockingQueue<String> blockingQueue) {
this.blockingQueue = blockingQueue;
}
public void consumeMsg() {
InnerConsumerWorker worker = new InnerConsumerWorker();
worker.start();
}
public String getMsg(String state) {
try {
String msg = blockingQueue.take();
state = ThreadState.STOPPING.name();
return msg;
} catch (InterruptedException e) {
e.printStackTrace();
}
return null;
}
private final class InnerConsumerWorker {
private String state;
private Thread t;
public InnerConsumerWorker() {
this.t = new Thread(() -> {
this.state = ThreadState.RUNNING.name();
String msg = getMsg(state);
System.out.println("================= consumer : " + msg);
if (this.state.equals(ThreadState.STOPPING.name())
&& !t.isInterrupted()) {
t.interrupt();
}
});
}
public void start() {
t.start();
this.state = ThreadState.STARTED.name();
}
}
}
实现方式3，基于BlockingQueue实现生产者和消费者模式
- 测试
public class DemoTest {
public static void main(String[] args) throws InterruptedException {
BlockingQueue<String> queue = new ArrayBlockingQueue<>(10);
Producer producer0 = new Producer(queue);
producer0.produceMsg("消息" + 0);
Thread.sleep(3000);
Consumer consumer0 = new Consumer(queue);
consumer0.consumeMsg();
for (int i = 1 ; i < 10 ; i++) {
Thread.sleep(3000);
Producer producer = new Producer(queue);
producer.produceMsg("消息" + i);
}
for (int i = 1 ; i < 10 ; i++) {
Thread.sleep(3000);
Consumer consumer = new Consumer(queue);
consumer.consumeMsg();
}
}
}
- 生产者
public class Producer {
private final BlockingQueue<String> blockingQueue;
public Producer(BlockingQueue<String> blockingQueue) {
this.blockingQueue = blockingQueue;
}
public void produceMsg(String msg) {
InnerProducerWorker worker = new InnerProducerWorker(msg);
worker.start();
}
private void addTask(String msg, String state) {
try {
blockingQueue.put(msg);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("================= producer : " + msg);
state = ThreadState.STOPPING.name();
}
private final class InnerProducerWorker {
private String state;
private Thread t;
public InnerProducerWorker(String msg) {
this.t = new Thread(() -> {
this.state = ThreadState.RUNNING.name();
addTask(msg, state);
if (this.state.equals(ThreadState.STOPPING.name())
&& !t.isInterrupted()) {
t.interrupt();
}
});
}
public void start() {
t.start();
this.state = ThreadState.STARTED.name();
}
}
}
- 消费者
public class Consumer {
private final BlockingQueue<String> blockingQueue;
public Consumer(BlockingQueue<String> blockingQueue) {
this.blockingQueue = blockingQueue;
}
public void consumeMsg() {
InnerConsumerWorker worker = new InnerConsumerWorker();
worker.start();
}
public String getMsg(String state) {
try {
String msg = blockingQueue.take();
state = ThreadState.STOPPING.name();
return msg;
} catch (InterruptedException e) {
e.printStackTrace();
}
return null;
}
private final class InnerConsumerWorker {
private String state;
private Thread t;
public InnerConsumerWorker() {
this.t = new Thread(() -> {
this.state = ThreadState.RUNNING.name();
String msg = getMsg(state);
System.out.println("================= consumer : " + msg);
if (this.state.equals(ThreadState.STOPPING.name())
&& !t.isInterrupted()) {
t.interrupt();
}
});
}
public void start() {
t.start();
this.state = ThreadState.STARTED.name();
}
}
}
参考
《Java并发编程的艺术》
网友评论