Spring集成高性能队列Disruptor

Disruptor简介

Disruptor（中文翻译为“破坏者”或“颠覆者”）是一种高性能、低延迟的并发编程框架，最初由LMAX Exchange开发。它的主要目标是解决在金融交易系统等需要高吞吐量和低延迟的应用中的并发问题。

Disruptor特点

无锁并发：Disruptor基于无锁的数据结构，它允许多个线程并发地读取和写入数据，而无需使用传统锁定机制，从而避免锁定的竞争和性能瓶颈。
环形缓冲区：Disruptor使用一个环形缓冲区，将数据项存储在其中。这个环形缓冲区可以通过预分配的方式来减少内存分配和垃圾回收的开销。
事件驱动：Disruptor的核心思想是将数据项（事件）从生产者传递到消费者，通过一种发布-订阅的模型来实现。生产者将事件写入缓冲区，而消费者从缓冲区中读取事件进行处理。
高性能：Disruptor专注于提供极高的吞吐量和低延迟，适用于需要快速处理大量数据的应用，如金融交易系统、网络数据传输等。
易于使用：尽管Disruptor的内部实现复杂，但它提供了简单而清晰的API，使开发人员能够相对容易地集成和使用它。
并发编程的辅助工具：Disruptor不仅仅是一个并发编程框架，还提供了一些辅助工具，如事件处理器、工作池等，帮助开发者更好地处理并发任务。

Disruptor应用

Disruptor可以理解为一个可以集成在项目里的MQ,它主要也分为了生产者,消息队列和消费者这么几部分,接下来用一个例子演示

引进依赖

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

定义监听实体

@Data
public class DisruptorEvent {/*** 定义加入队列的时间毫秒值*/private long creatTime;/*** 携带的其他信息*/private String data;}

定义消费者

实现EventHandler接口,重写onEvent方法

@Slf4j
public class DisruptorConsumer implements EventHandler<DisruptorEvent> {@Overridepublic void onEvent(DisruptorEvent disruptorEvent, long l, boolean b) throws Exception {log.debug("消费者开始消费数据:[{}]", disruptorEvent);//模拟复杂环境下系统延迟Thread.sleep(1100);}
}

初始化Disruptor

由于我们后续还要用到这个对象,所以要把他交给IOC容器来管理,定义bean名称,避免后续有多个Disruptor对象

@Configuration
public class DisruptorConfig {/*** 队列长度,注意必须是2的n次幂*/private static final int RING_BUFFER_SIZE = 1024;@Bean(name = "cardDisruptor")public Disruptor<DisruptorEvent> disruptorStart() {DisruptorConsumer consumer = new DisruptorConsumer();Disruptor<DisruptorEvent> disruptor = new Disruptor<>(DisruptorEvent::new,RING_BUFFER_SIZE,Executors.defaultThreadFactory(),ProducerType.MULTI,new BlockingWaitStrategy());disruptor.handleEventsWith(consumer);disruptor.start();return disruptor;}
}

定义生产者

@Slf4j
public class DisruptorProducer {private static Disruptor<DisruptorEvent> disruptor;public DisruptorProducer(Disruptor<DisruptorEvent> disruptor) {DisruptorProducer.disruptor = disruptor;}public static void push(DisruptorEvent disruptorEvent) {//获取队列RingBuffer<DisruptorEvent> ringBuffer = disruptor.getRingBuffer();boolean flag = ringBuffer.tryPublishEvent(new EventTranslator<DisruptorEvent>() {@Overridepublic void translateTo(DisruptorEvent event, long l) {//这里一定要写成这种set的形式(也就是说不要改变这个event的内存指向地址),不然消费者会拿不到值event.setData(disruptorEvent.getData());event.setCreatTime(disruptorEvent.getCreatTime());}});if (!flag) {throw new RuntimeException("发送消息失败!");}}
}

测试

使用postman模拟并发10次

public void test() {DisruptorEvent event = new DisruptorEvent();event.setData("哈哈哈");event.setCreatTime(System.currentTimeMillis());DisruptorProducer.push(event);log.info("请求成功");}