Spring整合Disruptor3

什麼是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 {
    @Override
    public Object newInstance() {
        return new NotifyEvent();
    }
}

• 建立消費者，此處用於處理業務邏輯

public class NotifyEventHandler implements EventHandler<NotifyEvent>,WorkHandler<NotifyEvent> {

    @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<NotifyEvent> 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<NotifyEvent>(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<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")
  @ResponseBody
  public String testLog() {
    log.info("=============");
    notifyService.sendNotify("Hello,World!");
    return "hello,world";
  }