Spring Boot3.x集成Disruptor4.0

本文主要是介绍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的等待&&唤醒机制实现等待逻辑，是默认策略，比较节省CPUBusySpinWaitStrategy：持续自旋，JDK9之下慎用（最好别用）DummyWaitStrategy：返回的Sequence值为0，正常环境是用不上的LiteBlockingWaitStrategy：基于BlockingWaitStrategy，在没有锁竞争的时候会省去唤醒操作，但是作者说测试不充分，不建议使用TimeoutBlockingWaitStrategy：带超时的等待，超时后会执行业务指定的处理逻辑LiteTimeoutBlockingWaitStrategy：基于TimeoutBlockingWaitStrategy，在没有锁竞争的时候会省去唤醒操作SleepingWaitStrategy：三段式，第一阶段自旋，第二阶段执行Thread.yield交出CPU，第三阶段睡眠执行时间，反复的的睡眠YieldingWaitStrategy：二段式，第一阶段自旋，第二阶段执行Thread.yield交出CPUPhasedBackoffWaitStrategy：四段式，第一阶段自旋指定次数，第二阶段自旋指定时间，第三阶段执行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*/@Beanpublic 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# 服务IPaddress: 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

这篇关于Spring Boot3.x集成Disruptor4.0的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---