java多线程入门（六）Disruptor使用-阿里云开发者社区

java多线程入门（六）Disruptor使用

2022-06-14 664

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： java多线程入门（六）Disruptor使用

1.简单认识下Disruptor

Disruptor是一款高效的无锁内存队列。它使用无锁的方式实现了一个环形队列，非常适合生产者和消费者模式，比如事件和消息的发布。

2.demo

2.1首先要声明一个消息传递的类

public class MsgEvent{
 private String value;
}
复制代码

2.2声明一个生产者

-需要一个Disruptor

class MsgProducer {
  private Disruptor<MsgEvent> disruptor;
  public MsgProducer(Disruptor<MsgEvent> disruptor){
      this.disruptor = disruptor;
  }
  public void send(String data){
      //得到队列
      RingBuffer<MsgEvent> ringBuffer = this.disruptor.getRingBuffer();
      //ringBuffer是个队列，其next方法返回的是下最后一条记录之后的位置，这是个可用位置
      long next = ringBuffer.next();
      try{
          //取出事件位置是空事件
          MsgEvent event = ringBuffer.get(next);
          //空的事件位置设置值
          event.setValue(data);
      }finally {
        //发布
        ringBuffer.publish(next);
      }
  }
复制代码

2.3声明一个消费者

-消费者实现接口EventHandler，实现onEvent方法

//消费者 
 class MyConsumer implements EventHandler<MsgEvent>{
  private String name;
  @Override
  public void onEvent(MsgEvent msgEvent, long l, boolean b) throws Exception {
      System.out.println(this.name+" -> 接收到信息： "+msgEvent.getValue());
  }
}
复制代码

//消费者
class MyWorkHandelr implements WorkHandler<MsgEvent>{
   private String name;
    public MyWorkHandelr(String name){
        this.name = name;
    }
    @Override
    public void onEvent(MsgEvent msgEvent) throws Exception {
        System.out.println(this.name+" -> 接收到信息： "+msgEvent.getValue());
    }
}
复制代码

2.4 测试

public static void test(){
  // RingBuffer生产工厂,初始化RingBuffer的时候使用
      EventFactory<MsgEvent> factory = new EventFactory<MsgEvent>() {
          @Override
          public MsgEvent newInstance() {
              return new MsgEvent();
          }
      };
         Disruptor<MsgEvent> disruptor = new Disruptor<>(factory, 1024, Executors.defaultThreadFactory(), ProducerType.SINGLE, new BlockingWaitStrategy());
      //绑定配置关系
      disruptor.handleEventsWith(new MyConsumer("cm_1"), new MyConsumer("cm_2"), new MyConsumer("cm_3"));
      disruptor.start();
      // 定义要发送的数据
      MsgProducer msgProducer = new MsgProducer(disruptor);
      for(int i=0;i<10;i++) {
        msgProducer.send(""+i);
      }
      disruptor.shutdown();
  }
复制代码

BlockingWaitStrategy

加锁;这是默认策略，这里用了锁和条件condition进行数据监控和线程的唤醒，这个策略最节省cpu但是高并发下性能最差

SleepingWaitStrategy

自旋、yield、sleep;对延时要求不是特别高的场合，对生产者线程的影响最小，典型用于异步日志类似的场景。

-YieldingWaitStrategy

自旋、yield、自旋;性能最好，适合用于低延迟的系统，在要求极高性能且之间处理线数小于 cpu 逻辑核心数的场景中，推荐使用。（无锁策略。主要是使用了 Thread.yield() 多线程交替执行）

BusySpinWaitStrategy

自旋;通过不断重试，减少切换线程导致的系统调用，而降低延迟。推荐在线程绑定到固定的CPU的场景下使用

TimeoutBlockingWaitStrategy

加锁，有超时限制; CPU资源紧缺，吞吐量和延迟并不重要的场景

PhasedBackoffWaitStrategy

自旋、yield、自定义策略; CPU资源紧缺，吞吐量和延迟并不重要的场景

3.使用的场景

网络异常，图片无法展示

//c1,c2,c3单独消费

disruptor.handleEventsWith(new MyConsumer("cm_1"), new MyConsumer("cm_2"), new MyConsumer("cm_3"));

网络异常，图片无法展示

//c3依赖 c1和c2

disruptor.handleEventsWith(new MyConsumer("cm_1"), new MyConsumer("cm_2")).then(new MyConsumer("cm_3"));

//先c1再c2、c3

disruptor.handleEventsWith(new MyConsumer("cm_1")).then(new MyConsumer("cm_2"),new MyConsumer("cm_3"));

网络异常，图片无法展示

//共同消费c1,c2再c3

disruptor.handleEventsWithWorkerPool(new MyWorkHandelr("cm_1"), new MyWorkHandelr("cm_2")).thenHandleEventsWithWorkerPool(new MyWorkHandelr("cm_3"));