Disruptor并发框架使用二

╰半夏微凉° 2023-03-02 05:07 95阅读 0赞

一、Disruptor场景使用  
    我们在创建Disruptor框架实例的时候，我们可以直接使用RingBuffer，以及其他的API操作。  
    场景1：使用EventProcessor消息处理器或者使用WorkerPool消息处理器

使用EventProcessor消息处理器示例：

public class ProductDto  { 
        private String  id;
        private String name;
        private BigDecimal price;
        private AtomicInteger count = new AtomicInteger();
    }

public class ProductHandler implements  WorkHandler<ProductDto>,EventHandler<ProductDto> { 
    
        public void onEvent(ProductDto event,long sequence,long endOfBatch) throws Exception { 
            this.onEvent(event);
        }
    
    
        @Override
        public void onEvent(ProductDto event) throws Exception { 
            // 此处做具体的消费逻辑
            event.setId(UUID.randomUUID().toString());
            System.out.println(event.getId());
        }
     }

public class Main1 { 
    
        @Test
        public  void test01() throws ExecutionException, InterruptedException { 
            final  int buffer_size = 1024 ;
            final  int threads_size = 4;
    
            /** createSingleProducer 创建一个单生产者的ringbuffer * 第一个参数：事件工厂，主要是产生数据和填充Ringbuffer区域 * 第二个参数：指的是RingBuffer的大小，必须是2的指数倍，目的是求模运算转化为&运算提高效率 * 第三个参数：RingBuffer的生产都在没有可用区块的时候（可能消费者或者事件处理器），太慢的等待策略， */
           final RingBuffer<ProductDto> ringBuffer = RingBuffer.createSingleProducer(new EventFactory<ProductDto>() { 
                @Override
                public ProductDto newInstance() { 
                    return new ProductDto();
                }
            }, buffer_size, new YieldingWaitStrategy());
    
           // 线程池
            ExecutorService executors = Executors.newFixedThreadPool(threads_size);
    
            // 创建sequenceBarrier
            SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
    
            // 创建消息处理器
            BatchEventProcessor<ProductDto> transProcessor = new BatchEventProcessor<ProductDto>(ringBuffer, sequenceBarrier, new ProductHandler());
    
            // 目的是把消费者的位置信息引用注入到生产者，如果只有一个消费者可以忽略
            ringBuffer.addGatingSequences(transProcessor.getSequence());
    
            // 把消息处理器提交到线程池
           executors.submit(transProcessor);
    
           // 如果存在多个消费者，那重复执行上面的3行代码，把ProductHandler转化为其他的消费类
            Future<Void> future = executors.submit(new Callable<Void>() { 
                @Override
                public Void call() throws Exception { 
                    long seq;
                    for (int i = 0; i < 10; i++) { 
                        seq = ringBuffer.next(); // ringbuffer 一个可用区块，先占一个位置
                        ringBuffer.get(seq).setPrice(new BigDecimal(Math.random() * 111)); // 给此区块放入数据
                        ringBuffer.publish(seq); // 发布这个区块的数据使handler(consumer)可见
    
                    }
                    return null;
                }
            });
            future.get(); // 等到生产者结束
            Thread.sleep(1000); // 等待1秒，消费者都处理完成
            transProcessor.halt(); // 通知事件(消息)处理器，可以结束了，并不代表马上结束
            executors.shutdown(); // 终止线程
        }
    }

public class Main2 {   
        public static void main(String[] args) throws InterruptedException {   
            int BUFFER_SIZE=1024;  
            int THREAD_NUMBERS=4;  
            
            EventFactory<Trade> eventFactory = new EventFactory<Trade>() {   
                public Trade newInstance() {   
                    return new Trade();  
                }  
            };  
            
            RingBuffer<Trade> ringBuffer = RingBuffer.createSingleProducer(eventFactory, BUFFER_SIZE);  
              
            SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();  
              
            ExecutorService executor = Executors.newFixedThreadPool(THREAD_NUMBERS);  
              
            WorkHandler<Trade> handler = new TradeHandler();  
    
            WorkerPool<Trade> workerPool = new WorkerPool<Trade>(ringBuffer, sequenceBarrier, new IgnoreExceptionHandler(), handler);  
              
            workerPool.start(executor);  
              
            //下面这个生产8个数据
            for(int i=0;i<8;i++){   
                long seq=ringBuffer.next();  
                ringBuffer.get(seq).setPrice(Math.random()*9999);  
                ringBuffer.publish(seq);  
            }  
              
            Thread.sleep(1000);  
            workerPool.halt();  
            executor.shutdown();  
        }  
    }