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

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

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

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

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

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

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

在pom文件中引入disruptor

    com.lmax
    disruptor
    3.4.2

創(chuàng)建事件

@Data
public class NotifyEvent {
    private String message;
}

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

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

創(chuàng)建消費者，此處用于處理業(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，對Disruptor進行初始化

@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式寫法，如果是用jdk1.8以下版本使用以上注釋的一段
     
    }
}

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

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