Spring Boot3.x集成Disruptor4.0

Disruptor介绍

Disruptor是一个高性能内存队列，研发的初衷是解决内存队列的延迟问题(在性能测试中发现竟然与I/O操作处于同样的数量级)。基于Disruptor开发的系统单线程能支撑每秒600万订单，2010年在QCon演讲后，获得了业界关注。2011年，企业应用软件专家Martin Fowler专门撰写长文介绍。同年它还获得了Oracle官方的Duke大奖。

Disruptor 是一个 Java 的并发编程框架，大大的简化了并发程序开发的难度，在性能上也比 Java 本身提供的一些并发包要好。它源于LMAX对并发性、性能和非阻塞算法的研究，如今构成了其Exchange基础架构的核心部分。

Disruptor的队列功能和传统的MQ队列服务不同（比如：kafka\rabbitMq等等），Disruptor而是一个基于JDK的高性能内存队列，例如与Java的BlockingQueue进行对比。与队列一样，Disruptor的目的是在同一进程内的线程之间传递数据（例如消息或事件）。

目前，包括Apache Storm、Camel、Log4j 2在内的很多知名项目都应用了Disruptor以获取高性能。在美团技术团队它也有不少应用，有的项目架构借鉴了它的设计机制。

Disruptor功能

高性能消息传递：Disruptor 能够通过避免锁和减少线程间的数据交换来提高性能。
支持多生产者和多消费者：可以由多个生产者向队列中添加事件，同时多个消费者处理这些事件。
事件处理模型：Disruptor 使用预分配事件的环形数组结构，每个事件槽可以被重复使用，减少了对象创建的开销。
内存屏障优化：利用内存屏障来减少不必要的CPU缓存刷新，提高效率。

Disruptor优点

极高的吞吐量和低延迟：通过减少锁的使用和优化内存操作，Disruptor 能够实现极高的数据处理速率和低延迟。
避免了线程阻塞：使用无锁的设计，避免了传统队列中的线程阻塞问题。
资源利用率高：通过重复使用事件对象，减少了垃圾回收的压力。

Disruptor缺点

复杂性：Disruptor 的使用和理解比标准的队列或者其他并发模型要复杂，需要更多的学习和调试。
适用场景有限：主要适用于需要极高性能和低延迟的系统，对于一般的应用场景可能是过度设计。
调试困难：由于其无锁的设计和复杂的内部结构，当出现问题时，调试可能比较困难。
总的来说，Disruptor 是一个专为高性能计算设计的工具，适用于那些对性能有极端要求的场景。对于普通应用或者数据量不大的情况，使用传统的并发模型可能更为合适。

Disruptor特征

Disruptor的目标之一是在低延迟环境中使用，在低延迟系统中，必须减少或移除内存分配；

在基于Java的系统中，目的是减少由于垃圾收集导致的系统停顿；为了支持这一点，用户可以预先分配Disruptor中事件所需的存储空间（也就是声明RingBuffer的大小）。

在构造RingBuffer期间，EventFactory由用户提供，并将在Disruptor的Ring Buffer中每个事件元素创建时候被调用。将新数据发布到Disruptor时，API将允许用户获取构造的对象，以便他们可以调用方法或更新该存储对象上的字段，Disruptor保证这些操作只要正确实现就是并发安全的。

官方文档

github开源地址

GitHub - LMAX-Exchange/disruptor: High Performance Inter-Thread Messaging Library

github介绍文档

LMAX Disruptor

开发示例

我们以一个项目来演示，开发一个订单业务消息处理服务，来模拟采用Disruptor队列来对订单进行管理；

采用一个生产者，多个消费者模式，并且多个消费者按不同的顺序进行链路排例，对生产者消息进行消费；

如：

某电商平台存在以下服务功能

订单管理服务：生成订单后，过行订单管理与跟踪
用户等级服务：用户购买商品后，重新评估用户星级等级，提供对标服务
电商客服服务：负责售前售后服务，有3组，分别为A组、B组、C组，每个订单只分配到其中一组提供对接客服服务
仓储管理服务：管理平台所有商品仓库，并提供已购买商品
物流投递服务：负责从仓储中获取商品，投送到用户手中

电商平台每完成一笔支付订单，将消息发送到此Disruptor示例服务。

首先分别经过订单管理服务和用户等级服务，注意：两个服务均为独立消费消息
完成前置两个服务消费后，才能执行电商客服A组、电商客服B组、电商客服C组其中任意一个服务独立消费消息，注意：三选一，不能全部执行
完成前置电商客服服务消费后，执行仓储管理服务消费消息
完成前置仓储管理服务消费后，最后执行物流投递服务

消费链路流程如下：

工程环境

JDK：17
SpringBoot：3.1.3-SNAPSHOT

注：假设你已创建基础工程，并完成SpringBoot组件引入

引入Disruptor依赖

<dependencies>
    <dependency>
        <groupId>com.lmax</groupId>
        <artifactId>disruptor</artifactId>
        <version>4.0.0</version>
    </dependency>
</dependencies>

项目Disruptor配置

package com.example.disruptor;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.web.reactive.function.server.RequestPredicate;
import org.springframework.web.reactive.function.server.RouterFunction;
import org.springframework.web.reactive.function.server.ServerResponse;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.*;
import static org.springframework.web.reactive.function.server.RequestPredicates.accept;
import static org.springframework.web.reactive.function.server.RouterFunctions.route;

/**
 * @Description 服务配置类
 */
@Slf4j
@Configuration(proxyBeanMethods = false)
public class WebConfig {

    private static final RequestPredicate JSON_ACCEPT = accept(new MediaType(MediaType.APPLICATION_JSON, StandardCharsets.UTF_8));
    //缓冲区大小，必需是2的N次方
    final static int BUFFER_SIZE = 1024 * 1024;

    /**
     * 创建基于环的可重用队列存储，实现消息数据存储与推送到处理器执行
     * @return
     */
    @Bean("orderRingBuffer")
    public RingBuffer<OrderEvent> createRingBuffer(){
        // 创建消息处理器，注：正常业务模式下，由不同的业务类实现；此处为了简化演示，从createEventHandler()方法中获取模拟实现类；
        EventHandler<OrderEvent> orderHandler = createEventHandler("消息序例:{}, 发送到《订单管理服务》, 订单详情:{}");
        EventHandler<OrderEvent> userLevelHandler = createEventHandler("消息序例:{}, 发送到《用户等级服务》, 订单详情:{}");
        EventHandler<OrderEvent> customerService0Handler = createEventHandler(0, 3, "消息序例:{}, 发送到《电商客服A组》, 订单详情:{}");
        EventHandler<OrderEvent> customerService1Handler = createEventHandler(1, 3, "消息序例:{}, 发送到《电商客服B组》, 订单详情:{}");
        EventHandler<OrderEvent> customerService2Handler = createEventHandler(2, 3, "消息序例:{}, 发送到《电商客服C组》, 订单详情:{}");
        EventHandler<OrderEvent> storageHandler = createEventHandler("消息序例:{}, 发送到《仓储管理服务》, 订单详情:{}");
        EventHandler<OrderEvent> deliveryHandler = createEventHandler("消息序例:{}, 发送到《物流投递服务》, 订单详情:{}");

        //创建环形缓冲区处理器事件生成器
        Disruptor<OrderEvent> disruptor = new Disruptor<OrderEvent>(OrderEvent::new,
                //缓冲区大小
                BUFFER_SIZE,
                //默认线程工厂
                Executors.defaultThreadFactory(),
                //ProducerType.SINGLE（表示生产者只有一个）和ProducerType.MULTY（表示有多个生产者）
                ProducerType.SINGLE,
                /**
                 可用事件策略：
                 BlockingWaitStrategy：用了ReentrantLock的等待&&唤醒机制实现等待逻辑，是默认策略，比较节省CPU
                 BusySpinWaitStrategy：持续自旋，JDK9之下慎用（最好别用）
                 DummyWaitStrategy：返回的Sequence值为0，正常环境是用不上的
                 LiteBlockingWaitStrategy：基于BlockingWaitStrategy，在没有锁竞争的时候会省去唤醒操作，但是作者说测试不充分，不建议使用
                 TimeoutBlockingWaitStrategy：带超时的等待，超时后会执行业务指定的处理逻辑
                 LiteTimeoutBlockingWaitStrategy：基于TimeoutBlockingWaitStrategy，在没有锁竞争的时候会省去唤醒操作
                 SleepingWaitStrategy：三段式，第一阶段自旋，第二阶段执行Thread.yield交出CPU，第三阶段睡眠执行时间，反复的的睡眠
                 YieldingWaitStrategy：二段式，第一阶段自旋，第二阶段执行Thread.yield交出CPU
                 PhasedBackoffWaitStrategy：四段式，第一阶段自旋指定次数，第二阶段自旋指定时间，第三阶段执行Thread.yield交出CPU，第四阶段调用成员变量的waitFor方法，这个成员变量可以被设置为BlockingWaitStrategy、LiteBlockingWaitStrategy、SleepingWaitStrategy这三个中的一个
                 注意：
                 BlockingWaitStrategy 是最低效的策略，但其对CPU的消耗最小并且在各种不同部署环境中能提供更加一致的性能表现
                 SleepingWaitStrategy 的性能表现跟BlockingWaitStrategy差不多，对CPU的消耗也类似，但其对生产者线程的影响最小，适合用于异步日志类似的场景
                 YieldingWaitStrategy 的性能是最好的，适合用于低延迟的系统。在要求极高性能且事件处理线数小于CPU逻辑核心数的场景中，推荐使用此策略；例如，CPU开启超线程的特性
                 */
                new YieldingWaitStrategy());

        //消息消费配置处理器执行链路：orderHandler, userLevelHandler》customerService[0~2]Handler》storageHandler》deliveryHandler
        //分别独立执行：orderHandler（订单管理服务）, userLevelHandler（用户等级服务）
        disruptor.handleEventsWith(orderHandler, userLevelHandler)
                //前置消费后，再执行其中任意一个处理器：customerService[0~2]Handler（电商客服A组\电商客服B组\电商客服C组,三选一）
                .then(customerService0Handler, customerService1Handler, customerService2Handler)
                //前置消费后，再执行处理器：storageHandler（仓储管理服务）
                .then(storageHandler)
                //前置消费后，再执行处理器：deliveryHandler（物流投递服务）
                .then(deliveryHandler);

        //启动disruptor服务
        disruptor.start();
        return disruptor.getRingBuffer();
    }

    /**
     * 创建消息处理器
     * @param msg
     * @return
     */
    private EventHandler<OrderEvent> createEventHandler(final String msg){
        return (event, sequence, endOfBatch)->{
            log.info(msg, sequence,event);
            //业务逻辑代码...
        };
    }

    /**
     * 创建消息处理器,支持相同处理器多选一（取模计算）
     * @param index
     * @param handlerCount
     * @param msg
     * @return
     */
    private EventHandler<OrderEvent> createEventHandler(final int index, final int handlerCount, final String msg){
        return (event, sequence, endOfBatch)->{
            if (sequence % handlerCount == index) {
                log.info(msg, sequence, event);
                //业务逻辑代码...
            }
        };
    }

    /**
     * 后台服务请求router
     * @param orderRingBuffer
     * @return
     */
    @Bean
    public RouterFunction<ServerResponse> webRouterFunction(RingBuffer<OrderEvent> orderRingBuffer) {
        return route()
                .POST("/order/push", JSON_ACCEPT, (request)->{
                    String orderId = request.queryParam("orderId").orElse("");
                    String name = request.queryParam("name").orElse("");
                    String price = request.queryParam("price").orElse("");
                    orderRingBuffer.publishEvent((orderEvent, sequence) -> {
                        orderEvent.setOrderId(orderId);
                        orderEvent.setName(name);
                        orderEvent.setPrice(Double.parseDouble(price));
                    });
                    return ServerResponse.status(HttpStatus.OK).bodyValue("ok,200");
                })
                .build();
    }
}

订单对象

package com.example.disruptor;

import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@NoArgsConstructor
public class OrderEvent {
    private String orderId;
    private String name;
    private Double price;
}

启动类

package com.example;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class DisruptorSamplesApplication {
    public static void main(String[] args) {
        SpringApplication.run(DisruptorSamplesApplication.class, args);
    }
}

YML配置文件

# 本地服务访问
server:
  # 服务端口
  port: 8080
  # 服务IP
  address: 0.0.0.0
# 配置日志
logging:
  level:
    org.springframework: info
# 开启debug模式
debug: false

工程测试

Postman发送POST请求，将模拟表单数据提交到服务端；

服务打印日志

2024-04-30T18:08:05.227+08:00  INFO 39828 --- [pool-4-thread-1] com.example.disruptor.WebConfig: 消息序例:0, 发送到《订单管理服务》, 订单详情:OrderEvent(orderId=VL-20240495001, name=苹果1斤, price=7.65)
2024-04-30T18:08:05.227+08:00  INFO 39828 --- [pool-4-thread-2] com.example.disruptor.WebConfig: 消息序例:0, 发送到《用户等级服务》, 订单详情:OrderEvent(orderId=VL-20240495001, name=苹果1斤, price=7.65)
2024-04-30T18:08:05.230+08:00  INFO 39828 --- [pool-4-thread-3] com.example.disruptor.WebConfig: 消息序例:0, 发送到《电商客服A组》, 订单详情:OrderEvent(orderId=VL-20240495001, name=苹果1斤, price=7.65)
2024-04-30T18:08:05.230+08:00  INFO 39828 --- [pool-4-thread-6] com.example.disruptor.WebConfig: 消息序例:0, 发送到《仓储管理服务》, 订单详情:OrderEvent(orderId=VL-20240495001, name=苹果1斤, price=7.65)
2024-04-30T18:08:05.230+08:00  INFO 39828 --- [pool-4-thread-7] com.example.disruptor.WebConfig: 消息序例:0, 发送到《物流投递服务》, 订单详情:OrderEvent(orderId=VL-20240495001, name=苹果1斤, price=7.65)

通过日志清楚的展示了，消费者消息处理器按程序执行配置的链路顺序正确打印；

结束

以上介绍了disruptor的基本信息与特点，并通过代码工程演示了dirsruptor在项目中如何开发，以及使用场景等；本文内容介绍有限，实际应用过程中disruptor还有很多其它用法，并未通过本文全整的展述，比如：如何使用多个消费者在线程池下完成消费，以及多生产者模式；可以通过官方文档与源码了解更多dirsruptor用法与功能；本文如有不足之处，欢迎指正与交流；

参考：

高性能队列——Disruptor-CSDN博客

LMAX Disruptor