Disruptor 高性能环形消息框架

官方文档：Disruptor

1. 简介

Disruptor是一个高性能的互进程（Inter-process）和多线程（Multi-threaded）消息处理库，由LMAX交易所开发，用于在Java虚拟机（JVM）上实现高性能的交换和处理数据。Disruptor的核心目标是提供一种低延迟、高吞吐量的解决方案。

一些关键特性：

Disruptor设计用来减少延迟，因为它避免了使用锁和线程间上下文切换，而是采用了一种基于缓存行（cache line）的设计理念。
通过使用无锁编程和环形缓冲区（Ring Buffer）来实现高效的数据交换，Disruptor能够达到纳秒级别的延迟。
Disruptor的API简洁，易于理解和使用，使得开发者可以快速地集成到现有的系统中。
它支持多种消息处理模式，包括单线程、多线程和多进程处理。
Disruptor基于事件驱动模型，可以处理事件的发布和订阅。
它保证了事件的顺序和一致性，这对于需要顺序处理的业务场景非常重要。
Disruptor可以轻松地扩展以适应不同的处理需求，无论是增加消费者数量还是处理不同类型的事件。
它提供了精细的内存管理策略，包括预分配内存和内存屏障（Memory Barriers）的使用，以确保数据的可见性和一致性。

Disruptor的工作流程大致如下：

生产者（Producer）：向环形缓冲区发布事件。
消费者（Consumer）：从环形缓冲区读取事件并处理。
环形缓冲区（Ring Buffer）：一个固定大小的缓冲区，事件被顺序地写入，并且可以被多个消费者并发读取。
序列屏障（Sequence Barrier）：用于同步生产者和消费者之间的进度，确保消费者不会读取到未完全写入的事件。

2. 比较

与Spring消息监听器、Redis发布/订阅、Guava的EventBus一样，提供生产消息与消费消息的能力，极大的解耦应用程序各模块。

特性/技术	Disruptor	Redis发布订阅	Guava EventBus	Spring消息监听器
设计目的	高性能、低延迟的消息处理	分布式消息传递	简化事件发布和订阅流程	与Spring框架集成的声明式事件处理
性能	极高，纳秒级延迟	较高，受网络延迟影响	适中，适用于中等负载	适中，依赖于Spring事件传播机制
可靠性	高，适合关键任务	较低，消息可能会丢失	适中，需要正确管理订阅和发布	高，Spring框架提供事务支持
易用性	低，需要深入了解并发编程	高，简单的发布订阅模型	高，直观的API和灵活的线程模型	高，Spring框架提供注解支持
分布式支持	否，仅限于单个JVM内部	是，支持跨多个节点和应用的消息传递	否，仅限于单个JVM内部	否，仅限于单个JVM内部（除非结合消息中间件）
线程模型	无锁设计，多线程	多线程，基于发布订阅机制	支持同步和异步事件分发	支持同步和异步事件分发
适用场景	高性能计算，如金融交易系统	分布式系统的消息传递，如微服务架构	简单的事件驱动应用，需要灵活的事件处理	需要Spring框架支持的企业级应用
配置复杂度	高，需要配置事件、工厂和处理器	低，Redis简单配置即可使用	低，通过注解或API简单配置	低，Spring框架自动配置
社区和文档	活跃，由LMAX提供支持	非常活跃，Redis社区广泛支持	活跃，Guava库由Google维护	非常活跃，Spring社区广泛支持
扩展性	高，可以自定义事件和处理器	高，可以与其他Redis特性结合使用	高，可以自定义事件处理逻辑	高，可以自定义事件和监听器
持久性	否，内存中处理，不提供持久化	是，可以结合Redis持久化选项	否，内存中处理，不提供持久化	可以结合数据库事务管理持久化

3. 实例

3.1 添加依赖

<!--        disruptor-->
<dependency>
    <groupId>com.lmax</groupId>
    <artifactId>disruptor</artifactId>
    <version>3.4.2</version>
</dependency>

3.2 消息实体

package org.example.event;

public class TradeEvent {
    private long tradeId;
    private String symbol;
    private double price;
    private int volume;

    // Constructor, getters and setters
    public TradeEvent() {}

    public long getTradeId() {
        return tradeId;
    }

    public void setTradeId(long tradeId) {
        this.tradeId = tradeId;
    }

    public String getSymbol() {
        return symbol;
    }

    public void setSymbol(String symbol) {
        this.symbol = symbol;
    }

    public double getPrice() {
        return price;
    }

    public void setPrice(double price) {
        this.price = price;
    }

    public int getVolume() {
        return volume;
    }

    public void setVolume(int volume) {
        this.volume = volume;
    }
}

3.3 消息实体工厂

import com.lmax.disruptor.EventFactory;


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

3.4 消息处理器

public class TradeEventHandler implements EventHandler<TradeEvent> {
    @Override
    public void onEvent(TradeEvent event, long sequence, boolean endOfBatch) {
        System.out.println(String.format("Trade Event. ID: %d, Symbol: %s, Price: %.2f, Volume: %d",
                event.getTradeId(), event.getSymbol(), event.getPrice(), event.getVolume()));
    }
}

3.5 配置disruptor启动器

package org.example.config;

import com.lmax.disruptor.dsl.Disruptor;
import org.example.event.TradeEvent;
import org.example.event.TradeEventFactory;
import org.example.event.TradeEventHandler;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

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

@Configuration
public class DisruptorConfig {

    @Bean
    public Disruptor<TradeEvent> disruptor() {
        int bufferSize = 1024;
        Executor executor = Executors.newCachedThreadPool();

        Disruptor<TradeEvent> disruptor = new Disruptor<>(
                new TradeEventFactory(), bufferSize, executor);

        disruptor.handleEventsWith(new TradeEventHandler());
        System.out.println("Disruptor created.");
        disruptor.start();
        return disruptor;
    }

}

3.6 测试

curl --request POST \
  --url 'http://localhost:8080/trade?apipost_id=35ef4f1dbd9000' \
  --header 'Accept: */*' \
  --header 'Accept-Encoding: gzip, deflate, br' \
  --header 'Connection: keep-alive' \
  --header 'Content-Type: application/json' \
  --header 'User-Agent: PostmanRuntime-ApipostRuntime/1.1.0' \
  --data '{
    "tradeId":1111,
    "symbol":"测试标记",
    "price":32.3,
    "volume":3
}'

控制台输出

Trade Event. ID: 1111, Symbol: 测试标记, Price: 32.30, Volume: 3