介绍 disruptor 事件监听器的串行操作与并行操作方式，源码下载

代码：

实体类：

  1. package com.dmbjz.height.chain;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  6. import lombok.experimental.Accessors;
  8. import java.util.concurrent.atomic.AtomicInteger;
  10. /* 需要操作的数据实体类 */
  11. @Data
  12. @AllArgsConstructor
  13. @NoArgsConstructor
  14. @Accessors(chain = true)
  15. public class Trade {
  17.     private String id;
  18.     private String name;
  19.     private double price;
  20.     private AtomicInteger count = new AtomicInteger(0);
  22. }

事件执行方法：

  1. package com.dmbjz.height.chain;
  3. import com.lmax.disruptor.EventTranslator;
  5. import java.util.Random;
  7. /** 事件执行方法 */
  8. public class TradeEventTranslator implements EventTranslator<Trade> {
  10.     private Random random = new Random();
  12.     @Override
  13.     public void translateTo(Trade event, long sequence) {
  14.         event.setPrice(random.nextDouble() * 9999);
  15.     }
  17. }

生产者代码：

  1. package com.dmbjz.height.chain;
  3. import com.lmax.disruptor.dsl.Disruptor;
  4. import lombok.AllArgsConstructor;
  5. import lombok.Data;
  6. import lombok.NoArgsConstructor;
  7. import lombok.experimental.Accessors;
  9. import java.util.concurrent.CountDownLatch;
  12. /* 生产者 */
  13. @Data
  14. @AllArgsConstructor
  15. @NoArgsConstructor
  16. @Accessors(chain = true)
  17. public class TradePushlisher implements Runnable {
  19.     private Disruptor<Trade> disruptor;
  20.     private CountDownLatch latch;
  22.     @Override
  23.     public void run() {
  25.         TradeEventTranslator eventTranslator = new TradeEventTranslator();
  26.         for(int i =0; i < 1; i ++){
  27.             //新的提交任务的方式
  28.             disruptor.publishEvent(eventTranslator);            
  29.         }
  30.         latch.countDown();
  32.     }
  34. }

事件监听器一：

  1. package com.dmbjz.height.chain;
  3. import com.lmax.disruptor.EventHandler;
  4. import com.lmax.disruptor.WorkHandler;
  6. import java.util.concurrent.TimeUnit;
  9. /* 监听事件，使用 WorkHandler 与 EventHandler 进行事件监听
  10. *         WorkHandler 相较于 EventHandler仅需要一个参数
  11. */
  12. public class UserHandler1 implements EventHandler<Trade>, WorkHandler<Trade>{
  14.     //EventHandler方法
  15.     @Override
  16.     public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {
  17.         this.onEvent(event);
  18.     }
  20.     //WorkHandler方法
  21.     @Override
  22.     public void onEvent(Trade event) throws Exception {
  23.         System.out.println("UserHandler1  : SET NAME");
  24.         TimeUnit.SECONDS.sleep(1);
  25.         event.setName("H1");
  26.     }
  28. }

事件监听器二：

  1. package com.dmbjz.height.chain;
  3. import com.lmax.disruptor.EventHandler;
  5. import java.util.UUID;
  6. import java.util.concurrent.TimeUnit;
  9. /* 监听事件 */
  10. public class UserHandler2 implements EventHandler<Trade> {
  12.     @Override
  13.     public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {
  14.         System.out.println("UserHandler2 : SET ID");
  15.         TimeUnit.SECONDS.sleep(1);
  16.         event.setId(UUID.randomUUID().toString());
  17.     }
  19. }

事件监听器三：

  1. package com.dmbjz.height.chain;
  3. import com.lmax.disruptor.EventHandler;
  6. /* 监听事件 */
  7. public class UserHandler3 implements EventHandler<Trade> {
  9.     @Override
  10.     public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {
  11.         System.out.println("UserHandler3 : NAME: "
  12.                                 + event.getName() 
  13.                                 + ", ID: " 
  14.                                 + event.getId()
  15.                                 + ", PRICE: " 
  16.                                 + event.getPrice()
  17.                                 + " INSTANCE : " + event.toString());
  18.     }
  20. }

方法案例：

串行方法：

使用链式调用设置事件监听器，事件监听器将会按顺序依次执行

  1. package com.dmbjz.height.chain;
  3. import com.lmax.disruptor.BusySpinWaitStrategy;
  4. import com.lmax.disruptor.EventFactory;
  5. import com.lmax.disruptor.RingBuffer;
  6. import com.lmax.disruptor.dsl.Disruptor;
  7. import com.lmax.disruptor.dsl.ProducerType;
  8. import sun.nio.ch.ThreadPool;
  10. import java.util.concurrent.*;
  12. public class Main {
  15.     public static void main(String[] args) throws Exception {
  17.         //构建一个线程池用于提交任务
  18.         ExecutorService taskPool = Executors.newFixedThreadPool(1);
  19.         //创建线程池用于构建Disruptor
  20.         ThreadPoolExecutor threadPool = new ThreadPoolExecutor(3,5,10,TimeUnit.MICROSECONDS,new ArrayBlockingQueue<>(10));
  22.         //构建Disruptor
  23.         Disruptor<Trade> disruptor = new Disruptor<Trade>(
  24.                 () -> new Trade(),
  25.                 1024*1024,
  26.                 threadPool,
  27.                 ProducerType.SINGLE,
  28.                 new BusySpinWaitStrategy()
  29.         );
  32.         /* 把消费者设置到Disruptor中 handleEventsWith */
  34.         //串行操作：按添加顺序执行监听器
  35.         disruptor.handleEventsWith(new UserHandler1())
  36.             .handleEventsWith(new UserHandler2())
  37.             .handleEventsWith(new UserHandler3());
  39.         //启动disruptor
  40.         RingBuffer<Trade> ringBuffer = disruptor.start();
  42.         //使用 CountDownLatch 确保资源流释放
  43.         CountDownLatch countDownLatch = new CountDownLatch(1);
  45.         long begin = System.currentTimeMillis();
  47.         //提交任务
  48.         taskPool.submit(new TradePushlisher(disruptor,countDownLatch));
  51.         try {
  52.             countDownLatch.await();
  53.         } finally {
  54.             disruptor.shutdown();
  55.             taskPool.shutdown();
  56.             threadPool.shutdown();
  57.         }
  59.         System.out.println("总耗时: " + (System.currentTimeMillis() - begin));
  62.     }
  64. }

image.png

并行方法：

使用并行方法执行事件监听器

  1. package com.dmbjz.height.chain;
  3. import com.lmax.disruptor.BusySpinWaitStrategy;
  4. import com.lmax.disruptor.EventFactory;
  5. import com.lmax.disruptor.RingBuffer;
  6. import com.lmax.disruptor.dsl.Disruptor;
  7. import com.lmax.disruptor.dsl.ProducerType;
  8. import sun.nio.ch.ThreadPool;
  10. import java.util.concurrent.*;
  12. public class Main {
  15.     public static void main(String[] args) throws Exception {
  17.         //构建一个线程池用于提交任务
  18.         ExecutorService taskPool = Executors.newFixedThreadPool(1);
  19.         //创建线程池用于构建Disruptor
  20.         ThreadPoolExecutor threadPool = new ThreadPoolExecutor(3,5,10,TimeUnit.MICROSECONDS,new ArrayBlockingQueue<>(10));
  22.         //构建Disruptor
  23.         Disruptor<Trade> disruptor = new Disruptor<Trade>(
  24.                 () -> new Trade(),
  25.                 1024*1024,
  26.                 threadPool,
  27.                 ProducerType.SINGLE,
  28.                 new BusySpinWaitStrategy()
  29.         );
  32.         /* 把消费者设置到Disruptor中 handleEventsWith */
  34.         //并行操作写法一：同时执行监听器
  35.         disruptor.handleEventsWith(new UserHandler1());
  36.         disruptor.handleEventsWith(new UserHandler2());
  37.         disruptor.handleEventsWith(new UserHandler3());
  39.         //并行操作写法2:
  40.         //disruptor.handleEventsWith(new UserHandler1(),new UserHandler2(),new UserHandler3());
  43.         //启动disruptor
  44.         RingBuffer<Trade> ringBuffer = disruptor.start();
  46.         //使用 CountDownLatch 确保资源流释放
  47.         CountDownLatch countDownLatch = new CountDownLatch(1);
  49.         long begin = System.currentTimeMillis();
  51.         //提交任务
  52.         taskPool.submit(new TradePushlisher(disruptor,countDownLatch));
  55.         try {
  56.             countDownLatch.await();
  57.         } finally {
  58.             disruptor.shutdown();
  59.             taskPool.shutdown();
  60.             threadPool.shutdown();
  61.         }
  63.         System.out.println("总耗时: " + (System.currentTimeMillis() - begin));
  66.     }
  68. }


image.png