`

生产者消费者模式-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回收（转帖）
评论
发表评论

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

相关推荐

    转帖经典---JAVA设计模式

    《转帖经典---JAVA设计模式》这本书或资料可能涵盖了这些模式的详细解释、示例代码以及如何在实际项目中应用这些模式。通过学习和理解这些设计模式，开发者能够更好地设计和重构软件，提升代码质量。

    转帖--oracle分析函数+PLSQL小结

    PL/SQL（Procedural Language/Structured Query Language）是Oracle数据库内置的面向过程的编程语言，它结合了SQL语句和过程式编程语言的特点，用于编写存储过程、函数、触发器和游标等数据库对象，以实现更复杂的...

    H42131-转帖《关于用净值计算法计算收益》.doc

    H42131-转帖《关于用净值计算法计算收益》.doc

    [转帖]jQuery实现省市联动

    标题中的“jQuery实现省市联动”是指在Web应用中，通过jQuery库来实现在选择省份时，下拉框（或其它选择组件）中的城市选项能够自动更新，以展示与所选省份对应的城市列表。这种功能常见于地址输入或者注册表单中，...

    论坛转帖工具.rar

    标题中的“论坛转帖工具.rar”表明这是一个用于在论坛之间转移帖子的软件工具，通常用于帮助用户方便地将一个论坛的帖子内容复制到另一个论坛，可能是为了分享信息、讨论或保存重要的帖子。这类工具可能包括自动抓取...

    UBB论坛转帖圣手.exe

    UBB论坛转帖圣手.exeUBB论坛转帖圣手.exe

    浮点栈结构(转帖) - 大呵呵 - 博客园_float_floatregister_

    浮点栈结构是计算机科学中的一个重要概念，尤其是在处理器架构和编程语言实现中。本文将深入探讨浮点栈结构，特别是Intel CPU中的浮点寄存器（Float Register）使用情况。 浮点栈结构主要指的是用于存储浮点数值的...

    编辑人员转帖去水印工具

    本篇文章将详细探讨“编辑人员转帖去水印工具”，并介绍如何使用名为Teorex Inpaint的1.0.0.2版本的软件来实现这一目标。 首先，我们要理解什么是水印。水印通常是指在图像或视频中添加的半透明标记，它可以是文字...

    运放参数的详细解释和分析-转帖TI工作人员的资料（doc版）

    在TI论坛看到的帖子，感觉挺好，就整理成word文档了，内容主要是运放datasheet常见参数的解释和分析。TI原帖地址“http://www.deyisupport.com/question_answer/analog/amplifiers/f/52/t/20214.aspx”...感谢原作者。

    一键转帖功能插件 for 帝国CMS 6.0 GBK utf8 V1.0.rar

    本文将深入探讨“一键转帖功能插件”在帝国CMS 6.0系统中的应用与实现，该插件适用于GBK及UTF-8编码环境，旨在提升网站内容的分享与传播效率。我们将从安装步骤、工作原理、可能遇到的问题以及解决方案等方面进行...

    ASP学习资料(HTML格式)

    "交流论坛--Java Servlet和JSP教程之三.htm"同样可能是对比分析ASP与Java Web技术，探讨如何在ASP环境中理解和应用Servlet和JSP的概念。 "交流论坛--手把手教你建ASP语音聊天室.htm"涉及到了更高级的应用，即如何...

    asphtml学习

    7. **Servlet与JSP教程**：`交流论坛--Java Servlet和JSP教程之三.htm`继续了对Java Web开发的讲解，Servlet和JSP是Java Web应用的核心部分。 8. **ASP语音聊天室**：`交流论坛--手把手教你建ASP语音聊天室.htm`...

    2021-2022年收藏的精品资料转帖经济学资源.doc

    【标题】：“2021-2022年收藏的精品资料转帖经济学资源.doc” 【描述】：“精品教育教学资料” 这篇文档整理了2021至2022年间收集的一些高质量经济学教育资源，主要关注的是综合性的经济学研究领域。这些资料对于...

    转帖：液晶显示器原理与维修手册

    - **省电控制**: MCU通过控制电源电路实现省电模式切换。 #### 三、检修提要 - **信号接口电路分析**: GM2121作为信号接口的核心，负责A/D转换和像素时钟产生等功能。 - **关键部件检测**: 重点检查A/D转换器、LCD...

    [转帖]世界编程大赛第一名写的程序

    标题和描述中的“世界编程大赛第一名写的程序”这一知识点，实际上指向了计算机科学与编程竞赛领域的一个重要概念：即在高水平的编程比赛中，优胜者所编写的代码往往蕴含着高级算法、数据结构以及编程技巧。...

    转帖h3c一些命令配置

    LACP（Link Aggregation Control Protocol）用于实现端口聚合，提高链路带宽和冗余。设置LACP系统优先级和创建端口聚合组： ``` [Sysname]lacp system-priority 64 [Sysname]link-aggregation group 1 description ...

    discuz X2转帖工具、采集工具

    用户可以预设帖子的内容、标题、分类等参数，实现批量导入。 2. 数据模板：支持设置数据模板，包括文本、图片、链接等，使得帖子内容多样化，增加论坛的活跃度。 3. 规则设定：用户可以根据需求设定特定的规则，例如...

    贴吧转帖工具

    通过这款工具，用户可以实现一键转帖和一键8经验签到的功能，极大地简化了传统操作流程，节省了用户的时间，提升了用户体验。 一键转帖功能是该工具的核心亮点。它允许用户快速地将一篇帖子从一个贴吧复制并发布到...

    转帖：Android应用的自动升级、更新模块的实现docx.docx

    ```java public static int getVerCode(Context context) { // ... } public static String getVerName(Context context) { // ... } ``` 3. **动态获取版本名称**: 如果将`versionName`设置为字符串资源，如`...

    转帖工具ConvertX fordiscuz7.1/7.2 修改增强版.rar

    1.修改自Convert X转帖工具 2.新增批量替换关键词(原来是单个词语替换,可以利用这个功能删除一些网站的防转帖代码) 3.批量随机新增文字(新增内容可自定义,从而实现伪原创) 4.cookie记录替换和新增关键词(避免每次...

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