`

生产者消费者模式-Java实现（转帖）

 
阅读更多

转帖地址：https://www.cnblogs.com/chentingk/p/6497107.html

  • 感知阶段

　　　　随着软件业的发展，互联网用户的日渐增多，并发这门艺术的兴起似乎是那么合情合理。每日PV十多亿的淘宝，处理并发的手段可谓是业界一流。用户访问淘宝首页的平均等待时间只有区区几秒，但是服务器所处理的流程十分复杂。首先负责首页的服务器就有好几千台，通过计算把与用户路由最近的服务器处理首页的返回。其次是网页上的资源，就JS和CSS文件就有上百个，还有图片资源等。它能在几秒内加载出来可见阿里几千名顶尖工程师的智慧是如何登峰造极。

　　　　而在大型电商网站中，他们的服务或者应用解耦之后，是通过消息队列在彼此间通信的。消息队列和应用之间的架构关系就是生产者消费者模型。

　　　　在介绍之前，先找找现实间的模型。笔者最近发觉，很多技术模型是和生活中的模型息息相关的。相信多数人都进过肯德基和麦当劳消费，笔者进店消费的时候发现他们的点单流程和并发模型十分接近。虽然每家店的流程有所差异，但是大概就只有两种模型。在肯德基里，你点单之后点单员会把所点的食物完成封装之后拿来你面前，然后让你结账，有时候有些耗时操作没完成就会留下一个餐台号稍后送来。而在麦当劳的点餐模型大致是，你点完快餐之后要求你立即付款，付完款之后下一位点餐，而取餐的是在旁边等待，另一个服务员专责负责配餐。

                                               肯德基流程

                                                      麦当劳点餐图

　　　　在并发模型中，肯德基比较倾向于一个线程把所有的服务都做完，而麦当劳倾向于服务解耦，让他们更专注于自己的业务。而肯德基的模型与BIO服务器的模型设计类似，麦当劳的模型则与生产者消费者模型十分相似。

  • 生产消费者模型

　　　　生产者消费者模型具体来讲，就是在一个系统中，存在生产者和消费者两种角色，他们通过内存缓冲区进行通信，生产者生产消费者需要的资料，消费者把资料做成产品。生产消费者模式如下图。

　　　　　　　　　　　　　　

　　　　在日益发展的服务类型中，譬如注册用户这种服务，它可能解耦成好几种独立的服务（账号验证，邮箱验证码，手机短信码等）。它们作为消费者，等待用户输入数据，在前台数据提交之后会经过分解并发送到各个服务所在的url，分发的那个角色就相当于生产者。消费者在获取数据时候有可能一次不能处理完，那么它们各自有一个请求队列，那就是内存缓冲区了。做这项工作的框架叫做消息队列。

  • 生产者消费者模型的实现

　　　　生产者是一堆线程，消费者是另一堆线程，内存缓冲区可以使用List数组队列，数据类型只需要定义一个简单的类就好。关键是如何处理多线程之间的协作。这其实也是多线程通信的一个范例。

　　　　在这个模型中，最关键就是内存缓冲区为空的时候消费者必须等待，而内存缓冲区满的时候，生产者必须等待。其他时候可以是个动态平衡。值得注意的是多线程对临界区资源的操作时候必须保证在读写中只能存在一个线程，所以需要设计锁的策略。

　　　　下面这个例子是书上介绍的，生产者负责生产一个数字并存入缓冲区，消费者从缓冲区中取出数据并且求出它的平方并输出。

复制代码
package ProducterAndConsumer.Version1;

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 生产者
 * @author ctk
 * 生产者消费者模型
 */

public class Producer implements Runnable {
    private volatile boolean isRunning = true;
    private BlockingQueue<PCData> queue;// 内存缓冲区
    private static AtomicInteger count = new AtomicInteger();// 总数 原子操作
    private static final int SLEEPTIME = 1000;

    public Producer(BlockingQueue<PCData> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        PCData data = null;
        Random r = new Random();
        System.out.println("start producting id:" + Thread.currentThread().getId());
        try {
            while (isRunning) {
                Thread.sleep(r.nextInt(SLEEPTIME));
                data = new PCData(count.incrementAndGet());
                System.out.println(data + " 加入队列");
                if (!queue.offer(data, 2, TimeUnit.SECONDS)) {
                    System.err.println(" 加入队列失败");
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            Thread.currentThread().interrupt();
        }

    }

    public void stop() {
        isRunning = false;
    }
}
复制代码
复制代码
package ProducterAndConsumer.Version1;
/**
 * 消费者
 * @author ctk
 */
import java.text.MessageFormat;
import java.util.Random;
import java.util.concurrent.BlockingQueue;

public class Consumer implements Runnable{
    private BlockingQueue<PCData> queue;
    private static final int SLEEPTIME = 1000;
    public Consumer(BlockingQueue<PCData> queue){
        this.queue = queue;
    }

    @Override
    public void run() {
        System.out.println("start Consumer id :"+Thread.currentThread().getId());
        Random r = new Random();
        try{
            while(true){
                PCData data = queue.take();
                if(data != null)
                {
                    int re = data.getData() * data.getData();
                    System.out.println(MessageFormat.format("{0}*{1}={2}", data.getData(),data.getData(),re));
                    Thread.sleep(r.nextInt(SLEEPTIME));
                }
            }
        }catch (InterruptedException e) {
            e.printStackTrace();
            Thread.currentThread().interrupt();
        }
    }

}
复制代码
复制代码
package ProducterAndConsumer.Version1;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * 主函数
 * @author ctk
 *
 */
public class Main {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<PCData> queue = new LinkedBlockingDeque<>(10);
        Producer p1 = new Producer(queue);
        Producer p2 = new Producer(queue);
        Producer p3 = new Producer(queue);
        Consumer c1 = new Consumer(queue);
        Consumer c2 = new Consumer(queue);
        Consumer c3 = new Consumer(queue);
        ExecutorService service = Executors.newCachedThreadPool();
        service.execute(p1);
        service.execute(p2);
        service.execute(p3);
        service.execute(c1);
        service.execute(c2);
        service.execute(c3);
        Thread.sleep(10*1000);
        p1.stop();
        p2.stop();
        p3.stop();
        Thread.sleep(3000);
        service.shutdown();
    }
}
复制代码
复制代码
package ProducterAndConsumer.Version1;
/**
 * 容器数据类型
 * @author ctk
 *
 */
public class PCData {
    private final int intData;
    public PCData(int d){
        intData = d;
    }
    public PCData(String d){
        intData = Integer.valueOf(d);
    }
    public int getData(){
        return intData;
    }
    @Override
    public String toString(){
        return "data:"+intData;
    }
}
复制代码

　　　　因为BlockingQueue是一个阻塞队列，它的存取可以保证只有一个线程在进行，所以根据逻辑，生产者在内存满的时候进行等待，并且唤醒消费者队列，反过来消费者在饥饿状态下等待并唤醒生产者进行生产。

　　　　下面的两个版本是使用notify/wait()和await()/signal()方法进行设计的。在结构上是一致遵从模型图的。

复制代码
package ProducterAndConsumer.Version2;

import java.util.List;

/**
 * 消费者
 * 
 * @author ctk
 *
 */

public class Consumer implements Runnable {
    private List<PCData> queue;

    public Consumer(List<PCData> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            while (true) {
                if (Thread.currentThread().isInterrupted())
                    break;
                PCData data = null;
                synchronized (queue) {
                    if (queue.size() == 0) {
                        queue.wait();
                        queue.notifyAll();
                    }
                    data = queue.remove(0);
                }
                System.out.println(
                        Thread.currentThread().getId() + " 消费了:" + data.get() + " result:" + (data.get() * data.get()));
                Thread.sleep(1000);
            }

        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

}


package ProducterAndConsumer.Version2;

import java.util.List;
import java.util.Random;

/**
 * 生产者
 * 
 * @author MacBook
 *
 */
public class Producer implements Runnable {
    private List<PCData> queue;
    private int length;

    public Producer(List<PCData> queue, int length) {
        this.queue = queue;
        this.length = length;
    }

    @Override
    public void run() {
        try {
            while (true) {

                if (Thread.currentThread().isInterrupted())
                    break;
                Random r = new Random();
                long temp = r.nextInt(100);
                System.out.println(Thread.currentThread().getId() + " 生产了：" + temp);
                PCData data = new PCData();
                data.set(temp);
                synchronized (queue) {
                    if (queue.size() >= length) {
                        queue.notifyAll();
                        queue.wait();
                    } else
                        queue.add(data);
                }
                Thread.sleep(1000);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

}


package ProducterAndConsumer.Version2;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Main {
    public static void main(String[] args) {
        List<PCData> queue = new ArrayList<PCData>();
        int length = 10;
        Producer p1 = new Producer(queue,length);
        Producer p2 = new Producer(queue,length);
        Producer p3 = new Producer(queue,length);
        Consumer c1 = new Consumer(queue);
        Consumer c2 = new Consumer(queue);
        Consumer c3 = new Consumer(queue);
        ExecutorService service = Executors.newCachedThreadPool();
        service.execute(p1);
        service.execute(p2);
        service.execute(p3);
        service.execute(c1);
        service.execute(c2);
        service.execute(c3);
        
    }
}


package ProducterAndConsumer.Version2;
/**
 * 基本数据类型
 * @author ctk
 *
 */
public class PCData {
    private long value;
    public void set(long value){
        this.value = value;
        
    }
    public long get(){
        return value;
    }
}
复制代码

　　　　

　　　　

复制代码
package ProducterAndConsumer.Version3;

import java.util.List;
/**
 * 消费者
 * @author ctk
 *
 */
public class Consumer implements Runnable{
    private List<PCData> queue;
    public Consumer(List<PCData> queue){
        this.queue = queue;
    }
    @Override
    public void run() {
        try {
            while (true) {
                if (Thread.currentThread().isInterrupted())
                    break;
                PCData data = null;
                Main.lock.lock();
                if (queue.size() == 0){
                    Main.full.signalAll();
                    Main.empty.await();
                }
                Thread.sleep(1000);
                data = queue.remove(0);
                Main.lock.unlock();
                System.out.println("消费者ID:"+Thread.currentThread().getId()+" 消费了:"+data.getData()+" result:"+(data.getData()*data.getData()));
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    
}


package ProducterAndConsumer.Version3;

import java.util.List;
import java.util.Random;
/**
 * 生产者
 * @author ctk
 *
 */
public class Producter implements Runnable{
    private List<PCData> queue;
    private int len;
    public Producter(List<PCData> queue,int len){
        this.queue = queue;
        this.len = len;
    }
    @Override
    public void run() {
        try{
            while(true){
                if(Thread.currentThread().isInterrupted())
                    break;
                Random r = new Random();
                PCData data = new PCData();
                data.setData(r.nextInt(500));
                Main.lock.lock();
                if(queue.size() >= len)
                {
                    Main.empty.signalAll();
                    Main.full.await();
                }
                Thread.sleep(1000);
                queue.add(data);
                Main.lock.unlock();
                System.out.println("生产者ID:"+Thread.currentThread().getId()+" 生产了:"+data.getData());
            }
        }catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}


package ProducterAndConsumer.Version3;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
public class Main {
    public static ReentrantLock lock = new ReentrantLock();
    public static Condition empty = lock.newCondition();
    public static Condition full = lock.newCondition();
    public static void main(String[] args) {
        List<PCData> queue = new ArrayList<PCData>();
        int length = 10;
        Producter p1 = new Producter(queue,length);
        Producter p2 = new Producter(queue,length);
        Producter p3 = new Producter(queue,length);
        Consumer c1 = new Consumer(queue);
        Consumer c2 = new Consumer(queue);
        Consumer c3 = new Consumer(queue);
        ExecutorService service = Executors.newCachedThreadPool();
        service.execute(p1);
        service.execute(p2);
        service.execute(p3);
        service.execute(c1);
        service.execute(c2);
        service.execute(c3);
    }
}


package ProducterAndConsumer.Version3;

public class PCData {
    private int data;

    public int getData() {
        return data;
    }

    public void setData(int data) {
        this.data = data;
    }
}
复制代码

　　　await的版本我个人写出来之后感觉，每次控制台只输出了一句话，说明在同一时间内生产者或者消费者只有一个是激活的，而wait的版本，一次可能有多个生成者激活。我个人觉得wait的版本更接近我的构想。

  • 生产消费者模型思维

　　　　下午翻书，偶然发现并行计算的流水线思维。并行计算的要点就是分治法思维，如果能证明分割的两部分在因果上没有关联，则可以进行并行计算。譬如书上的例子（A+B）*C，这个算式是不能使用并行计算分割的，因为它的结果是A+B之后的结果乘以C。但是并行流水线的思维是，我们可以请两个工人，每个工人负责一步的处理。

　　　　分解后的架构是：P1：D = A + B；P2：R = D*3；

　　　　在这两个线程处理中并不需要存在因果，所以他们可以并行计算了。

　　　　设计这个模式是基于生产消费者模型的，流水线需要使用流水线传递半成品，流水线就是内存缓冲区，对于P2来说，P1就是生产者，而对于系统需要的结果来说，P2就是生产者。

　　　　　　　　　　　　　　

  • 后记

　　　　偶然读到一本书，上面提到的建立高速公路的学习方法是十分高效的学习方法，在学习新的技术的时候它们或多或少都会在现实中有所映射，所以读万卷书行万里路，经历和学术需要并行增长。技术模型不仅应用在技术领域，管理领域也可以参照思考，learn more，study less。

分享到：
| Java中单例对象不会被GC回收（转帖）
评论
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

    冒险岛群聊机器人 蠢猫（无视计算 星之力计算 世界组队表 占卜 怪物查询）.zip

    无人机最强算法源码，易于部署和学习交流使用

    node-v8.14.1-linux-x64.tar.xz

    Node.js，简称Node，是一个开源且跨平台的JavaScript运行时环境，它允许在浏览器外运行JavaScript代码。Node.js于2009年由Ryan Dahl创立，旨在创建高性能的Web服务器和网络应用程序。它基于Google Chrome的V8 JavaScript引擎，可以在Windows、Linux、Unix、Mac OS X等操作系统上运行。 Node.js的特点之一是事件驱动和非阻塞I/O模型，这使得它非常适合处理大量并发连接，从而在构建实时应用程序如在线游戏、聊天应用以及实时通讯服务时表现卓越。此外，Node.js使用了模块化的架构，通过npm（Node package manager，Node包管理器）,社区成员可以共享和复用代码，极大地促进了Node.js生态系统的发展和扩张。 Node.js不仅用于服务器端开发。随着技术的发展，它也被用于构建工具链、开发桌面应用程序、物联网设备等。Node.js能够处理文件系统、操作数据库、处理网络请求等，因此，开发者可以用JavaScript编写全栈应用程序，这一点大大提高了开发效率和便捷性。 在实践中，许多大型企业和组织已经采用Node.js作为其Web应用程序的开发平台，如Netflix、PayPal和Walmart等。它们利用Node.js提高了应用性能，简化了开发流程，并且能更快地响应市场需求。

    HostelDataset(been personally processed）

    此数据集主要是一个跟酒店类型有关的数据集，源于kaggle网站上的Guatemalas_Travel数据集，因设计所需，本人对原来的数据集进行了必要的修改。

    3D相册.zip

    3D相册.zip

    中国统计NJ面板数据-（更新至2022年）平均每天能源消费量.xls

    数据来源：中国统计NJ-2023版

    3dmax空对象清除器下载

    3dmax空对象清除器，在场景中找到空对象，向你显示这些对象的列表，并允许你从场景中删除它们。

    node-v10.13.0-linux-s390x.tar.xz

    Node.js，简称Node，是一个开源且跨平台的JavaScript运行时环境，它允许在浏览器外运行JavaScript代码。Node.js于2009年由Ryan Dahl创立，旨在创建高性能的Web服务器和网络应用程序。它基于Google Chrome的V8 JavaScript引擎，可以在Windows、Linux、Unix、Mac OS X等操作系统上运行。 Node.js的特点之一是事件驱动和非阻塞I/O模型，这使得它非常适合处理大量并发连接，从而在构建实时应用程序如在线游戏、聊天应用以及实时通讯服务时表现卓越。此外，Node.js使用了模块化的架构，通过npm（Node package manager，Node包管理器）,社区成员可以共享和复用代码，极大地促进了Node.js生态系统的发展和扩张。 Node.js不仅用于服务器端开发。随着技术的发展，它也被用于构建工具链、开发桌面应用程序、物联网设备等。Node.js能够处理文件系统、操作数据库、处理网络请求等，因此，开发者可以用JavaScript编写全栈应用程序，这一点大大提高了开发效率和便捷性。 在实践中，许多大型企业和组织已经采用Node.js作为其Web应用程序的开发平台，如Netflix、PayPal和Walmart等。它们利用Node.js提高了应用性能，简化了开发流程，并且能更快地响应市场需求。

    中国统计NJ面板数据-（更新至2022年）入境外国游客分组构成.xls

    数据来源：中国统计NJ-2023版

    数据要素资产化与数据资产入表路线框架总体规划方案.pptx

    数据要素资产化与数据资产入表路线框架总体规划方案.pptx

    用户手册 8100C/8200C/8300C系列无功补偿装置

    施耐德电气 8100C 8200C 8300C 用户手册

    毫无特色的 QQ 机器人.zip

    无人机最强算法源码，易于部署和学习交流使用

    员工培训方案gl.ppt

    员工培训方案gl.ppt

    infrared-remote-controller-master

    旅行商问题

    小米扫地机器人工程源码程序STM32103 freeRTOS设计源代码.zip

    小米扫地机器人工程源码程序STM32103 freeRTOS设计源代码，某米早起的扫地机器人的源码，使用的是STM32开发，freeRTOS系统，包含完整的工程源码，可以做为你的学习设计参考。 HARDWARE文件夹 硬件初始化源文件 application_Robot 机器人行为相关的文件 application_Share 机器人共享相关的文件（一般为算法） application_Motor 机器人电机应用相关文件 application_Key 机器人按键应用相关文件 application_Led 机器人LED应用相关文件 application_Led 机器人adc获取与处理应用相关文件 application_PowerChip 机器人电池管理芯片与充电应用相关文件 application_Bmi160 机器人BMI160 陀螺仪芯片 应用相关文件 application_Drop 机器人掉落传感器应用相关文件 application_UltraSonic 机器人超声波传感器应用相关文件 appl

    39黎秋菊.ipynb

    39黎秋菊.ipynb

    node-v11.14.0-linux-armv7l.tar.xz

    Node.js，简称Node，是一个开源且跨平台的JavaScript运行时环境，它允许在浏览器外运行JavaScript代码。Node.js于2009年由Ryan Dahl创立，旨在创建高性能的Web服务器和网络应用程序。它基于Google Chrome的V8 JavaScript引擎，可以在Windows、Linux、Unix、Mac OS X等操作系统上运行。 Node.js的特点之一是事件驱动和非阻塞I/O模型，这使得它非常适合处理大量并发连接，从而在构建实时应用程序如在线游戏、聊天应用以及实时通讯服务时表现卓越。此外，Node.js使用了模块化的架构，通过npm（Node package manager，Node包管理器）,社区成员可以共享和复用代码，极大地促进了Node.js生态系统的发展和扩张。 Node.js不仅用于服务器端开发。随着技术的发展，它也被用于构建工具链、开发桌面应用程序、物联网设备等。Node.js能够处理文件系统、操作数据库、处理网络请求等，因此，开发者可以用JavaScript编写全栈应用程序，这一点大大提高了开发效率和便捷性。 在实践中，许多大型企业和组织已经采用Node.js作为其Web应用程序的开发平台，如Netflix、PayPal和Walmart等。它们利用Node.js提高了应用性能，简化了开发流程，并且能更快地响应市场需求。

    域名注册查询的微信小程序页面源码

    模板描述：域名注册查询 微信小程序。域名注册查询的微信小程序页面源码是使用微信WEB开发者工具开发，亲测可直接运行。该程序从设计上满足了相关应用服务的设计要求，是一款非常值得学习简介的DEMO小程序源码。

    水下机器人无线充电上位机.zip

    无人机最强算法源码，易于部署和学习交流使用

    DCS 400系列可控硅直流传动系统手册

    DCS 400 手册

    node-v11.15.0-linux-armv6l.tar.xz

    Node.js，简称Node，是一个开源且跨平台的JavaScript运行时环境，它允许在浏览器外运行JavaScript代码。Node.js于2009年由Ryan Dahl创立，旨在创建高性能的Web服务器和网络应用程序。它基于Google Chrome的V8 JavaScript引擎，可以在Windows、Linux、Unix、Mac OS X等操作系统上运行。 Node.js的特点之一是事件驱动和非阻塞I/O模型，这使得它非常适合处理大量并发连接，从而在构建实时应用程序如在线游戏、聊天应用以及实时通讯服务时表现卓越。此外，Node.js使用了模块化的架构，通过npm（Node package manager，Node包管理器）,社区成员可以共享和复用代码，极大地促进了Node.js生态系统的发展和扩张。 Node.js不仅用于服务器端开发。随着技术的发展，它也被用于构建工具链、开发桌面应用程序、物联网设备等。Node.js能够处理文件系统、操作数据库、处理网络请求等，因此，开发者可以用JavaScript编写全栈应用程序，这一点大大提高了开发效率和便捷性。 在实践中，许多大型企业和组织已经采用Node.js作为其Web应用程序的开发平台，如Netflix、PayPal和Walmart等。它们利用Node.js提高了应用性能，简化了开发流程，并且能更快地响应市场需求。

Magicbox
Global site tag (gtag.js) - Google Analytics