Disruptor简单使用

良辰美景TT

发布于 2019-03-29 16:24:59

7700

发布于 2019-03-29 16:24:59

文章被收录于专栏：java、Spring、技术分享java、Spring、技术分享

Disruptor从功能上来说，可以实现队列的功能，也可以把它当成单机版的JMS来看待。从性能上来说，它比ArrayBlockingQueue有更好的性能表现，对于生产者消费者模型的业务，Disruptor是一个更好的选择可以很好的实现业务的分离。

简单入门

定义消息类，这里的消息在Disruptor里称为Event，也就是我们系统里生产消费的业务对象，示例代码如下：

package com.example.disruptor;

/**
 * 产品
 */
public class Product {

    private int id;

    private String name;

    private double weight;

    public int getId() {
        return id;
    }

    public void setId(int id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public double getWeight() {
        return weight;
    }

    public void setWeight(double weight) {
        this.weight = weight;
    }
}

定义生产者，也就是事件的来源。

package com.example.disruptor.singleton;

import com.example.disruptor.Product;
import com.lmax.disruptor.RingBuffer;

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

public class Producer implements Runnable {

    public static final int NUMBER = 10000000;
    private final CountDownLatch latch;
    private AtomicInteger idCount = new AtomicInteger(0);
    private RingBuffer<Product> ringBuffer;

    public Producer(RingBuffer<Product> ringBuffer, CountDownLatch latch) {
        this.ringBuffer = ringBuffer;
        this.latch = latch;
    }


    private void createData() {
        //1.可以把ringBuffer看做一个事件队列，那么next就是得到下面一个事件槽
        long sequence = ringBuffer.next();
        try {
            //2.用上面的索引取出一个空的事件用于填充（获取该序号对应的事件对象）
            Product product = ringBuffer.get(sequence);
            //3.获取要通过事件传递的业务数据
            product.setId(idCount.incrementAndGet());
        } finally {
            //4.发布事件
            //注意，最后的 ringBuffer.publish 方法必须包含在 finally 中以确保必须得到调用；
            // 如果某个请求的 sequence 未被提交，将会堵塞后续的发布操作或者其它的 producer。
            ringBuffer.publish(sequence);
        }
    }

    public void run() {
        for (int i = 0; i < NUMBER; i++) {
            createData();
        }
        //通过latch告诉主线程，完成了产品的生产
        latch.countDown();
    }
}

生产者在生成消息的过程中需要得到Disruptor里的ringBuffer,将生产的消息加入到ringBuffer里。Disruptor 的事件发布过程是一个两阶段提交的过程：　　第一步：先从 RingBuffer 获取下一个可以写入的事件的序号；　　第二步：获取对应的事件对象，将数据写入事件对象；　　第三部：将事件提交到 RingBuffer; 事件只有在提交之后才会通知消息消费者进行处理；

定义消息的消费者，在Disruptor里是EventHandler类型的实例。

package com.example.disruptor.singleton;

import com.example.disruptor.Product;
import com.lmax.disruptor.EventHandler;

import java.util.concurrent.CountDownLatch;

public class Consumer implements EventHandler<Product> {

   private int count = 0;

   private CountDownLatch latch;
    public Consumer(CountDownLatch latch) {
        this.latch = latch;
    }

    public void onEvent(Product event, long sequence, boolean endOfBatch) throws Exception {
       ;count++;
       //通过latch告诉主线程，完成了产品的消费
       if(count == Producer.NUMBER){
           latch.countDown();
       }
    }

    public int getCount() {
        return count;
    }
}

通过Disruptor类，将生产者与消费者进行整合。具体的代码如下：

package com.example.disruptor.singleton;

import com.example.disruptor.Product;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Main {

    //定义ringBuffer的大小
    private static final int RING_BUFFER_SIZE = 1024 * 8;

    public static void main(String[] args) {
        //构造消费者一个线程池， 实际项目中最好不要用Executors来构建
        ExecutorService consumerExecutor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);
        //构造生产者线程池
        ExecutorService produceerExecutor = Executors.newFixedThreadPool(1);
        //创建disruptor
        Disruptor<Product> disruptor =
                new Disruptor<Product>(new EventFactory<Product>() {
                    public Product newInstance() {
                        return new Product();
                    }
                }, RING_BUFFER_SIZE, consumerExecutor, ProducerType.SINGLE, new YieldingWaitStrategy());

        CountDownLatch latch = new CountDownLatch(2);
        // 连接消费事件方法
        Consumer consumer = new Consumer(latch);
        disruptor.handleEventsWith(consumer);
        // 启动
        disruptor.start();
        //生产者开始生产数据
        Producer producer = new Producer(disruptor.getRingBuffer(), latch);
        produceerExecutor.submit(producer);

        try {
            latch.await();
        } catch (InterruptedException e) {
        }

        System.out.println(consumer.getCount());
        //关闭打开的资源
        disruptor.shutdown();
        consumerExecutor.shutdown();
        produceerExecutor.shutdown();
    }
}

在构造Disruptor对象，有几个核心的概念： 1：事件工厂(Event Factory)定义了如何实例化事件(Event)，Disruptor 通过 EventFactory 在 RingBuffer 中预创建 Event 的实例。 2：ringBuffer这个数组的大小，一般根据业务指定成2的指数倍。 3：消费者线程池，事件的处理是在构造的线程池里来进行处理的。 4：指定等待策略，Disruptor 定义了 com.lmax.disruptor.WaitStrategy 接口用于抽象 Consumer 如何等待Event事件。Disruptor 提供了多个 WaitStrategy 的实现，每种策略都具有不同性能和优缺点，根据实际运行环境的 CPU 的硬件特点选择恰当的策略，并配合特定的 JVM 的配置参数，能够实现不同的性能提升。　　BlockingWaitStrategy 是最低效的策略，但其对CPU的消耗最小并且在各种不同部署环境中能提供更加一致的性能表现；　　SleepingWaitStrategy 的性能表现跟 BlockingWaitStrategy 差不多，对 CPU 的消耗也类似，但其对生产者线程的影响最小，适合用于异步日志类似的场景；　　YieldingWaitStrategy 的性能是最好的，适合用于低延迟的系统。在要求极高性能且事件处理线数小于 CPU 逻辑核心数的场景中，推荐使用此策略；例如，CPU开启超线程的特性。

多消费者模型

在生产者消费者模型中，为了防止生产者生产的数据覆盖掉还未消费的数据，Disruptor中每个消费者都各自有个Sequence，而消费者的Sequence状态需要通过SequenceBarrier同步到ringBuffer中。生产者产生数据的Sequence是通过ringBuffer进行生成的。下面是具体的代码：

定义生产者

package com.example.disruptor.mult;

import com.example.disruptor.Product;
import com.lmax.disruptor.RingBuffer;

import java.util.concurrent.atomic.AtomicInteger;

public class Producer implements Runnable {

    public static final int NUMBER = 10000000;
    public static AtomicInteger idCount = new AtomicInteger(0);
    private RingBuffer<Product> ringBuffer;

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


    private void createData() {
        //1.可以把ringBuffer看做一个事件队列，那么next就是得到下面一个事件槽
        long sequence = ringBuffer.next();
        try {
            //2.用上面的索引取出一个空的事件用于填充（获取该序号对应的事件对象）
            Product product = ringBuffer.get(sequence);
            //3.获取要通过事件传递的业务数据
            product.setId(idCount.incrementAndGet());
        } finally {
            //4.发布事件
            //注意，最后的 ringBuffer.publish 方法必须包含在 finally 中以确保必须得到调用；
            // 如果某个请求的 sequence 未被提交，将会堵塞后续的发布操作或者其它的 producer。
            ringBuffer.publish(sequence);
        }
    }

    public void run() {
        for (int i = 0; i < NUMBER; i++) {
            createData();
        }
    }
}

定义消费者

package com.example.disruptor.mult;

import com.example.disruptor.Product;
import com.lmax.disruptor.WorkHandler;

/**
 * 多消费者需要继承自WorkHandler
 */
public class Consumer implements WorkHandler<Product> {


    private int count = 0;

    public Consumer() {
    }


    public void onEvent(Product event) throws Exception {
        count++;
    }

    public int getCount() {
        return count;
    }

}

启动类：

package com.example.disruptor.mult;

import com.example.disruptor.Product;
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

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

public class Main {

    //定义ringBuffer的大小
    private static final int RING_BUFFER_SIZE = 1024 * 8;

    public static void main(String[] args) {
        //线程数
        int processor = Runtime.getRuntime().availableProcessors() * 2;
        //构造消费者一个线程池， 实际项目中最好不要用Executors来构建
        ExecutorService consumerExecutor = Executors.newFixedThreadPool(processor);
        //构造生产者线程池
        ExecutorService produceerExecutor = Executors.newFixedThreadPool(processor);
        //定义一个ringBuffer,也就是相当于一个队列
        RingBuffer ringBuffer = RingBuffer.create(ProducerType.MULTI, new EventFactory<Product>() {
            public Product newInstance() {
                return new Product();
            }
        }, RING_BUFFER_SIZE, new YieldingWaitStrategy());
        //定义一个消费者池，
        Consumer[] consumers = new Consumer[processor];
        for (int i = 0; i < processor; i++) {
            consumers[i] = new Consumer();
        }
        WorkerPool workerPool = new WorkerPool<Product>(ringBuffer,
                ringBuffer.newBarrier(), new IgnoreExceptionHandler(), consumers);
        //每个消费者，也就是 workProcessor都有一个sequence，表示上一个消费的位置,这个在初始化时都是-1
        Sequence[] sequences = workerPool.getWorkerSequences();
        //将其保存在ringBuffer中的 sequencer 中，在为生产申请slot时要用到,也就是在为生产者申请slot时不能大于此数组中的最小值,否则产生覆盖
        ringBuffer.addGatingSequences(sequences);
        //用executor 来启动 workProcessor 线程
        workerPool.start(consumerExecutor);

        //生产者开始生产数据
        for (int i = 0; i < processor; i++) {
            Producer producer = new Producer(ringBuffer);
            produceerExecutor.submit(producer);
        }

        while (true) {
            int count = 0;
            for (Consumer consumer : consumers) {
                count += consumer.getCount();
            }
            System.out.println("生产了多少数据" + Producer.idCount.get());
            System.out.println("消费了多少数据" + count);
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }
}

ringBuffer需要将WorkPool里所有消费者的Sequence加到ringBuffer中，以防止出现数据覆盖的问题。

将disruptor当成JMS，处理消息流

可以将disruptor当成单机版的JMS，用来处理数据流，disruptor提供了消费者处理消息的先后顺序，能很好的实现根据指定规则来实现消息的处理。比如可以将消息形成如下图的数据流：

定义上面四个handler的处理逻辑，我这里只贴出一个类的实现

package com.example.disruptor.complex;


import com.example.disruptor.Product;
import com.lmax.disruptor.EventHandler;

public class StartHandler implements EventHandler<Product> {

    public void onEvent(Product product, long l, boolean b) throws Exception {
        System.out.println("start set name");
        product.setName("start");
    }

}

定义生产都，用于生产消息

package com.example.disruptor.complex;

import com.example.disruptor.Product;
import com.lmax.disruptor.RingBuffer;

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

public class Producer implements Runnable {

    public static final int NUMBER = 2;
    private AtomicInteger idCount = new AtomicInteger(0);
    private RingBuffer<Product> ringBuffer;

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


    private void createData() {
        //1.可以把ringBuffer看做一个事件队列，那么next就是得到下面一个事件槽
        long sequence = ringBuffer.next();
        try {
            //2.用上面的索引取出一个空的事件用于填充（获取该序号对应的事件对象）
            Product product = ringBuffer.get(sequence);
            //3.获取要通过事件传递的业务数据
            product.setId(idCount.incrementAndGet());
        } finally {
            //4.发布事件
            //注意，最后的 ringBuffer.publish 方法必须包含在 finally 中以确保必须得到调用；
            // 如果某个请求的 sequence 未被提交，将会堵塞后续的发布操作或者其它的 producer。
            ringBuffer.publish(sequence);
        }
    }

    public void run() {
        for (int i = 0; i < NUMBER; i++) {
            createData();
        }
    }
}

将disruptor与上面四个handler进行关联

package com.example.disruptor.complex;

import com.example.disruptor.Product;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

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

public class Main {

    //定义ringBuffer的大小
    private static final int RING_BUFFER_SIZE = 1024 * 8;

    public static void main(String[] args) {
        //构造消费者一个线程池， 实际项目中最好不要用Executors来构建
        ExecutorService consumerExecutor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);
        //构造生产者线程池
        ExecutorService produceerExecutor = Executors.newFixedThreadPool(1);
        //创建disruptor
        Disruptor<Product> disruptor =
                new Disruptor<Product>(new EventFactory<Product>() {
                    public Product newInstance() {
                        return new Product();
                    }
                }, RING_BUFFER_SIZE, consumerExecutor, ProducerType.SINGLE, new BlockingWaitStrategy());

        //定义处理消息的handler
        StartHandler start = new StartHandler();
        LeftHandler left = new LeftHandler();
        RightHandler right = new RightHandler();
        EndHandler end = new EndHandler();
        //定义处理消息的顺序
        disruptor.handleEventsWith(start).then(left, right).then(end);

        // 启动
        disruptor.start();
//        //生产者开始生产数据
        Producer producer = new Producer(disruptor.getRingBuffer());
        produceerExecutor.submit(producer);


        //关闭打开的资源
/*        disruptor.shutdown();
        consumerExecutor.shutdown();
        produceerExecutor.shutdown();*/
    }
}

可以看到，disruptor通过提供了then方法来实现消息的先后顺序语义。

本文参与腾讯云自媒体分享计划，分享自作者个人站点/博客。

原始发表：2019.03.25 ，如有侵权请联系 cloudcommunity@tencent.com 删除

编程算法

本文分享自作者个人站点/博客前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体分享计划，欢迎热爱写作的你一起参与！

编程算法

登录后参与评论

0 条评论

热度