Disruptor-高性能队列


Disruptor是什么?

官网介绍:https://github.com/LMAX-Exchange/disruptor/wiki(https://github.com/LMAX-Exchange/disruptor/wiki)

Disruptor是一个无锁的并发框架

Martin Fowler在自己网站上写了一篇LMAX架构的文章,在文章中他介绍了LMAX是一种新型零售金融交易平台,它能够以很低的延迟产生大量交易。这个系统是建立在JVM平台上,其核心是一个业务逻辑处理器,它能够在一个线程里每秒处理6百万订单。业务逻辑处理器完全是运行在内存中,使用事件源驱动方式。业务逻辑处理器的核心是Disruptor。

特点:

  • 大大简化了并发程序开发的难度,性能上比Java本身提供的一些并发包好。
  • 是一个高性能异步处理框架,实现了观察者模式。
  • 是无锁的,CPU友好。它不会清除缓存中的数据,只会覆盖,降低了垃圾回收机制的启动频率。
  • 业务逻辑是纯内存操作,使用事件驱动方式。

Disruptor核心

Disruptor核心是一个RingBuffer

  • RingBuffer是一个数组,没有首尾指针。
  • RingBuffer是一个首尾相接的环,用于不同线程间传递数据的Buffer。
  • RingBuffer有一个序号,这个序号指向数组中下一个可用的元素。
  • 随着数据不断的填充这个数组,这个序号会一直增长,直到绕过这个环。
  • 前序号指向的元素,可以通过mod计算:序号%长度=索引。
  • 建议将长度设置为2的N次方,有利于二进制计算:序号&(长度-1)=索引。

如果RingBuffer满了,那么是继续覆盖,还是等待消费,这个是由生产者和消费者决定的。

假设RingBuffer满了,生产者有两个选择

  1. 等待RingBuffer有空位在填充
  2. 直接覆盖

同时消费者也可以做出选择

  1. 等待RingBuffer满了再消费
  2. RingBuffer填充一个就消费一个

Sequence

  • 通过顺序递增的序号来编号,管理正在进行交换的数据(事件)
  • 对数据处理的过程总是沿着需要逐个递增处理,实现线程安全
  • 一个Sequence用于跟踪标识某个特定的事件处理者的处理进度
    • 生产者和消费者都各自拥有各自的Sequence
    • 如果多个生产者和多个消费者,他们每一个都会拥有各自的Sequence
  • Sequence可以看成是一个volatile long 类型的数值

Sequencer

  • Sequencer是Disruptor高性能的核心
  • 他是一个interface,主要实现生产者和消费者的快速、正确的传递数据的并发算法

SequenceBarrier

  • 用于保持RingBuffer的生产者和消费者之间的平衡关系
  • 决定消费者是否还有可处理事件的逻辑

EventProcessor

主要事件循环,处理Disruptor中的Event,拥有消费者的EventProcessor

Disruptor Quick Start

官网参考:https://github.com/LMAX-Exchange/disruptor/wiki/Getting-Started(https://github.com/LMAX-Exchange/disruptor/wiki/Getting-Started)

开发模型

  1. 定义Event,代表Disruptor所处理的数据单元
    /**
     * @author fangxi
     */
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public class OrderEvent {
    
        private Long id;
        private String productName;
        private BigDecimal price;
    
    }
  1. 定义Event工厂,实现EventFactocy<?>接口,用来填充RingBuffer容器,如果使用1.8的话,则不需要这个工厂对象,直接在构造方法里面OrderEvent::new
/**
 * @author fangxi
 * <p>
 * 创建orderEvent对象
 */
public class OrderEventFactory implements EventFactory<OrderEvent> {

    @Override
    public OrderEvent newInstance() {
        return new OrderEvent();
    }

}
  1. 定义Event处理器(消费者),实现EventHandler<?>,用来从RingBuffer中取出数据并处理。如果使用1.8处理,可以不需要这个类,用lambda代替
/**
 * @author fangxi
 * 消费者
 */
public class OrderEventHandler implements EventHandler<OrderEvent> {

    /**
     * 有消息发送,这个就会被监听
     */
    @Override
    public void onEvent(OrderEvent orderEvent,long sequence, boolean endOfBatch) throws Exception {
        System.out.println(orderEvent);
    }
}
  1. 组合1-3步
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.util.DaemonThreadFactory;

import java.math.BigDecimal;

/**
 * @author fangxi
 */
public class DisruptorTest {

    public static void main(String[] args) {
        int bufferSize = 1024;
        // 实例化Disruptor对象
        Disruptor<OrderEvent> disruptor = new Disruptor<>(OrderEvent::new, bufferSize, DaemonThreadFactory.INSTANCE);
        // 添加消费者监听
        // disruptor.handleEventsWith(new OrderEventHandler());
        // 消费者,lambda表示
        disruptor.handleEventsWith((event, sequence, endOfBatch) -> System.out.println("消费者:" + event));
        // 启动
        // 实际存储数据的容器
        RingBuffer<OrderEvent> ringBuffer = disruptor.start();
        // 投递100条数据
        for (int i = 0; i < 100; i++) {
            // 生产者,就用lambda表示
            ringBuffer.publishEvent((event, sequence) -> {
                event.setId(sequence);
                event.setPrice(new BigDecimal("100").add(new BigDecimal(sequence)));
                event.setProductName(sequence + "号商品");
            });
        }
        disruptor.shutdown();
    }

}

Disruptor高阶操作

涉及到的Event对象

@Data
public class Trade {

    private String id;
    private String name;
    private String nickname;
    private BigDecimal price;

}

串形操作和并行操作

让消费者按照我们制定的顺序操作

先设置三个消费者

// 三个消费者,将分配设置name,id和打印
EventHandler<Trade> nameHandler = (trade, sequence, endOfBatch) -> {
    System.out.println("Set Name");
    trade.setName("name" + sequence);
		// 模拟耗时操作
    TimeUnit.SECONDS.sleep(1);
};
EventHandler<Trade> idHandler = (trade, sequence, endOfBatch) -> {
    System.out.println("Set Id");
    trade.setId(UUID.randomUUID().toString());
    TimeUnit.SECONDS.sleep(1);
};
EventHandler<Trade> printHandler = (trade, sequence, endOfBatch) -> {
    System.out.println(trade);
};

串形操作

// 串形操作,按照顺序操作
disruptor.handleEventsWith(idHandler).handleEventsWith(nameHandler).handleEventsWith(printHandler);

串形操作打印的结果和handleEventsWith的链式调用顺序有关

并行操作

// 并行操作,三个handler并行执行
disruptor.handleEventsWith(idHandler, nameHandler, printHandler);

生产者

Random random = new Random();
// 也可以直接通过Disruptor对象提交任务
disruptor.publishEvent((trade, sequence) -> {
    double price = random.nextDouble() * 9999;
    trade.setPrice(new BigDecimal(String.valueOf(price)));
});

菱形操作

让串行操作和并行操作同时存在

上面的例子中,可以设置idHandler和设置nameHandler可以同时执行,让printHandler在这两个之后执行

// 菱形操作,让idHandler和nameHandler同时执行,printHandler在这两个操作之后在执行
disruptor.handleEventsWith(idHandler, nameHandler).handleEventsWith(printHandler);
// 也可以这样写
disruptor.handleEventsWith(idHandler, nameHandler).then(printHandler);

多边形操作

实现多边形操作,需要再新增两个handler

// 新增两个handler
EventHandler<Trade> nicknameHandler = (trade, sequence, endOfBatch) -> {
    System.out.println("nicknameHandler");
    trade.setNickname("[" + trade.getName() + "]");
};
EventHandler<Trade> idNotLineHandler = (trade, sequence, endOfBatch) -> {
    System.out.println("idNotLineHandler");
    trade.setId(trade.getId().replaceAll("-", ""));
};

要实现一下的操作

// idHander和nameHandler并行
disruptor.handleEventsWith(idHandler, nameHandler);
// idHander之后执行idNotLineHandler
disruptor.after(idHandler).handleEventsWith(idNotLineHandler);
// nameHandler之后执行nicknameHandler
disruptor.after(nameHandler).handleEventsWith(nicknameHandler);
// 在idNotLineHandler, nicknameHandler之后执行printHandler
disruptor.after(idNotLineHandler, nicknameHandler).handleEventsWith(printHandler);

Author: Re:0
Reprint policy: All articles in this blog are used except for special statements CC BY 4.0 reprint policy. If reproduced, please indicate source Re:0 !
 Previous
基于RPC框架的通用代码实现 基于RPC框架的通用代码实现
基于RPC框架的通用代码实现在业务代码中,会经常有大量的CRUD这种枯燥无味我代码需要我们去编写。当然,也有现成的工具,比如mybatis generator等工具,让我们脱离手动编写这些代码。但这些仅仅只是针对mapper层,虽然Serv
2023-01-19
Next 
HDFS HDFS
HDFS产出背景及定义随着数据量的越来越大,在一个操作系统存不下所有的数据,那么就分配更多的操作系统管理的磁盘中,但是不大方便维护,迫切需要一种系统来管理多台机器上的文件,这就是分布式文件系统的。HDFS只是分布式文件系统中的一种。
2022-09-06
  TOC