前言

Disruptor是一个高功能的无锁并发结构,其主要运用场景是在高并发、低推迟的体系中,如金融范畴的交易体系,游戏服务器等。其长处便是非常快,声称能支撑每秒600万订单。需求留意的是,Disruptor是单机结构,对标JDK中的Queue,而非可用于分布式体系的MQ

本文基于Disruptor v3.4.*版本

Demo

既然是简略运用,这阶段只需求重视:

  • 出产者
  • 顾客:EventHandler
  • 音讯的传递:音讯的载体Event

简略比如

首要,我们定义音讯的载体Event,出产者向顾客传递的音讯经过Event承载

class LongEvent {
    private long value;
    public void set(long value) {
        this.value = value;
    }
    @Override
    public String toString() {
        return "LongEvent{" + "value=" + value + '}';
    }
}

然后定义Event出产工厂,这用于初始化Event

EventFactory<LongEvent> factory = new EventFactory<LongEvent>() {
    @Override
    public LongEvent newInstance() {
        return new LongEvent();
    }
};

接下来就可以构建Disruptor了,以下是完整代码

// 音讯载体(event)
static class LongEvent {
    private long value;
    public void set(long value) {
        this.value = value;
    }
    @Override
    public String toString() {
        return "LongEvent{" + "value=" + value + '}';
    }
}
// 发布音讯的转换器
public static void translate(LongEvent event, long sequence, ByteBuffer buffer)
{
    event.set(buffer.getLong(0));
}
public static void main(String[] args) throws Exception {
    // event出产工厂,初始化RingBuffer的时分运用
    EventFactory<LongEvent> factory = new EventFactory<LongEvent>() {
        @Override
        public LongEvent newInstance() {
            return new LongEvent();
        }
    };
    // 指定RingBuffer的大小(有必要是2的n次方)
    int bufferSize = 1024;
    // 结构Disruptor(默认运用多出产者模式、BlockingWaitStrategy堵塞战略)
    Disruptor<LongEvent> disruptor = new Disruptor<>(LongEvent::new, bufferSize, DaemonThreadFactory.INSTANCE);
    //  Disruptor<LongEvent> disruptor = new Disruptor<>(factory, bufferSize, DaemonThreadFactory.INSTANCE, ProducerType.MULTI, new BlockingWaitStrategy());
    // 设置顾客
    EventHandler<LongEvent> handler = (event, sequence, endOfBatch) -> {
        System.out.println("Event: " + event);
    };
    disruptor.handleEventsWith(handler);
    // 发动disruptor,发动一切需求运转的线程
    disruptor.start();
    RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
    ByteBuffer bb = ByteBuffer.allocate(8);
    for (long i = 0; i < 100; i++) {
        bb.putLong(i);
        // 发布事件
        ringBuffer.publishEvent(LongEventMain::translate, bb);
    }
}

顾客组合(多运用场景)

Disruptor不仅可以当高功能的队列运用,还支持顾客的串行、并行消费等

以下只展现关键代码(设置顾客),其余部分参考上一节的简略demo

  1. 单链串行

    Disruptor-简单使用

    disruptor.handleEventsWith(handlerA).then(handlerB);
    
  2. 并行

    Disruptor-简单使用

    disruptor.handleEventsWith(handlerA, handlerB);
    
  3. 链内串行,多链并行

    Disruptor-简单使用

    disruptor.handleEventsWith(handlerA).then(handlerC);
    disruptor.handleEventsWith(handlerB).then(handlerD);
    
  4. 菱形(C、D都履行完才到E)

    Disruptor-简单使用

    disruptor.handleEventsWith(handlerA).then(handlerC);
    disruptor.handleEventsWith(handlerB).then(handlerD);
    disruptor.after(handlerC, handlerD).then(handlerE);
    
  5. 分组(AB都履行完才到CD)

    Disruptor-简单使用

    disruptor.handleEventsWith(handlerA, handlerB).then(handlerC, handlerD);
    
  6. 分组不重复消费

    组内竞赛,组外串行:每个音讯在每个分组中只要一个顾客能消费成功,假如便是分组A中只要HandlerA2能得到数据,分组B中只要HandlerB1获得

    Disruptor-简单使用

    // 留意:此处的handler实现的是WorkHandler接口
    disruptor.handleEventsWithWorkerPool(handlerA1, handlerA2, handlerA3)
                    .then(handlerB1, handlerB2, handlerB3);
    
  7. 分组不重复消费(菱形)

    Disruptor-简单使用

    // handlerA、handlerB实现WorkHandler接口
    // handlerC 实现EventHandler或WorkHandler接口均可
    disruptor.handleEventsWithWorkerPool(handlerA1, handlerA2, handlerA3)
                    .then(handlerB1, handlerB2, handlerB3)
                    .then(handlerC);
    

    等候战略

    顾客速度比出产者快时,需求等候。因此就有了不同的等候战略以习惯不同场景

    • BlockingWaitStrategy

      默认战略。运用锁和 Condition 的等候、唤醒机制。速度慢,但节省CPU资源并且在不同布置环境中能提供更加共同的功能体现。

    • YieldingWaitStrategy

      二段式,一阶段自旋100次,二阶段履行Thread.yield,需求低推迟的场景可运用此战略

    • SleepingWaitStrategy

      三段式,一阶段自旋,二阶段履行Thread.yield,三阶段睡眠

    • BusySpinWaitStrategy

      功能最高的战略,与 YieldingWaitStrategy 相同在低推迟场景运用,可是此战略要求顾客数量低于 CPU 逻辑内核总数

    其他小技巧

    1. 铲除音讯载体 Event 中的数据

      假如 Event 中存在大对象,应该在顾客链的末尾,增加一个铲除数据的顾客,以协助jvm废物收回。demo中的 LongEvent 是 private long value; 所以没必要增加。

总结

本文介绍了 Disruptor 的简略运用,以及复杂场景下顾客的配置。下篇开坑 Disruptor 源码解析。


参考资料

Disruptor官方文档