什么是Disruptor

从功能上来看，Disruptor 是实现了“队列”的功能，而且是一个有界队列。那么它的应用场景自然就是“生产者-消费者”模型的应用场合了。
可以拿 JDK 的 BlockingQueue 做一个简单对比，以便更好地认识 Disruptor 是什么。
我们知道 BlockingQueue 是一个 FIFO 队列，生产者(Producer)往队列里发布(publish)一项事件(或称之为“消息”也可以)时，消费者(Consumer)能获得通知；如果没有事件时，消费者被堵塞，直到生产者发布了新的事件。
这些都是 Disruptor 能做到的，与之不同的是，Disruptor 能做更多：

• 同一个“事件”可以有多个消费者，消费者之间既可以并行处理，也可以相互依赖形成处理的先后次序(形成一个依赖图)；
• 预分配用于存储事件内容的内存空间；
• 针对极高的性能目标而实现的极度优化和无锁的设计；

以上虽然简单地描述了 Disruptor 是什么，但对于它"能做什么"，还不是那么明白。简而言之，当你需要在两个独立的处理过程之间交换数据时，就可以使用 Disruptor 。当然使用队列也可以，只不过 Disruptor 的性能更好。

实战

本文先不具体去阐述Disruptor的工作具体原理，只是简单地将Spring与其整合。整合过程很简单，具体步骤如下：

• 在pom文件中引入disruptor

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

• 创建事件

@Data
public class NotifyEvent {private String message;
}

• 创建消息工厂用于生产消息

public class NotifyEventFactory implements EventFactory {@Overridepublic Object newInstance() {return new NotifyEvent();}
}

• 创建消费者，此处用于处理业务逻辑

public class NotifyEventHandler implements EventHandler<NotifyEvent>,WorkHandler<NotifyEvent> {@Overridepublic void onEvent(NotifyEvent notifyEvent, long l, boolean b) throws Exception {System.out.println("接收到消息");this.onEvent(notifyEvent);}@Overridepublic void onEvent(NotifyEvent notifyEvent) throws Exception {System.out.println(notifyEvent+">>>"+UUID.randomUUID().toString());}
}

• 自定义异常

@Log4j2
public class NotifyEventHandlerException implements ExceptionHandler {@Overridepublic void handleEventException(Throwable throwable, long sequence, Object event) {throwable.fillInStackTrace();log.error("process data error sequence ==[{}] event==[{}] ,ex ==[{}]", sequence, event.toString(), throwable.getMessage());}@Overridepublic void handleOnStartException(Throwable throwable) {log.error("start disruptor error ==[{}]!", throwable.getMessage());}@Overridepublic void handleOnShutdownException(Throwable throwable) {log.error("shutdown disruptor error ==[{}]!", throwable.getMessage());}
}

• 整合Spring，对Disruptor进行初始化

@Service
public class NotifyServiceImpl implements INotifyService, DisposableBean,InitializingBean {private Disruptor<NotifyEvent> disruptor;private static final int RING_BUFFER_SIZE = 1024 * 1024;@Overridepublic void destroy() throws Exception {disruptor.shutdown();}@Overridepublic void afterPropertiesSet() throws Exception {disruptor = new Disruptor<NotifyEvent>(new NotifyEventFactory(),RING_BUFFER_SIZE, Executors.defaultThreadFactory(), ProducerType.SINGLE,new BlockingWaitStrategy());disruptor.setDefaultExceptionHandler(new NotifyEventHandlerException());disruptor.handleEventsWith(new NotifyEventHandler());disruptor.start();}@Overridepublic void sendNotify(String message) {RingBuffer<NotifyEvent> ringBuffer = disruptor.getRingBuffer();
//        ringBuffer.publishEvent(new EventTranslatorOneArg<NotifyEvent,  String>() {
//            @Override
//            public void translateTo(NotifyEvent event, long sequence, String data) {
//                event.setMessage(data);
//            }
//        }, message);ringBuffer.publishEvent((event, sequence, data) -> event.setMessage(data), message); //lambda式写法，如果是用jdk1.8以下版本使用以上注释的一段}
}

• 在需要调用的地方注入INotifyService并调用sendNotify方法

  @GetMapping("test")@ResponseBodypublic String testLog() {log.info("=============");notifyService.sendNotify("Hello,World!");return "hello,world";}