Disruptor实现生产者消费者

冷不防 2023-06-26 10:57 36阅读 0赞

### 一、引入jar依赖 ###

<dependency>
        <groupId>com.lmax</groupId>
        <artifactId>disruptor</artifactId>
        <version>3.2.1</version>
    </dependency>

### 二、首先声明一个Event来包含需要传递的数据 ###

/**
     * @program: test_demo
     * @description: 定义事件event 通过Disruptor进行交换的数据类型
     * @author: HQ Zheng
     * @create: 2019-12-31 14:18
     */
    public class LongEvent {
        private Long value;
    
        public Long getValue() {
            return value;
        }
    
        public void setValue(Long value) {
            this.value = value;
        }
    }

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

/**
     * @program: test_demo
     * @description: 事件工厂
     * @author: HQ Zheng
     * @create: 2019-12-31 14:21
     */
    public class LongEventFactory implements EventFactory<LongEvent> {
    
        public LongEvent newInstance() {
    
            return new LongEvent();
        }
    
    }

### 四、事件消费者，也就是一个事件处理器。这个事件处理器简单地把事件中存储的数据打印到终端 ###

/**
     * @program: test_demo
     * @description: 事件消费者
     * @author: HQ Zheng
     * @create: 2019-12-31 14:23
     */
    public class LongEventHandler implements EventHandler<LongEvent> {
    
        public void onEvent(LongEvent event, long sequence, boolean endOfBatch) throws Exception {
            System.out.println("消费者:"+event.getValue());
        }
    
    }

### 五、事件生产者，事件产生者 ###

/**
     * @program: test_demo
     * @description: 生产者
     * @author: HQ Zheng
     * @create: 2019-12-31 14:25
     */
    public class LongEventProducer {
    
        public final RingBuffer<LongEvent> ringBuffer;
    
        public LongEventProducer(RingBuffer<LongEvent> ringBuffer) {
            this.ringBuffer = ringBuffer;
        }
    
        public void onData(ByteBuffer byteBuffer) {
            // 1.ringBuffer 事件队列 下一个槽
            long sequence = ringBuffer.next();
            Long data = null;
            try {
                //2.取出空的事件队列
                LongEvent longEvent = ringBuffer.get(sequence);
                data = byteBuffer.getLong(0);
                //3.获取事件队列传递的数据
                longEvent.setValue(data);
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            } finally {
                System.out.println("生产这准备发送数据");
                //4.发布事件
                ringBuffer.publish(sequence);
    
            }
        }
    
    }

### 六、测试类 ###

/**
     * @program: test_demo
     * @description: 测试
     * @author: HQ Zheng
     * @create: 2019-12-31 14:27
     */
    public class DisruptorTest {
        public static void main(String[] args) {
            // 1.创建一个可缓存的线程 提供线程来出发Consumer 的事件处理
            ExecutorService executor = Executors.newCachedThreadPool();
            // 2.创建工厂
            EventFactory<LongEvent> eventFactory = new LongEventFactory();
            // 3.创建ringBuffer 大小
            int ringBufferSize = 1024 * 1024; // ringBufferSize大小一定要是2的N次方
            // 4.创建Disruptor
            Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(eventFactory, ringBufferSize, executor,
                    ProducerType.SINGLE, new YieldingWaitStrategy());
            // 5.连接消费端方法
            disruptor.handleEventsWith(new LongEventHandler());
            //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, i);
                producer.onData(byteBuffer);
            }
            //10.关闭disruptor和executor
            disruptor.shutdown();
            executor.shutdown();
    
        }
    
    
    }

结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3htMzkzMzkyNjI1_size_16_color_FFFFFF_t_70][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3htMzkzMzkyNjI1_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191231165201397.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3htMzkzMzkyNjI1,size_16,color_FFFFFF,t_70