并发编程Disruptor入门

叁歲伎倆 2022-07-13 13:26 142阅读 0赞

**获得Disruptor**  
可以通过Maven或者下载jar来安装Disruptor。只要把对应的jar放在Java classpath就可以了。

**基本的事件生产和消费**  
我们从一个简单的例子开始学习Disruptor：生产者传递一个long类型的值给消费者，而消费者消费这个数据的方式仅仅是把它打印出来。首先声明一个Event来包含需要传递的数据

public class LongEvent { 
        private long value;
        public long getValue() { 
            return value; 
        } 
    
        public void setValue(long value) { 
            this.value = value; 
        } 
    }

由于需要让Disruptor为我们创建事件，我们同时还声明了一个EventFactory来实例化Event对象

public class LongEventFactory implements EventFactory{ 
    
        public Object newInstance() {
            return new LongEvent();
        }
    
    }

我们还需要一个事件消费者，也就是一个事件处理器。这个事件处理器简单地把事件中存储的数据打印到终端

public class LongEventHandler implements EventHandler<LongEvent>{ 
    
        public void onEvent(LongEvent longEvent, long l, boolean b)
                throws Exception {
            System.err.println("收到数据了:"+longEvent.getValue());
    
        }
    
    }

事件都会有一个生成事件的源，这个例子中假设事件是由于磁盘IO或者network读取数据的时候触发的，事件源使用一个ByteBuffer来模拟它接受到的数据，也就是说，事件源会在IO读取到一部分数据的时候触发事件（触发事件不是自动的，程序员需要在读取到数据的时候自己触发事件并发布）

public class LongEventProducer { 
    
        private final RingBuffer<LongEvent> ringBuffer;
    
        public LongEventProducer(RingBuffer<LongEvent> ringBuffer) {
            this.ringBuffer = ringBuffer;
        }
    
        /** * onData用来发布事件，每调用一次就发布一次事件事件 它的参数会通过事件传递给消费者 * * @param b */
        public void onData(ByteBuffer b) {
            // 可以把ringBuffer看做一个事件队列，那么next就是得到下面一个事件槽
            long sequence = ringBuffer.next();
            try {
                // 用上面的索引取出一个空的事件用于填充
                LongEvent event = ringBuffer.get(sequence);
                // for the sequence
                event.setValue(b.getLong(0));
            } finally {
                // 发布事件
                ringBuffer.publish(sequence);
            }
        }
    }

很明显的是：当用一个简单队列来发布事件的时候会牵涉更多的细节，这是因为事件对象还需要预先创建。发布事件最少需要两步：获取下一个事件槽并发布事件（发布事件的时候要使用try/finnally保证事件一定会被发布）。如果我们使用RingBuffer.next()获取一个事件槽，那么一定要发布对应的事件。如果不能发布事件，那么就会引起Disruptor状态的混乱。尤其是在多个事件生产者的情况下会导致事件消费者失速，从而不得不重启应用才能会恢复。

Disruptor 3.0提供了lambda式的API。这样可以把一些复杂的操作放在Ring Buffer，所以在Disruptor3.0以后的版本最好使用Event Publisher或者Event Translator来发布事件

public class LongEventProducerWithTranslator {
    
        // 一个translator可以看做一个事件初始化器，publicEvent方法会调用它填充Event
        private static final EventTranslatorOneArg<LongEvent, ByteBuffer> TRANSLATOR = new EventTranslatorOneArg<LongEvent, ByteBuffer>() {
            public void translateTo(LongEvent event, long sequence, ByteBuffer b) {
                event.setValue(b.getLong(0));
            }
        };
        private final RingBuffer<LongEvent> ringBuffer;
    
        public LongEventProducerWithTranslator(RingBuffer<LongEvent> ringBuffer) {
            this.ringBuffer = ringBuffer;
        }
    
        public void onData(ByteBuffer b) {
            ringBuffer.publishEvent(TRANSLATOR, b);
        }
    
    }

上面写法的另一个好处是，Translator可以分离出来并且更加容易单元测试。Disruptor提供了不同的接口(EventTranslator, EventTranslatorOneArg, EventTranslatorTwoArg, 等等)去产生一个Translator对象。很明显，Translator中方法的参数是通过RingBuffer来传递的。

最后一步就是把所有的代码组合起来完成一个完整的事件处理系统。Disruptor在这方面做了简化，使用了DSL风格的代码（其实就是按照直观的写法，不太能算得上真正的DSL）。虽然DSL的写法比较简单，但是并没有提供所有的选项。如果依靠DSL已经可以处理大部分情况了。

public class LongEventMain {
    
        public static void main(String[] args) throws InterruptedException {
            // Executor that will be used to construct new threads for consumers
            Executor executor = Executors.newCachedThreadPool();
            // The factory for the event
            LongEventFactory factory = new LongEventFactory();
            // Specify the size of the ring buffer, must be power of 2.
            int bufferSize = 1024;
            // Construct the Disruptor
            Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(factory,
                    bufferSize, executor);
            // Connect the handler
            disruptor.handleEventsWith(new LongEventHandler());
            // Start the Disruptor, starts all threads running
            disruptor.start();
            // Get the ring buffer from the Disruptor to be used for publishing.
            RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
    
            LongEventProducer producer = new LongEventProducer(ringBuffer);
    
            ByteBuffer b = ByteBuffer.allocate(8);
            for (long l = 0; l<100; l++) {
                b.putLong(0, l);
                producer.onData(b);
                //Thread.sleep(1000);
            }
            disruptor.shutdown();
        }
    }

得到结果：

收到数据了:0
    收到数据了:1
    收到数据了:2
    收到数据了:3
    收到数据了:4
    收到数据了:5
    收到数据了:6
    收到数据了:7
    收到数据了:8
    收到数据了:9
    收到数据了:10
    收到数据了:11
    收到数据了:12
    收到数据了:13
    收到数据了:14
    收到数据了:15
    收到数据了:16
    收到数据了:17
    收到数据了:18
    收到数据了:19
    收到数据了:20
    收到数据了:21
    收到数据了:22
    收到数据了:23
    收到数据了:24
    收到数据了:25
    收到数据了:26
    收到数据了:27
    收到数据了:28
    收到数据了:29
    收到数据了:30
    收到数据了:31
    收到数据了:32
    收到数据了:33
    收到数据了:34
    收到数据了:35
    收到数据了:36
    收到数据了:37
    收到数据了:38
    收到数据了:39
    收到数据了:40
    收到数据了:41
    收到数据了:42
    收到数据了:43
    收到数据了:44
    收到数据了:45
    收到数据了:46
    收到数据了:47
    收到数据了:48
    收到数据了:49
    收到数据了:50
    收到数据了:51
    收到数据了:52
    收到数据了:53
    收到数据了:54
    收到数据了:55
    收到数据了:56
    收到数据了:57
    收到数据了:58
    收到数据了:59
    收到数据了:60
    收到数据了:61
    收到数据了:62
    收到数据了:63
    收到数据了:64
    收到数据了:65
    收到数据了:66
    收到数据了:67
    收到数据了:68
    收到数据了:69
    收到数据了:70
    收到数据了:71
    收到数据了:72
    收到数据了:73
    收到数据了:74
    收到数据了:75
    收到数据了:76
    收到数据了:77
    收到数据了:78
    收到数据了:79
    收到数据了:80
    收到数据了:81
    收到数据了:82
    收到数据了:83
    收到数据了:84
    收到数据了:85
    收到数据了:86
    收到数据了:87
    收到数据了:88
    收到数据了:89
    收到数据了:90
    收到数据了:91
    收到数据了:92
    收到数据了:93
    收到数据了:94
    收到数据了:95
    收到数据了:96
    收到数据了:97
    收到数据了:98
    收到数据了:99