JAVA并发编程——生产者与消费者模式（传统版&阻塞队列版） java多线程设计模式

1.前言
2.生产者与消费者模型传统版代码示例
3.生产者与消费者模型阻塞队列版代码示例
1.前言
今天我们要用两种方式来实现生产者和消费者模式，我们要先介绍一个概念，生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而是通过消息队列进行通讯，所以生产者生产完数据后不用等待消费者处理，而是直接扔给队列，消费者不找生产者要数据，而是直接从队列里取数据，队列相当于一个缓冲区，平衡了生产者和消费者的处理能力。这个队列就是用来给生产者和消费者解耦的。我们今天要用两种方式来实现这个模型，一种是阻塞队列版本，一种是传统版本。
2.生产者与消费者模型传统版代码示例
我们先来对传统消费者和生产者模型的一个JAVA代码的实现：
假设我们现在有这么一个需求，一个生产者和一个消费者，他们共同控制一个变量number，一个线程生产数据，将number变为1，一个线程消费数据，将number从1变为0，从此往复循环。
我们使用lock和Condition版本的锁，关于Lock和Condition，可以根据我写的另外一篇文章：
JAVA并发编程——Synchronized与Lock的区别以及Lock的使用
我们先设定三个变量：

//生产者和消费者线程操作的共享对象 private volatile int number = 0; //一把锁 private Lock lock = new ReentrantLock(); //condition用以精确唤醒线程，用来代替wait()和notify()方法 private Condition condition = lock.newCondition();

生产方法：

public void increment() throws Exception { try { //加锁 lock.lock(); //如果线程不是0，则认定为还未消费，先唤醒消费线程 while (number != 0) { condition.await(); } //生产数据 number++; System.out.println(Thread.currentThread().getName() + "\t" + number); //唤醒所有线程 condition.signalAll(); } finally { lock.unlock(); } }

消费方法：

public void decrement() throws Exception { try { //加锁 lock.lock(); //如果线程是0，则认定为还未生产，先唤醒生产线程 while (number == 0) { condition.await(); } //消费 number--; System.out.println(Thread.currentThread().getName() + "\t" + number); //唤醒所有线程 condition.signalAll(); } finally { lock.unlock(); } }

接下来我们各开启1000个线程进行生产和消费：

public static void main(String[] args) {ShareData shareData = https://www.it610.com/article/new ShareData(); for (int i = 1; i <= 500; i++) { new Thread(() -> { try { shareData.increment(); } catch (Exception e) { e.printStackTrace(); } }, "线程AA生产：").start(); }for (int i = 1; i <= 500; i++) { new Thread(() -> { try { shareData.decrement(); } catch (Exception e) { e.printStackTrace(); } }, "线程BB消费：").start(); }for (int i = 1; i <= 500; i++) { new Thread(() -> { try { shareData.increment(); } catch (Exception e) { e.printStackTrace(); } }, "线程CC生产：").start(); }for (int i = 1; i <= 500; i++) { new Thread(() -> { try { shareData.decrement(); } catch (Exception e) { e.printStackTrace(); } }, "线程DD消费：").start(); }}}

运行结果为：

文章图片

成功地实现了传统版本生产者和消费者模式。
3.生产者与消费者模型阻塞队列版代码示例
接下来我们使用阻塞队列版本来实现以下生产者和消费者模式，至于阻塞队列，在之前的文章讲过：
JAVA并发编程——阻塞队列理论及其API介绍
我们设定三个变量和一个构造方法

private volatile boolean flag = true; //默认开启，true生产+f消费false 退出生产+消费 private AtomicInteger atomicInteger = new AtomicInteger(); //代表生产的数据 private BlockingQueue blockingQueue = null; //阻塞队列 public MyResources(BlockingQueue blockingQueue) { this.blockingQueue = blockingQueue; //阻塞队列的类型通过构造方法确定 }

【JAVA并发编程——生产者与消费者模式（传统版&阻塞队列版）】生产方法：

public void myProd() throws InterruptedException { String data = https://www.it610.com/article/null; //生产的数据内容 boolean retValue; //生产是否成功 while (flag) { //true代表生产着消费者模式开启，false代表结束 data = atomicInteger.getAndIncrement() +""; //阻塞队列超时等待方法 retValue = https://www.it610.com/article/blockingQueue.offer(data, 2L, TimeUnit.SECONDS); if (retValue) System.out.println(Thread.currentThread().getName() +"\t 插入队列" + data + "成功"); else System.out.println(Thread.currentThread().getName() + "\t 插入队列" + data + "失败"); Thread.sleep(1000); } System.out.println("falg = false 生产叫停！"); }

消费方法：

public void myCustomer() throws InterruptedException { String result = null; while (flag) { //true代表生产着消费者模式开启，false代表结束 //阻塞队列超时等待方法 result = blockingQueue.poll(2L, TimeUnit.SECONDS); if (result == null || result.equalsIgnoreCase("")) { flag = false; System.out.println("超过两秒钟没有数据消费，退出"); System.out.println(); System.out.println(); return; } System.out.println(Thread.currentThread().getName() + "\t 消费队列" + result + "成功"); Thread.sleep(1000); } }

停止生产方法：

public void stop() { this.flag = false; //当flag等于0的时候，生产和消费就停止了 }

接下来我们创建两个线程进行生产和消费，生产和消费超过10秒后，停止两个线程的运行：

public static void main(String[] args) throws InterruptedException { MyResources myResources = new MyResources(new ArrayBlockingQueue(3)); new Thread(() -> { try { myResources.myProd(); } catch (InterruptedException e) { e.printStackTrace(); } }, "生产线程").start(); new Thread(() -> { try { myResources.myCustomer(); } catch (InterruptedException e) { e.printStackTrace(); } }, "消费线程").start(); Thread.sleep(10000); System.out.println("生产叫停！"); myResources.stop(); }

运行结果：