Disruptor专题简单案例资料

package com.example.demo.producer; import com.example.demo.consumer.LongEventHandler; import com.example.demo.consumer.LongEventHandler2; import com.example.demo.entity.LongEvent; import com.example.demo.factory.LongEventFactory; import com.lmax.disruptor.EventFactory; import com.lmax.disruptor.EventHandler; import com.lmax.disruptor.RingBuffer; import com.lmax.disruptor.YieldingWaitStrategy; import com.lmax.disruptor.dsl.Disruptor; import com.lmax.disruptor.dsl.ProducerType; import java.nio.ByteBuffer; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; public class Main { public static void main(String[] args) { // 1.创建一个可缓存的线程 提供线程来出发Consumer 的事件处理 ExecutorService executor = Executors.newCachedThreadPool(); // 2.创建工厂 EventFactory<LongEvent> eventFactory = new LongEventFactory(); // 3.创建ringBuffer 大小 // ringBufferSize大小一定要是2的N次方 int ringbuffer = 1024*1024; // 4.创建Disruptor Disruptor<LongEvent> disruptor = new Disruptor(eventFactory, ringbuffer, executor, ProducerType.MULTI, new YieldingWaitStrategy()); // 5.连接消费端方法 disruptor.handleEventsWith(new EventHandler[]{new LongEventHandler()}); disruptor.handleEventsWith(new EventHandler[]{new LongEventHandler2()}); // 6.启动 disruptor.start(); // 7.创建RingBuffer容器 RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer(); // 8.创建生产者 LongEventProducer producer = new LongEventProducer(ringBuffer); // 9.指定缓冲区大小 ByteBuffer byteBuffer = ByteBuffer.allocate(8); for(int i = 1; i < 100; ++i) { byteBuffer.putLong(0, (long)i); producer.onData(byteBuffer); } //10.关闭disruptor和executor executor.shutdown(); disruptor.shutdown(); } }