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

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

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);
      }
  }
复制代码

//消费者 
 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());
    }
}
复制代码

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();
  }
复制代码