生产者消费者模式－java原生、Disruptor实现方案

生产者消费者模式介绍

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

这个阻塞队列就是用来给生产者和消费者解耦的。阻塞队列如何实现高并发多线程安全也是生产者消费者模式中的核心关键。

在日常开发过程中，我们常常会遇到一些高并发场景，例如很多秒杀场景，其实真实的秒杀场景会很复杂，这里只是简单描述下秒杀场景下的生产者消费者模式，在秒杀场景下生产者是普通参与秒杀的用户，消费者是秒杀系统，通常来说这样的场景下秒杀用户是非常多的，如果系统采用常规的实时同步交易，那么势必造成系统处理线程池被瞬间占满，后续请求全部被丢弃，这样造成的用户体验是非常差的，而且系统可能会出现快速雪崩。在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。秒杀场景是个非常适合生产者、消费者的模式，通过中间的缓冲队列解决秒杀请求接收和秒杀处理两者之间的处理时间差，并且能够在过程中通过反欺诈和公平算法来保证消费者的公平利益。下面我们就来用java实现下生产者和消费者模式，这里实现的是可以直接用于生产环境的架构，并不是简单的使用Queue做个简单的进栈出栈，而是通过java.util.concurrent下的相关类实现生产者消费者模式的多线程方案。

Java实现生产者消费者模式

队列的特性：先进先出（FIFO）—先进入队列的元素先出队列（可以理解为我们生活中的排队情况，早办完，早滚蛋）。生产者(Producer)往队列里发布(publish)事件，消费者(Consumer)获得通知，消费事件；如果队列中没有事件时，消费者堵塞，直到生产者发布了新事件。

说到队列，那就不得不提到Java中的concurrent包，其主要实现包括ArrayBlockingQueue、LinkedBlockingQueue、ConcurrentLinkedQueue、LinkedTransferQueue。下面，简单介绍下：

  • ArrayBlockingQueue：基于数组形式的队列，通过加锁的方式，来保证多线程情况下数据的安全；

  • LinkedBlockingQueue：基于链表形式的队列，也通过加锁的方式，来保证多线程情况下数据的安全；

  • ConcurrentLinkedQueue：基于链表形式的队列，通过compare and swap(简称CAS)协议的方式，来保证多线程情况下数据的安全，不加锁，主要使用了Java中的sun.misc.Unsafe类来实现；

  • LinkedTransferQueue：同上；

因为LinkedBlockingQueue采用了乐观锁方案，所以性能是非常高的，下面我们就用LinkedBlockingQueue作为队列缓冲区来实现生产者消费者模式。

待处理数据类

首先我们需要实现一个缓冲队列中的待处理类，这里例子中实现的比较简单，只是设置了一个int类型的变量，重写了构造函数并定义了get方法，大家可以根据自己的需要定义相关的内容。

public class PCData {
    private final int intData;

    public PCData(int intData) {
        this.intData = intData;
    }

    public int getIntData() {
        return intData;
    }

    @Override
    public String toString() {
        return "PCData{" +
                "intData=" + intData +
                '}';
    }
}

生产者类

下面我们定义生产者类，在生产者类中需要定义一个缓冲队列，这里使用了刚才提到的BlockingDeque。

private BlockingDeque<PCData> queue;

生产者中还需要再定义一个静态的AtomicInteger类型的对象，用于多线程中共享数据，用于生成PCData，为什么使用AtomicInteger类型，是因为AtomicInteger类型已经实现了线程安全的自增功能，在实际项目使用过程中，这个值可能是UUID或者其他的全局唯一的数值。

private static AtomicInteger count = new AtomicInteger();

还需要重写构造方法，在生成生产者的时候使用同一个缓冲队列，来保证生产者和开发者都使用一样的队列，在实际项目中也可以定一个全局的队列，来保证所有的生产者和消费者都使用同一个对列。

//定义入参为BlockingQueue的构造函数
    public Producer(BlockingDeque<PCData> queue){
        this.queue = queue;
    }

生产者的核心方法中主要实现了创建PCData类并将该待处理对象放入缓冲队列中，这里为了模拟处理耗时，sleep了1秒钟，所有继承子BlockingDeque的队列类都实现了offer方法，该方法主要是将待处理对象放入缓冲队列中，这样生产者就完成了生产者的基本工作，创建待处理类对象，并将其放入队列。

Thread.sleep(1000);
data = new PCData(count.incrementAndGet());
queue.offer(data);

下面是整个Producer的代码：

import java.util.concurrent.BlockingDeque;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author: feiweiwei
 * @Description: 生产者
 * @Created Date: 17:14 17/9/10.
 * @Modify by:
 */
public class Producer implements Runnable {

    private volatile boolean isrunning = true;
    //内存缓冲队列
    private BlockingDeque<PCData> queue;
    private static AtomicInteger count = new AtomicInteger();

    //定义入参为BlockingQueue的构造函数
    public Producer(BlockingDeque<PCData> queue){
        this.queue = queue;
    }

    public void stop(){
        this.isrunning = false;
    }

    @Override
    public void run() {
        PCData data = null;
        System.out.println("producer id = " + Thread.currentThread().getId());

        while (isrunning) {
            try {
                Thread.sleep(1000);
                data = new PCData(count.incrementAndGet());
                queue.offer(data);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        }
    }
}

消费者类

消费者类的核心工作就是将待处理数据从缓冲队列中取出，并处理。

在消费者类中同样有个BlockingDeque<PCData>的对象，同样也是在创建消费者类的时候从外部传入，这样可以保证所有生产者和消费者使用一样的队列。

在核心处理逻辑中通过BlockingDeque的take方法取出待处理对象，然后就可以对该对象进行处理了，调用take方法后，该待处理对象也自动从queue中弹出。

下面是消费者实现代码：

import java.util.concurrent.BlockingDeque;

/**
 * @Author: feiweiwei
 * @Description: 消费者类
 * @Created Date: 17:26 17/9/10.
 * @Modify by:
 */
public class Customer implements Runnable {

    private BlockingDeque<PCData> queue;
    private volatile boolean isrunning = true;

    //定义入参为BlockingQueue的构造函数
    public Customer(BlockingDeque<PCData> queue){
        this.queue = queue;
    }

    public void stop(){
        this.isrunning = false;
    }

    @Override
    public void run() {
        System.out.println("customer id = " + Thread.currentThread().getId());

        while (isrunning){
            try {
                PCData data = queue.take();
                if ( null != data){
                    int re = data.getIntData() * data.getIntData();
                    Thread.sleep(1000);
                    System.out.println(Thread.currentThread().getId() + " data is " + re + "done!");
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        }
    }
}

主调用Main

在主调用Main中，我们先创建一个队列长度为10的LinkedBlockingDeque对象作为缓冲队列。

BlockingDeque<PCData> queue = new LinkedBlockingDeque<PCData>(10);

再分别创建10个生产者对象和2个消费者，并将刚才创建的queue对象作为构造函数入参。

Producer[] producers = new Producer[10];
Customer[] customers = new Customer[2];

for(int i=0; i<10; i++){
    producers[i] = new Producer(queue);
}
for(int j=0; j<2; j++){
    customers[j] = new Customer(queue);
}

创建一个线程池将生产者和消费者调用起来，这里的线程池大家可以使用自定义的线程池。

ExecutorService es = Executors.newCachedThreadPool();
for(Producer producer : producers){
    es.execute(producer);
}

for(Customer customer : customers){
    es.execute(customer);
}

下面是Main代码：

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

/**
 * @Author: feiweiwei
 * @Description:
 * @Created Date: 17:29 17/9/10.
 * @Modify by:
 */
public class Main {

    public static void main(String[] args) throws InterruptedException {
        BlockingDeque<PCData> queue = new LinkedBlockingDeque<PCData>(10);
        Producer[] producers = new Producer[10];
        Customer[] customers = new Customer[2];

        for(int i=0; i<10; i++){
            producers[i] = new Producer(queue);
        }
        for(int j=0; j<2; j++){
            customers[j] = new Customer(queue);
        }
        ExecutorService es = Executors.newCachedThreadPool();
        for(Producer producer : producers){
            es.execute(producer);
        }

        for(Customer customer : customers){
            es.execute(customer);
        }
        Thread.sleep(10000);
        for(Producer producer : producers){
            producer.stop();
        }
        for(Customer customer : customers){
            customer.stop();
        }

        es.shutdown();
    }
}

Disruptor实现生产者消费者模式

刚才那个是我们自己使用java.util.concurrent下的类实现的生产者消费者模式，目前业界已经有比较成熟的方案，这里向大家推荐LMAX公司开源的Disruptor框架，Disruptor是一个开源的框架，可以在无锁的情况下对队列进行操作，那么这个队列的设计就是Disruptor的核心所在。

Screenshot 2017-09-18 14.38.08.png

在Disruptor中，采用了RingBuffer来作为队列的数据结构，RingBuffer就是一个环形的数组，既然是数组，我们便可对其设置大小。在这个ringBuffer中，除了数组之外，还有一个序列号，是用来指向数组中的下一个可用元素，供生产者使用或者消费者使用，也就是生产者可以生产的地方，或者消费者可以消费的地方。在Disruptor中使用的是位运算，并且在Disruptor中数组内的元素并不会被删除，而是新数据来覆盖原有数据，所以整个环链的处理效率非常高。

下面我们使用Disruptor来实现刚才用jdk自带库实现的生产者消费者。

Disruptor主要类

  • Disruptor：Disruptor的入口，主要封装了环形队列RingBuffer、消费者集合ConsumerRepository的引用；主要提供了获取环形队列、添加消费者、生产者向RingBuffer中添加事件（可以理解为生产者生产数据）的操作；

  • RingBuffer：Disruptor中队列具体的实现，底层封装了Object[]数组；在初始化时，会使用Event事件对数组进行填充，填充的大小就是bufferSize设置的值；此外，该对象内部还维护了Sequencer（序列生产器）具体的实现；

  • Sequencer：序列生产器，分别有MultiProducerSequencer（多生产者序列生产器） 和 SingleProducerSequencer（单生产者序列生产器）两个实现类。上面的例子中，使用的是SingleProducerSequencer；在Sequencer中，维护了消费者的Sequence（序列对象）和生产者自己的Sequence（序列对象）；以及维护了生产者与消费者序列冲突时候的等待策略WaitStrategy；

  • Sequence：序列对象，内部维护了一个long型的value，这个序列指向了RingBuffer中Object[]数组具体的角标。生产者和消费者各自维护自己的Sequence；但都是指向RingBuffer的Object[]数组；

  • Wait Strategy：等待策略。当没有可消费的事件时，消费者根据特定的策略进行等待；当没有可生产的地方时，生产者根据特定的策略进行等待；

  • Event：事件对象，就是我们Ringbuffer中存在的数据，在Disruptor中用Event来定义数据，并不存在Event类，它只是一个定义；

  • EventProcessor：事件处理器，单独在一个线程内执行，判断消费者的序列和生产者序列关系，决定是否调用我们自定义的事件处理器，也就是是否可以进行消费；

  • EventHandler：事件处理器，由用户自定义实现，也就是最终的事件消费者，需要实现EventHandler接口；

  • Producer：事件生产者，也就是我们上面代码中最后那部门的for循环；

待处理类

Disruptor的待处理类和自己实现的待处理类没有本质的区别，可以按照自己要求进行定义。

public class PCData {
    private int data;

    public int getData() {
        return data;
    }

    public void setData(int data) {
        this.data = data;
    }
}

待处理类工厂

这里需要实现disruptor的EventFactory接口，并且实现newInstance方法。这里我们实现的newInstance方法，其实就是创建待处理类的对象，该工厂类在创建Disruptor对象的时候会使用到。

import com.lmax.disruptor.EventFactory;

/**
 * @Author: feiweiwei
 * @Description: 待处理类工厂
 * @Created Date: 18:55 17/9/10.
 * @Modify by:
 */
public class PCDataFactory implements EventFactory<PCData> {
    @Override
    public PCData newInstance() {
        return new PCData();
    }
}

disruptor生产者类

同样需要在生产者中定义一个RingBuffer<PCData>的环形队列，还需要实现一个push的方法，通过ringBuffer.next()取到下一个待处理类序列号，使用ringBuffer.get(sequence)获取到这个序列号对应的待处理类，并对待处理类进行赋值为新的待处理类。
最后通过ringBuffer.publish(sequence)才会将待处理对象发布出来，消费者才能看到。

import com.lmax.disruptor.RingBuffer;

/**
 * @Author: feiweiwei
 * @Description: disruptor生产者类
 * @Created Date: 18:56 17/9/10.
 * @Modify by:
 */
public class Producer {
    private final RingBuffer<PCData> ringBuffer;

    public Producer(RingBuffer<PCData> ringBuffer) {
        this.ringBuffer = ringBuffer;
    }

    public void pushData(int data){
        long sequence = ringBuffer.next();
        try{
            PCData event = ringBuffer.get(sequence);
            event.setData(data);
        }finally {
            ringBuffer.publish(sequence);
        }
    }
}

disruptor消费者

disruptor的消费者类需要实现WorkHandler接口，并实现onEvent方法来处理待处理类，例子中只是对待处理类中的值做了平方。

import com.lmax.disruptor.WorkHandler;

/**
 * @Author: feiweiwei
 * @Description: disruptor消费者
 * @Created Date: 18:52 17/9/10.
 * @Modify by:
 */
public class Consumer implements WorkHandler<PCData> {
    @Override
    public void onEvent(PCData pcData) throws Exception {
        System.out.println(Thread.currentThread().getId() +
        "Event = " + pcData.getData()*pcData.getData());
    }
}

Main

待处理类、待处理工厂、生产者、消费者都定义好之后就可以进行使用了，定义一个缓行队列为1024的disruptor对象，这里构造函数入参看名字就知道了，很简单。

PCDataFactory factory = new PCDataFactory();
int bufferSize = 1024;
Disruptor<PCData> disruptor = new Disruptor<PCData>(factory,bufferSize,executor,
ProducerType.MULTI,new BlockingWaitStrategy());

给disruptor对象定义消费者，这里就简单定义两个consumer作为生产者。

disruptor.handleEventsWithWorkerPool(new Consumer(),new Consumer());

初始化Producer并且将ringBuffer作为构造函数入参，并通过生产者循环100次将数据push入队列，消费者会自动从队列取值进行处理。

RingBuffer<PCData> ringBuffer = disruptor.getRingBuffer();
Producer producer = new Producer(ringBuffer);

for(int i=0; i<100; i++){
    producer.pushData(i);
    Thread.sleep(100);
    System.out.println("push data " + i);
}

以下为Main全部代码：

package com.monkey01.producercustomer.disruptor;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

/**
 * @Author: feiweiwei
 * @Description:
 * @Created Date: 18:59 17/9/10.
 * @Modify by:
 */
public class Main {
    public static void main(String args[]) throws InterruptedException {
        Executor executor = Executors.newCachedThreadPool();
        PCDataFactory factory = new PCDataFactory();
        int bufferSize = 1024;
        Disruptor<PCData> disruptor = new Disruptor<PCData>(factory,bufferSize,executor,
        ProducerType.MULTI,new BlockingWaitStrategy());

        disruptor.handleEventsWithWorkerPool(new Consumer(),
                new Consumer());
        disruptor.start();

        RingBuffer<PCData> ringBuffer = disruptor.getRingBuffer();
        Producer producer = new Producer(ringBuffer);

        for(int i=0; i<100; i++){
            producer.pushData(i);
            Thread.sleep(100);
            System.out.println("push data " + i);
        }

        disruptor.shutdown();
    }
}

总结

大家看到这里也基本对生产者、消费者模式有个比较深入的了解了，也可以按照文中的例子，在自己的项目中使用，这个模式在日常项目中还是比较常见的，希望大家能够熟练使用该模式。

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 206,126评论 6赞 481
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 88,254评论 2赞 382
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 152,445评论 0赞 341
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 55,185评论 1赞 278
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 64,178评论 5赞 371
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,970评论 1赞 284
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 38,276评论 3赞 399
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,927评论 0赞 259
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 43,400评论 1赞 300
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,883评论 2赞 323
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,997评论 1赞 333
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,646评论 4赞 322
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 39,213评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,204评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,423评论 1赞 260
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 45,423评论 2赞 352
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,722评论 2赞 345

推荐阅读更多精彩内容

  • Spring Cloud
    Spring Cloud为开发人员提供了快速构建分布式系统中一些常见模式的工具（例如配置管理，服务发现，断路器，智...
    卡卡罗2017阅读 134,599评论 18赞 139
  • Java 基础面试宝典(自己总结)
    一、多线程 说明下线程的状态 java中的线程一共有 5 种状态。 NEW：这种情况指的是，通过 New 关键字创...
    Java旅行者阅读 4,656评论 0赞 44
  • Java后台面试 常见问题
    从三月份找实习到现在，面了一些公司，挂了不少，但最终还是拿到小米、百度、阿里、京东、新浪、CVTE、乐视家的研发岗...
    时芥蓝阅读 42,184评论 11赞 349
  • 有关父亲诞辰百年的纪念(五)
    傍晚到家,隔壁二大伯正和落炕(卧床不起)的父亲大嚷白喝地说着话。他喝 醉了。 二大伯是鳏夫,性...
    董耀奎阅读 239评论 0赞 0
  • 日精进：
    今天晚上西红门后街又着火了：安全重于泰山。 明天整理安全隐患
    京心达张新波阅读 205评论 0赞 0