demo1 单生产者多消费者创建。
maven 依赖
<!-- https://mvnrepository.com/artifact/com.lmax/disruptor -->
<dependency>
<groupId>com.lmax</groupId>
<artifactId>disruptor</artifactId>
<version>3.4.</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();
}
},
<< ,
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();
ExecutorService es = Executors.newFixedThreadPool();
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 = ; i < ; i++) {
disruptor.publishEvent(new MessageEventTranslator());
}
latch.countDown();
}
}
6 阻塞等待 & 关闭服务
// 阻塞等待
latch.await(); // 关闭服务
es.shutdown();
disruptor.shutdown();