大家好，我是君哥。
Disruptor 是一款高性能的内存有界队列，它通过内存预分配、无锁并发、解决伪共享问题、使用 RingBuffer 取代阻塞队列等措施来大幅提升队列性能。
但开发者们往往对它的使用场景不太了解，到底应该在哪些场景使用呢？今天咱们就来聊一聊 Disruptor 的使用场景。
Disruptor 是一个生产-消费模式的队列，这里我们使用官网的示例，生产者发送一个 long 类型的变量，消费者收到消息后把变量打印出来。首先定义消息体：
public class LongEvent {
private long value;
public void set(long value)
{
this.value = value;
}
@Override
public String toString()
{
return "LongEvent{" + "value=" + value + '}';
}
}
为了让 Disruptor 给消息预先分配内存，定义一个 EventFactory，代码如下：
public class LongEventFactory implements EventFactory<LongEvent>
{
@Override
public LongEvent newInstance()
{
return new LongEvent();
}
}
下面定义个消费者 LongEventHandler：
public class LongEventHandler implements EventHandler<LongEvent>
{
private String consumer;
public LongEventHandler(String consumer) {
this.consumer = consumer;
}
@Override
public void onEvent(LongEvent event, long sequence, boolean endOfBatch)
{
System.out.println("consumer: " + consumer + ",Event: " + event);
}
}
1.广播场景
广播场景在我们的开发工作中并不少见，比如系统收到上游系统的一个请求消息，然后把这个消息发送给多个下游系统来处理。Disruptor 支持广播模式。比如消费者生产的消息由三个消费者来消费：
public class Broadcast {
public static void main(String[] args) throws InterruptedException {
int bufferSize = 1024;
Disruptor<LongEvent> disruptor =
new Disruptor<>(LongEvent::new, bufferSize, DaemonThreadFactory.INSTANCE);
EventHandler<LongEvent> consumer1 = new LongEventHandler("consumer1");
EventHandler<LongEvent> consumer2 = new LongEventHandler("consumer2");
EventHandler<LongEvent> consumer3 = new LongEventHandler("consumer3");
disruptor.handleEventsWith(consumer1, consumer2, consumer3);
disruptor.start();
RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
ByteBuffer bb = ByteBuffer.allocate(8);
for (long l = 0; true; l++)
{
bb.putLong(0, l);
ringBuffer.publishEvent((event, sequence, buffer) -> event.set(buffer.getLong(0)), bb);
Thread.sleep(1000);
}
}
}
2.日志收集
再来看一个日志收集的例子。这里我们假设一个场景，业务系统集群有 3 个节点，每个节点打印的业务日志发送到 Disruptor，Disruptor 下游有 3 个消费者负责日志收集。
这里我们需要重新定义一个日志收集处理类，代码如下：
public class LogCollectHandler implements WorkHandler<LongEvent> {
public LogCollectHandler(String consumer) {
this.consumer = consumer;
}
private String consumer;
@Override
public void onEvent(LongEvent event)
{
System.out.println("consumer: " + consumer + ",Event: " + event);
}
}
下面这个代码是绑定消费者的代码：
public static void main(String[] args) throws InterruptedException {
int bufferSize = 1024;
Disruptor<LongEvent> disruptor =
new Disruptor<>(LongEvent::new, bufferSize, DaemonThreadFactory.INSTANCE);
WorkHandler<LongEvent> consumer1 = new LogCollectHandler("consumer1");
WorkHandler<LongEvent> consumer2 = new LogCollectHandler("consumer2");
WorkHandler<LongEvent> consumer3 = new LogCollectHandler("consumer3");
disruptor.handleEventsWithWorkerPool(consumer1, consumer2, consumer3);
disruptor.start();
}
需要注意的是，上面使用的是 Disruptor 的 handleEventsWithWorkerPool 方法，使用的消费者不是 EventHandler，而是 WorkHandler。消费者组里面的消费者如果是 WorkHandler，那消费者之间就是有竞争的，比如一个 Event 已经被 consumer1 消费过，那就不再会被其他消费者消费了。消费者组里面的消费者如果是 EventHandler，那消费者之间是没有竞争的，所有消息都会消费。
3.责任链
责任链这种设计模式我们都比较熟悉了，同一个对象的处理有多个不同的逻辑，每个逻辑作为一个节点组成责任链，比如收到一条告警消息，处理节点分为：给开发人员发送邮件、给运维人员发送短信、给业务人员发送 OA 消息。
Disruptor 支持链式处理消息，看下面的示例代码：
public static void main(String[] args) throws InterruptedException {
int bufferSize = 1024;
Disruptor<LongEvent> disruptor =
new Disruptor<>(LongEvent::new, bufferSize, DaemonThreadFactory.INSTANCE);
EventHandler<LongEvent> consumer1 = new LongEventHandler("consumer1");
EventHandler<LongEvent> consumer2 = new LongEventHandler("consumer2");
EventHandler<LongEvent> consumer3 = new LongEventHandler("consumer3");
disruptor.handleEventsWith(consumer1).then(consumer2).then(consumer3);
disruptor.start();
}
Disruptor 也支持多个并行责任链，下图是 2 条责任链的场景：
这里给出一个示例代码：
public static void main(String[] args) throws InterruptedException {
int bufferSize = 1024;
Disruptor<LongEvent> disruptor =
new Disruptor<>(LongEvent::new, bufferSize, DaemonThreadFactory.INSTANCE);
EventHandler<LongEvent> consumer1 = new LongEventHandler("consumer1");
EventHandler<LongEvent> consumer2 = new LongEventHandler("consumer2");
EventHandler<LongEvent> consumer3 = new LongEventHandler("consumer3");
EventHandler<LongEvent> consumer4 = new LongEventHandler("consumer4");
EventHandler<LongEvent> consumer5 = new LongEventHandler("consumer5");
EventHandler<LongEvent> consumer6 = new LongEventHandler("consumer6");
disruptor.handleEventsWith(consumer1).then(consumer2).then(consumer3);
disruptor.handleEventsWith(consumer4).then(consumer5).then(consumer6);
disruptor.start();
}
4.多任务协作
一个经典的例子，我们在泡咖啡之前，需要烧水、洗被子、磨咖啡粉，这三个步骤可以并行，但是需要等着三步都完成之后，才可以泡咖啡。
当然，这个例子可以用 Java 中的 CompletableFuture 来实现，代码如下：
public static void main(String[] args){
ExecutorService executor = ...;
CompletableFuture future1 = CompletableFuture.runAsync(() -> {
try {
washCup();
} catch (InterruptedException e) {
e.printStackTrace();
}
}, executor);
CompletableFuture future2 = CompletableFuture.runAsync(() -> {
try {
hotWater();
} catch (InterruptedException e) {
e.printStackTrace();
}
}, executor);
CompletableFuture future3 = CompletableFuture.runAsync(() -> {
try {
grindCoffee();
} catch (InterruptedException e) {
e.printStackTrace();
}
}, executor);
CompletableFuture.allOf(future1, future2, future3).thenAccept(
r -> {
System.out.println("泡咖啡");
}
);
System.out.println("我是主线程");
}
同样，使用 Disruptor 也可以实现这个场景，看下面代码：
public static void main(String[] args) throws InterruptedException {
int bufferSize = 1024;
Disruptor<LongEvent> disruptor =
new Disruptor<>(LongEvent::new, bufferSize, DaemonThreadFactory.INSTANCE);
EventHandler<LongEvent> consumer1 = new LongEventHandler("consumer1");
EventHandler<LongEvent> consumer2 = new LongEventHandler("consumer2");
EventHandler<LongEvent> consumer3 = new LongEventHandler("consumer3");
EventHandler<LongEvent> consumer4 = new LongEventHandler("consumer4");
disruptor.handleEventsWith(consumer1, consumer2, consumer3).then(consumer4);
disruptor.start();
}
5.多消费者组
类比主流消息队列的场景，Disruptor 也可以实现多消费者组的场景，组间并行消费互不影响，组内消费者竞争消息，如下图：
示例代码如下：
public static void main(String[] args) throws InterruptedException {
int bufferSize = 1024;
Disruptor<LongEvent> disruptor =
new Disruptor<>(LongEvent::new, bufferSize, DaemonThreadFactory.INSTANCE);
WorkHandler<LongEvent> consumer1 = new LogWorkHandler("consumer1");
WorkHandler<LongEvent> consumer2 = new LogWorkHandler("consumer2");
WorkHandler<LongEvent> consumer3 = new LogWorkHandler("consumer3");
WorkHandler<LongEvent> consumer4 = new LogWorkHandler("consumer4");
WorkHandler<LongEvent> consumer5 = new LogWorkHandler("consumer5");
WorkHandler<LongEvent> consumer6 = new LogWorkHandler("consumer6");
disruptor.handleEventsWithWorkerPool(consumer1, consumer2, consumer3);
disruptor.handleEventsWithWorkerPool(consumer4, consumer5, consumer6);
disruptor.start();
}
6.总结
通过消费者的灵活组合，Disruptor 的使用场景非常丰富。本文介绍了 Disruptor 的 5 个典型使用场景。在选型的时候，除了使用场景，更多地要考虑到 Disruptor 作为高性能内存队列的这个特点。
