摘要：我們知道是一個(gè)隊(duì)列，生產(chǎn)者往隊(duì)列里發(fā)布一項(xiàng)事件或稱(chēng)之為消息也可以時(shí)，消費(fèi)者能獲得通知如果沒(méi)有事件時(shí)，消費(fèi)者被堵塞，直到生產(chǎn)者發(fā)布了新的事件。實(shí)戰(zhàn)本文先不具體去闡述的工作具體原理，只是簡(jiǎn)單地將與其整合。

從功能上來(lái)看，Disruptor 是實(shí)現(xiàn)了“隊(duì)列”的功能，而且是一個(gè)有界隊(duì)列。那么它的應(yīng)用場(chǎng)景自然就是“生產(chǎn)者-消費(fèi)者”模型的應(yīng)用場(chǎng)合了。
可以拿 JDK 的 BlockingQueue 做一個(gè)簡(jiǎn)單對(duì)比，以便更好地認(rèn)識(shí) Disruptor 是什么。
我們知道 BlockingQueue 是一個(gè) FIFO 隊(duì)列，生產(chǎn)者(Producer)往隊(duì)列里發(fā)布(publish)一項(xiàng)事件(或稱(chēng)之為“消息”也可以)時(shí)，消費(fèi)者(Consumer)能獲得通知；如果沒(méi)有事件時(shí)，消費(fèi)者被堵塞，直到生產(chǎn)者發(fā)布了新的事件。
這些都是 Disruptor 能做到的，與之不同的是，Disruptor 能做更多：

同一個(gè)“事件”可以有多個(gè)消費(fèi)者，消費(fèi)者之間既可以并行處理，也可以相互依賴(lài)形成處理的先后次序(形成一個(gè)依賴(lài)圖)；

預(yù)分配用于存儲(chǔ)事件內(nèi)容的內(nèi)存空間；

針對(duì)極高的性能目標(biāo)而實(shí)現(xiàn)的極度優(yōu)化和無(wú)鎖的設(shè)計(jì)；

以上雖然簡(jiǎn)單地描述了 Disruptor 是什么，但對(duì)于它"能做什么"，還不是那么明白。簡(jiǎn)而言之，當(dāng)你需要在兩個(gè)獨(dú)立的處理過(guò)程之間交換數(shù)據(jù)時(shí)，就可以使用 Disruptor 。當(dāng)然使用隊(duì)列也可以，只不過(guò) Disruptor 的性能更好。

本文先不具體去闡述Disruptor的工作具體原理，只是簡(jiǎn)單地將Spring與其整合。整合過(guò)程很簡(jiǎn)單，具體步驟如下：

在pom文件中引入disruptor

    com.lmax
    disruptor
    3.4.2

創(chuàng)建事件

@Data
public class NotifyEvent {
    private String message;
}

創(chuàng)建消息工廠(chǎng)用于生產(chǎn)消息

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

創(chuàng)建消費(fèi)者，此處用于處理業(yè)務(wù)邏輯

public class NotifyEventHandler implements EventHandler,WorkHandler {

    @Override
    public void onEvent(NotifyEvent notifyEvent, long l, boolean b) throws Exception {
        System.out.println("接收到消息");
        this.onEvent(notifyEvent);

    }

    @Override
    public void onEvent(NotifyEvent notifyEvent) throws Exception {
        System.out.println(notifyEvent+">>>"+UUID.randomUUID().toString());
    }
}

自定義異常

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

    @Override
    public void handleOnStartException(Throwable throwable) {
        log.error("start disruptor error ==[{}]!", throwable.getMessage());
    }

    @Override
    public void handleOnShutdownException(Throwable throwable) {
        log.error("shutdown disruptor error ==[{}]!", throwable.getMessage());
    }
}

整合Spring，對(duì)Disruptor進(jìn)行初始化

@Service
public class NotifyServiceImpl implements INotifyService, DisposableBean,InitializingBean {
    private Disruptor disruptor;
    private static final int RING_BUFFER_SIZE = 1024 * 1024;

    @Override
    public void destroy() throws Exception {
        disruptor.shutdown();
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        disruptor = new Disruptor(new NotifyEventFactory(),RING_BUFFER_SIZE, Executors.defaultThreadFactory(), ProducerType.SINGLE,new BlockingWaitStrategy());
        disruptor.setDefaultExceptionHandler(new NotifyEventHandlerException());
        disruptor.handleEventsWith(new NotifyEventHandler());
        disruptor.start();
    }


    @Override
    public void sendNotify(String message) {
        RingBuffer ringBuffer = disruptor.getRingBuffer();
//        ringBuffer.publishEvent(new EventTranslatorOneArg() {
//            @Override
//            public void translateTo(NotifyEvent event, long sequence, String data) {
//                event.setMessage(data);
//            }
//        }, message);
        ringBuffer.publishEvent((event, sequence, data) -> event.setMessage(data), message); //lambda式寫(xiě)法，如果是用jdk1.8以下版本使用以上注釋的一段
     
    }
}

在需要調(diào)用的地方注入INotifyService并調(diào)用sendNotify方法

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