Spring Boot3.x集成Disruptor4.0

2024-05-07 21:20

本文主要是介绍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的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/968420

相关文章

一文详解SpringBoot中控制器的动态注册与卸载

《一文详解SpringBoot中控制器的动态注册与卸载》在项目开发中,通过动态注册和卸载控制器功能,可以根据业务场景和项目需要实现功能的动态增加、删除,提高系统的灵活性和可扩展性,下面我们就来看看Sp... 目录项目结构1. 创建 Spring Boot 启动类2. 创建一个测试控制器3. 创建动态控制器注

Java操作Word文档的全面指南

《Java操作Word文档的全面指南》在Java开发中,操作Word文档是常见的业务需求,广泛应用于合同生成、报表输出、通知发布、法律文书生成、病历模板填写等场景,本文将全面介绍Java操作Word文... 目录简介段落页头与页脚页码表格图片批注文本框目录图表简介Word编程最重要的类是org.apach

Spring Boot中WebSocket常用使用方法详解

《SpringBoot中WebSocket常用使用方法详解》本文从WebSocket的基础概念出发,详细介绍了SpringBoot集成WebSocket的步骤,并重点讲解了常用的使用方法,包括简单消... 目录一、WebSocket基础概念1.1 什么是WebSocket1.2 WebSocket与HTTP

SpringBoot+Docker+Graylog 如何让错误自动报警

《SpringBoot+Docker+Graylog如何让错误自动报警》SpringBoot默认使用SLF4J与Logback,支持多日志级别和配置方式,可输出到控制台、文件及远程服务器,集成ELK... 目录01 Spring Boot 默认日志框架解析02 Spring Boot 日志级别详解03 Sp

java中反射Reflection的4个作用详解

《java中反射Reflection的4个作用详解》反射Reflection是Java等编程语言中的一个重要特性,它允许程序在运行时进行自我检查和对内部成员(如字段、方法、类等)的操作,本文将详细介绍... 目录作用1、在运行时判断任意一个对象所属的类作用2、在运行时构造任意一个类的对象作用3、在运行时判断

java如何解压zip压缩包

《java如何解压zip压缩包》:本文主要介绍java如何解压zip压缩包问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录Java解压zip压缩包实例代码结果如下总结java解压zip压缩包坐在旁边的小伙伴问我怎么用 java 将服务器上的压缩文件解压出来,

SpringBoot中SM2公钥加密、私钥解密的实现示例详解

《SpringBoot中SM2公钥加密、私钥解密的实现示例详解》本文介绍了如何在SpringBoot项目中实现SM2公钥加密和私钥解密的功能,通过使用Hutool库和BouncyCastle依赖,简化... 目录一、前言1、加密信息(示例)2、加密结果(示例)二、实现代码1、yml文件配置2、创建SM2工具

Spring WebFlux 与 WebClient 使用指南及最佳实践

《SpringWebFlux与WebClient使用指南及最佳实践》WebClient是SpringWebFlux模块提供的非阻塞、响应式HTTP客户端,基于ProjectReactor实现,... 目录Spring WebFlux 与 WebClient 使用指南1. WebClient 概述2. 核心依

Spring Boot @RestControllerAdvice全局异常处理最佳实践

《SpringBoot@RestControllerAdvice全局异常处理最佳实践》本文详解SpringBoot中通过@RestControllerAdvice实现全局异常处理,强调代码复用、统... 目录前言一、为什么要使用全局异常处理?二、核心注解解析1. @RestControllerAdvice2

Spring IoC 容器的使用详解(最新整理)

《SpringIoC容器的使用详解(最新整理)》文章介绍了Spring框架中的应用分层思想与IoC容器原理,通过分层解耦业务逻辑、数据访问等模块,IoC容器利用@Component注解管理Bean... 目录1. 应用分层2. IoC 的介绍3. IoC 容器的使用3.1. bean 的存储3.2. 方法注