两个例子带你入门Disruptor

举报
勇哥java实战分享 发表于 2023/09/20 19:36:42 2023/09/20
【摘要】 Disruptor 是英国外汇交易公司 LMAX 开发的一个高性能队列。很多知名开源项目里,比如 canal、log4j2、 storm 都是用了 Disruptor 以提升系统性能 。这篇文章,我们通过两个例子一步一个脚印帮助同学们入门 Disruptor 。 1 环形缓冲区下图展示了 Disruptor 的流程图 。和线程池机制非常类似, Disruptor 也是非常典型的生产者/消费者...

Disruptor 是英国外汇交易公司 LMAX 开发的一个高性能队列。很多知名开源项目里,比如 canal、log4j2、 storm 都是用了 Disruptor 以提升系统性能 。

这篇文章,我们通过两个例子一步一个脚印帮助同学们入门 Disruptor 。

1 环形缓冲区

下图展示了 Disruptor 的流程图 。

和线程池机制非常类似, Disruptor 也是非常典型的生产者/消费者模式。线程池存储提交任务的容器是阻塞队列,而 Disruptor 使用的是环形缓冲区 RingBuffer

环形缓冲区的设计相比阻塞队列有如下优点:

  • 环形数组结构

为了避免垃圾回收,采用数组而非链表。同时,数组对处理器的缓存机制更加友好。

  • 元素位置定位

数组长度2^n,通过位运算,加快定位的速度。下标采取递增的形式,不用担心 index 溢出的问题。index 是 long 类型,即使100万QPS的处理速度,也需要30万年才能用完。

  • 无锁设计

每个生产者或者消费者线程,会先申请可以操作的元素在数组中的位置,申请到之后,直接在该位置写入或者读取数据。

此刻大家并不需要理解环形缓冲区的读写机制,只需要明白 环形缓冲区 RingBuffer 是 Disruptor 的精髓即可。

2 写一个Hello world

我们写一个非常简单的例子:生产者传递一个单一的长整型值给消费者,而消费者将简单地打印出这个值

2.1 添加依赖

<dependency>
    <groupId>com.lmax</groupId>
    <artifactId>disruptor</artifactId>
    <version>3.3.6</version>
</dependency>

2.2 定义事件

首先,我们将定义一个事件(Event),它将携带数据,并且在接下来的所有示例中都是通用的。

public class LongEvent {

    private long value;

    public void set(long value) {
        this.value = value;
    }

    @Override
    public String toString() {
        return "LongEvent{" + "value=" + value + '}';
    }
}

为了让 Disruptor 为我们预分配这些事件,我们需要一个EventFactory 来执行构造。这可以是一个方法引用,比如 LongEvent::new ,或者是 EventFactory 接口的显式实现:

public class LongEventFactory implements EventFactory<LongEvent> {
    @Override
    public LongEvent newInstance() {
        return new LongEvent();
    }
}

2.3 定义消费者

定义了事件,我们需要创建一个消费者来处理这些事件。我们会创建一个事件处理器(EventHandler),它会将把值打印到控制台上。

public class LongEventHandler implements EventHandler<LongEvent> {
    @Override
    public void onEvent(LongEvent longEvent, long sequence, boolean endOfBatch) throws Exception {
          System.out.println("currentThread:" + Thread.currentThread().getName() + " Event: " + longEvent);
    }
}

2.4 发布

public class LongEventMain {
    public static void main(String[] args) throws Exception {
        int bufferSize = 2;
        Disruptor<LongEvent> disruptor =
                new Disruptor<>(
                        new LongEventFactory(),
                        bufferSize,
                        DaemonThreadFactory.INSTANCE,
                        ProducerType.SINGLE,
                        new BlockingWaitStrategy());
        disruptor.handleEventsWith(new LongEventHandler());
        disruptor.start();

        RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
        ByteBuffer bb = ByteBuffer.allocate(8);
        for (long l = 0; true; l++) {
            bb.putLong(0, l);
            ringBuffer.publishEvent((event, sequence, buffer) -> event.set(buffer.getLong(0)), bb);
            Thread.sleep(1000);
        }
    }
}

整个发布流程分为四个部分:

  1. 指定环形缓冲区的大小,必须是2的幂次方,例子中设置的值是 1024 ;

  2. 构建 Disruptor ,参数分别是事件工厂EventFactory环形缓冲区的大小ringBufferSize处理器线程池生产者类型(单生产者/多生产者)、消费者阻塞策略

  3. 定义事件处理器eventHandler,我们这里的逻辑是打印数据打印在控制台;

  4. 启动 Disruptor,从 Disruptor 中获取环形缓冲区ringBuffer,在 for 循环里 ,调用环形队列的publishEvent方法。

    这里使用了 ByteBuffer 做为数据的存储容器 , 方便作为参数传递。

我们来看下执行结果 :

3 日志处理

3.1 应用场景

上面的例子比较简单,但假如要应用到生产环境,就显得非常粗糙。

我们模拟一个日志处理的场景,用户进入视频播放页面,浏览器定时的发送浏览日志到服务端,服务端将日志存储起来。

3.2 核心类设计

我们定义一个 DisruptorManager 管理器 , 管理器包含三个核心参数:消费者监听器 DataEventListener消费者数量环形队列长度

public class DisruptorManager<T> {

    private static final Integer DEFAULT_CONSUMER_SIZE = 4;

    public static final Integer DEFAULT_SIZE = 4096 << 1 << 1;

    private DataEventListener<T> dataEventListener;

    private DisruptorProducer<T> producer;

    private int ringBufferSize;

    private int consumerSize;

    public DisruptorManager(DataEventListener<T> dataEventListener) {
        this(dataEventListener, DEFAULT_CONSUMER_SIZE, DEFAULT_SIZE);
    }

    public DisruptorManager(DataEventListener<T> dataEventListener, final int consumerSize, final int ringBufferSize) {
        this.dataEventListener = dataEventListener;
        this.ringBufferSize = ringBufferSize;
        this.consumerSize = consumerSize;
    }

    public void start() {
        EventFactory<DataEvent<T>> eventFactory = new DisruptorEventFactory<>();
        Disruptor<DataEvent<T>> disruptor = new Disruptor<>(
                eventFactory,
                ringBufferSize,
                DisruptorThreadFactory.create("consumer-thread", false),
                ProducerType.MULTI,
                new BlockingWaitStrategy()
        );
        DisruptorConsumer<T>[] consumers = new DisruptorConsumer[consumerSize];
        for (int i = 0; i < consumerSize; i++) {
            consumers[i] = new DisruptorConsumer<>(dataEventListener);
        }
        disruptor.handleEventsWithWorkerPool(consumers);
        disruptor.start();
        RingBuffer<DataEvent<T>> ringBuffer = disruptor.getRingBuffer();
        this.producer = new DisruptorProducer<>(ringBuffer, disruptor);
    }

    public DisruptorProducer getProducer() {
        return this.producer;
    }

}

首先和 Hello world 代码中的不同的点,Disruptor 的构造函数中我们自定义了消费者的处理器线程。

DisruptorThreadFactory.create("consumer-thread", false),

然后我们定义消费者的业务逻辑 :

DisruptorConsumer<T>[] consumers = new DisruptorConsumer[consumerSize];
for (int i = 0; i < consumerSize; i++) {
    consumers[i] = new DisruptorConsumer<>(dataEventListener);
}
disruptor.handleEventsWithWorkerPool(consumers);

消费者本质上是workHandler的实现类,只不过初始化时将 DataEventListener 作为构造函数的参数。

public class DisruptorConsumer<T> implements WorkHandler<DataEvent<T>> {

    private DataEventListener<T> dataEventListener;

    public DisruptorConsumer(DataEventListener dataEventListener) {
        this.dataEventListener = dataEventListener;
    }

    @Override
    public void onEvent(DataEvent<T> dataEvent) throws Exception {
        if (dataEvent != null) {
            dataEventListener.processDataEvent(dataEvent);
        }
    }

}

因为我们是希望线程池并行的处理这些消息数据,使用的是disruptor.handleEventsWithWorkerPool 可以保证每个事件只会由一个工作处理器处理

在 springboot 项目中,我们需要初始化相关 bean。

@Configuration
@AutoConfigureBefore(RedisConfig.class)
public class DisruptorConfig {

    private final static Logger logger = LoggerFactory.getLogger(DisruptorConfig.class);

    private final static String LIST_KEY = "disruptor:list";

    @Autowired
    private RedissonClient redissonClient;

    @Bean
    public DataEventListener<String> createConsumerListener() {
        DataEventListener<String> dataEventListener = new DataEventListener<String>() {
            @Override
            public void processDataEvent(DataEvent<String> dataEvent) throws InterruptedException {
                logger.info("processDateEvent data:" + dataEvent.getData());
                redissonClient.getList(LIST_KEY).add(dataEvent.getData());
            }
        };
        return dataEventListener;
    }

    @Bean
    public DisruptorProducer<String> createProducer(DataEventListener dataEventListener) {
        DisruptorManager disruptorManage = new DisruptorManager(dataEventListener,
                8,
                1024 * 1024);
        disruptorManage.start();
        return disruptorManage.getProducer();
    }

首先,我们定义好消费者的事件监听器,然后定义 DisruptorProducer, 该类用来将数据提交到环形队列。

public class DisruptorProducer<T> {

    private final Logger logger = LoggerFactory.getLogger(DisruptorProducer.class);

    private final RingBuffer<DataEvent<T>> ringBuffer;

    private final Disruptor<DataEvent<T>> disruptor;

    private final EventTranslatorOneArg<DataEvent<T>, T> translatorOneArg = (event, sequence, t) -> event.setData(t);

    public DisruptorProducer(final RingBuffer<DataEvent<T>> ringBuffer, final Disruptor<DataEvent<T>> disruptor) {
        this.ringBuffer = ringBuffer;
        this.disruptor = disruptor;
    }

    /**
     * Send a data.
     *
     * @param data the data
     */
    public void onData(final T data) {
        try {
            ringBuffer.publishEvent(translatorOneArg, data);
        } catch (Exception ex) {
            logger.error("publish event error:", ex);
        }
    }

    public void shutdown() {
        if (null != disruptor) {
            disruptor.shutdown();
        }
    }

}

最后,在控制器中,接收前端请求:

@Autowired
private DisruptorProducer<String> producer;

@GetMapping("/pushlog")
public ResponseEntity pushlog(String log) {
    producer.onData(log);
    return ResponseEntity.successResult(null);
}

从下图中,我们可以看到从控制器接收到请求后,消费处理器线程不断地将数据打印出来,并且发送到队列中。

4 写到最后

日志处理的例子里,我们试图封装 Disruptor 相关 API ,以便在 springboot 项目中更方便的使用。

笔者在测试过程时,发现若消费逻辑慢的时候,生产者发送数据事件时,可能会阻塞。

为什么生产者会阻塞,Disruptor 的核心原理是什么 ,如何使用 Disruptor 的高级特性顺序依赖执行 ?

正因为有这些疑问,笔者觉得深入理解 Disruptor 原理特别有必要,笔者也会在接下来的文章里一一为大家答疑解惑。


参考资料:

https://lmax-exchange.github.io/disruptor/disruptor.html

https://zhuanlan.zhihu.com/p/45575008

【版权声明】本文为华为云社区用户原创内容,转载时必须标注文章的来源(华为云社区)、文章链接、文章作者等基本信息, 否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

评论(0

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。