Disruptor-高性能线程消息传递框架

2023-11-18 10:37:16 浏览数 (2)

前言碎语

Disruptor是英国LMAX公司开源的高性能的线程间传递消息的并发框架,和jdk中的BlockingQueue非常类似,但是性能却是BlockingQueue不能比拟的,下面是官方给出的一分测试报告,可以直观的看出两者的性能区别:

Disruptor 项目地址:https://github.com/LMAX-Exchange/disruptor

核心概念?

这么性能炸裂的框架肯定要把玩一番,试用前,我们先了解下disruptor的主要的概念,然后结合楼主的weblog项目(之前使用的BlockingQueue),来实践下

RingBuffer:环形的缓冲区,消息事件信息的载体。曾经 RingBuffer 是 Disruptor 中的最主要的对象,但从3.0版本开始,其职责被简化为仅仅负责对通过 Disruptor 进行交换的数据(事件)进行存储和更新。在一些更高级的应用场景中,Ring Buffer 可以由用户的自定义实现来完全替代。

Event:定义生产者和消费者之间进行交换的数据类型。

EventFactory:创建事件的工厂类接口,由用户实现,提供具体的事件

EventHandler:事件处理接口,由用户实现,用于处理事件。

目前为止,我们了解以上核心内容即可,更多的详情,可以移步wiki文档:https://github.com/LMAX-Exchange/disruptor

核心架构图:

实践Disruptor?

改造boot-websocket-log项目,这是一个典型的生产者消费者模式的实例。然后将BlockingQueue替换成Disruptor,完成功能,有兴趣的可以对比下。

第一步,定义事件类型

代码语言:javascript复制
/**
 * Created by kl on 2018/8/24.
 * Content :进程日志事件内容载体
 */
public class LoggerEvent {

    private LoggerMessage log;

    public LoggerMessage getLog() {
        return log;
    }

    public void setLog(LoggerMessage log) {
        this.log = log;
    }
}

第二步,定义事件工厂

代码语言:javascript复制
/**
 * Created by kl on 2018/8/24.
 * Content :进程日志事件工厂类
 */
public class LoggerEventFactory implements EventFactory{
    @Override
    public LoggerEvent newInstance() {
        return new LoggerEvent();
    }
}

第三步,定义数据处理器

代码语言:javascript复制
/**
 * Created by kl on 2018/8/24.
 * Content :进程日志事件处理器
 */
@Component
public class LoggerEventHandler implements EventHandler{

    @Autowired
    private SimpMessagingTemplate messagingTemplate;

    @Override
    public void onEvent(LoggerEvent stringEvent, long l, boolean b) {
        messagingTemplate.convertAndSend("/topic/pullLogger",stringEvent.getLog());
    }
}

第四步,创建Disruptor实操类,定义事件发布方法,发布事件

代码语言:javascript复制
/**
 * Created by kl on 2018/8/24.
 * Content :Disruptor 环形队列
 */
@Component
public class LoggerDisruptorQueue {

    private Executor executor = Executors.newCachedThreadPool();

    // The factory for the event
    private LoggerEventFactory factory = new LoggerEventFactory();

    private FileLoggerEventFactory fileLoggerEventFactory = new FileLoggerEventFactory();

    // Specify the size of the ring buffer, must be power of 2.
    private int bufferSize = 2 * 1024;

    // Construct the Disruptor
    private Disruptordisruptor = new Disruptor<>(factory, bufferSize, executor);;

    private DisruptorfileLoggerEventDisruptor = new Disruptor<>(fileLoggerEventFactory, bufferSize, executor);;

    private static  RingBufferringBuffer;

    private static  RingBufferfileLoggerEventRingBuffer;

    @Autowired
    LoggerDisruptorQueue(LoggerEventHandler eventHandler,FileLoggerEventHandler fileLoggerEventHandler) {
        disruptor.handleEventsWith(eventHandler);
        fileLoggerEventDisruptor.handleEventsWith(fileLoggerEventHandler);
        this.ringBuffer = disruptor.getRingBuffer();
        this.fileLoggerEventRingBuffer = fileLoggerEventDisruptor.getRingBuffer();
        disruptor.start();
        fileLoggerEventDisruptor.start();
    }

    public static void publishEvent(LoggerMessage log) {
        long sequence = ringBuffer.next();  // Grab the next sequence
        try {
            LoggerEvent event = ringBuffer.get(sequence); // Get the entry in the Disruptor
            // for the sequence
            event.setLog(log);  // Fill with data
        } finally {
            ringBuffer.publish(sequence);
        }
    }

    public static void publishEvent(String log) {
        if(fileLoggerEventRingBuffer == null) return;
        long sequence = fileLoggerEventRingBuffer.next();  // Grab the next sequence
        try {
            FileLoggerEvent event = fileLoggerEventRingBuffer.get(sequence); // Get the entry in the Disruptor
            // for the sequence
            event.setLog(log);  // Fill with data
        } finally {
            fileLoggerEventRingBuffer.publish(sequence);
        }
    }

}

文末结语

以上四步已经完成了Disruptor的使用,启动项目后就会不断的发布日志事件,处理器会将事件内容通过websocket传送到前端页面上展示,

boot-websocket-log项目地址:https://gitee.com/kailing/boot-websocket-log

Disruptor是高性能的进程内线程间的数据交换框架,特别适合日志类的处理。Disruptor也是从https://github.com/alipay/sofa-tracer了解到的,这是蚂蚁金服 团队开源的分布式链路追踪项目,其中日志处理部分就是使用了Disruptor。

0 人点赞