disruptor 單生產者多消費者

demo1 單生產者多消費者建立。

maven 依賴

<!-- https://mvnrepository.com/artifact/com.lmax/disruptor -->
        <dependency>
            <groupId>com.lmax</groupId>
            <artifactId>disruptor</artifactId>
            <version>3.4.2</version>
        </dependency>

 

1 對象 - Message

@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
public class Message2 {
    private String id;
    private String name;
    private double price;
}

2 在主函數中建立 disruptor

Disruptor<Message2> disruptor = new Disruptor<>(
                new EventFactory<Message2>() {
                    @Override
                    public Message2 newInstance() {
                        return new Message2();
                    }
                },
                1 << 10,
                Executors.defaultThreadFactory(),
                ProducerType.SINGLE,
                new BusySpinWaitStrategy()
        );

 

3 disruptor 綁定消費者

// disruptor 綁定消費者
disruptor.handleEventsWith(new MessageHandler1())；


//建立消費者
@Slf4j
public class MessageHandler1 implements EventHandler<Message2> {
    @Override
    public void onEvent(Message2 event, long sequence, boolean endOfBatch) throws Exception {
        event.setId(UUID.randomUUID().toString());
        log.info("【handler1，set id】 id: {}, name: {}, price: {}", event.getId(), event.getName(), event.getPrice());
    }
}

 

4 啓動 disruptor

RingBuffer<Message2> ringBuffer = disruptor.start();

 

5 disruptor 綁定生產者

//綁定生產者
CountDownLatch latch = new CountDownLatch(1);
ExecutorService es = Executors.newFixedThreadPool(4);
es.submit(new MessagePublish2(disruptor, latch));

// 生產者類
public class MessagePublish2 implements Runnable {
    private Disruptor<Message2> disruptor;
    private CountDownLatch latch;

    public MessagePublish2(Disruptor<Message2> disruptor, CountDownLatch latch) {
        this.disruptor = disruptor;
        this.latch = latch;
    }

    @Override
    public void run() {
        for (int i = 0; i < 3; i++) {
            disruptor.publishEvent(new MessageEventTranslator());
        }
        latch.countDown();
    }
}

 

6 阻塞等待 & 關閉服務

        // 阻塞等待
        latch.await();

        // 關閉服務
        es.shutdown();
        disruptor.shutdown();