Disruptor学习（一）——简单使用

深碍√TFBOYSˉ_ 2023-06-30 09:06 12阅读 0赞

[disruptor GitHub 地址][disruptor GitHub]

简介：disruptor是一个开源的高效的并发框架

事件：event  
事件工厂：eventFactory  
事件处理的具体实现：eventHandler  
生产者：Producer

为了使用disruptor,我们考虑一个简单的例子，这个例子就是讲=将一个long型的数字通过生产者传递到消费者，消费者将会简单的打印出值

jar包：最新版本为3.4.2，这里我们使用3.4.1，因为3.4.2是基于Java11进行开发的，我使用的是Java8 ，编译会报错（类文件具有错误的版本号55.0应为52.0）

<dependency>
                <groupId>com.weicoder</groupId>
                <artifactId>disruptor</artifactId>
                <version>3.4.1</version>
            </dependency>

第一步：定义事件

public class LongEvent { 
        private long value;
    
        public void set(long value) { 
            this.value = value;
        }
    }

第二步：定义事件工厂

import com.lmax.disruptor.EventFactory;
    
    public class LongEventFactory implements EventFactory<LongEvent> { 
        @Override
        public LongEvent newInstance() { 
            return new LongEvent();
        }
    }

第三步：定义事件处理器（一旦我们定义了一个事件，那我们就需要创建一个消费者去处理这些事件， 在我们这个例子我们只是想将值打印在控制台）

import com.lmax.disruptor.EventHandler;
    
    public class LongEventHandler implements EventHandler<LongEvent> { 
    
        @Override
        public void onEvent(LongEvent event, long sequence, boolean endOfBatch) throws Exception { 
            System.out.println("Event" + event);
        }
    }

第四步：定义生产者（在Disruptor的3.0版本中，添加了更丰富lambda风格的API，帮助开发人员封装复杂的环形缓冲区，所以在3.0之后发布消息的首选方式就是通过API 的 Event Publisher/Event Translator 部分， 这种方法的另外有个有点就是可以把 翻译器代码放在一个单独的类中，可以轻松的完成独立的单元测试.， Disruptor 提供了许多不同的接口（EventTranslator, EventTranslatorOneArg, EventTranslatorTwoArg等等）可以被实现）

import com.lmax.disruptor.EventTranslatorOneArg;
    import com.lmax.disruptor.RingBuffer;
    
    import java.nio.ByteBuffer;
    
    public class LongEventProducerWithTranslator { 
    
        private final RingBuffer<LongEvent> ringBuffer;
    
        public LongEventProducerWithTranslator(RingBuffer<LongEvent> ringBuffer) { 
            this.ringBuffer = ringBuffer;
        }
    
        private static final EventTranslatorOneArg<LongEvent, ByteBuffer> TRANSLATOR =
                (event, sequence, bb) -> event.set(bb.getLong(0));
    
        public void oneData(ByteBuffer bb){ 
            ringBuffer.publishEvent(TRANSLATOR,bb);
        }
    
    }

生产者还有一种旧版的，更加原始的方法，这种方法与使用简单队列相比，事件发布变得更复杂，这是因为事件需要预分配，它要求（最低级别）消息发布两阶段的方法， 即申明环形缓冲区中的时隙，然后发布可用数据，这需要将发布包装在try/finally 块中，如果我们在这个环形缓冲区申明一个插槽（RingBuffer.next()），则必须发布这个序列，如果不这样做，在多生产者的情况下，这将导致消费者停滞并且无法重新启动就无法恢复。因此，建议使用EventTranslator API，也就是上面那种方法

import com.lmax.disruptor.RingBuffer;
    import java.nio.ByteBuffer;
    
    public class LongEventProducer { 
    
        private  final RingBuffer<LongEvent> ringBuffer;
    
        public LongEventProducer(RingBuffer<LongEvent> ringBuffer) { 
            this.ringBuffer = ringBuffer;
        }
    
        public void oneData(ByteBuffer bb){ 
            long sequence = ringBuffer.next();
            try{ 
                LongEvent event = ringBuffer.get(sequence);
                event.set(bb.getLong(0));
            }finally { 
                ringBuffer.publish(sequence);
            }
        }
    }

第五步：将整个过程连接在一起。可以手动连接所有组件

import com.lmax.disruptor.RingBuffer;
    import com.lmax.disruptor.dsl.Disruptor;
    import com.lmax.disruptor.util.DaemonThreadFactory;
    
    import java.nio.ByteBuffer;
    public class LongEventMain { 
        public static void main(String[] args) throws InterruptedException { 
    
            //事件工厂
            LongEventFactory factory = new LongEventFactory();
    
            //指定环形缓冲区大小，必须是2的幂
            int bufferSize = 1024;
    
            //Disruptor 的构造函数
            Disruptor<LongEvent> disruptor = new Disruptor<>(factory, bufferSize, DaemonThreadFactory.INSTANCE);
    
            //连接处理器
            disruptor.handleEventsWith(new LongEventHandler());
    
            //开始Disruptor ,所有线程开始启动
            disruptor.start();
    
            //从Disruptor获取环形缓冲区以用于发布
            RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
    
            LongEventProducerWithTranslator producer = new LongEventProducerWithTranslator(ringBuffer);
            ByteBuffer bb =  ByteBuffer.allocate(8);
            for (long l = 0; true; l++) { 
                bb.put(0, (byte) l);
                producer.oneData(bb);
                Thread.sleep(1000);
            }
        }
    }

输出结果

Eventcom.liuxin.disruptorlearn.disruptor.LongEvent@193c30a5
    Eventcom.liuxin.disruptorlearn.disruptor.LongEvent@79eb6ada
    Eventcom.liuxin.disruptorlearn.disruptor.LongEvent@286f6ab3
    Eventcom.liuxin.disruptorlearn.disruptor.LongEvent@1526e7d
    Eventcom.liuxin.disruptorlearn.disruptor.LongEvent@7913cfea
    ......

[disruptor GitHub]: https://github.com/LMAX-Exchange/disruptor/wiki/Getting-Started