LMAX Disruptor Producer在消费者完成之前错误地包装覆盖阅读

Question

我最近被介绍给LMAX Disruptor，并决定试一试。多亏了开发人员，安装速度很快，没有任何麻烦。但如果有人能帮我，我想我遇到了一个问题。

问题：有人告诉我，当制作人发布活动时，它应该阻止，直到消费者有机会在包装之前取回它。我在消费者方面有一个序列障碍，我可以确认，如果生产者没有发布数据，消费者的waitFor呼叫将被阻止。但是，生产者似乎并没有受到任何监管，只会在环形缓冲区中封装并覆盖未处理的数据。

我有一个producer作为一个可运行的对象，在单独的线程上运行。

public class Producer implements Runnable {
    private final RingBuffer<Event> ringbuffer;
    public Producer(RingBuffer<Event> rb) {
        ringbuffer = rb;
    }
    public void run() {
           long next = 0L;
           while(true) {
               try {
                   next = ringbuffer.next();
                   Event e = ringbuffer.get(next);
                   ... do stuff...
                   e.set(... stuff...);
               }
               finally {
                   ringbuffer.publish(next);
               }
           }
    }
}

我有一个消费者在主线程上运行。

public class Consumer {
     private final ExecutorService exec;
     private final Disruptor<Event> disruptor;
     private final RingBuffer<Event> ringbuffer;
     private final SequenceBarrier seqbar;
     private long seq = 0L;

     public Consumer() {
         exec = Executors.newCachedThreadPool();
         disruptor = new Disruptor<>(Event.EVENT_FACTORY, 1024, Executors.defaultThreadFactory());
         ringbuffer = disruptor.start();
         seqbar = ringbuffer.newBarrier();

         Producer producer = new Producer(ringbuffer);
         exec.submit(producer);
    }

    public Data getData() {
         seqbar.waitFor(seq);
         Event e = ringbuffer.get(seq);
         seq++;
         return e.get();
    }
}

最后，我像这样运行代码：

public class DisruptorTest {
     public static void main(String[] args){
         Consumer c = new Consumer();
         while (true) {
             c.getData();
             ... Do stuff ...
         }
}

Answer 1

您需要向ringBuffer添加一个门控序列（com.lmax.disruptor.Sequence），该序列必须根据您的消费者的位置进行更新。

您可以使用EventHandler接口和提供的内置序列的BatchEventProcessor（com.lmax.disruptor.BatchEventProcessor.BatchEventProcessor）来实现事件处理

这是一个完全可行的例子

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import com.lmax.disruptor.BatchEventProcessor;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.dsl.Disruptor;

public class Main {

   static class Event {

      int id;
   }

   static class Producer implements Runnable {

      private final RingBuffer<Event> ringbuffer;

      public Producer(RingBuffer<Event> rb) {
         ringbuffer = rb;
      }

      @Override
      public void run() {
         long next = 0L;
         int id = 0;
         while (true) {
            try {
               next = ringbuffer.next();
               Event e = ringbuffer.get(next);
               e.id = id++;
            } finally {
               ringbuffer.publish(next);
            }
         }
      }
   }

   static class Consumer {

      private final ExecutorService exec;
      private final Disruptor<Event> disruptor;
      private final RingBuffer<Event> ringbuffer;
      private final SequenceBarrier seqbar;
      private BatchEventProcessor<Event> processor;

      public Consumer() {
         exec = Executors.newCachedThreadPool();
         disruptor = new Disruptor<>(() -> new Event(), 1024, Executors.defaultThreadFactory());
         ringbuffer = disruptor.start();
         seqbar = ringbuffer.newBarrier();

         processor = new BatchEventProcessor<Main.Event>(
               ringbuffer, seqbar, new Handler());
         ringbuffer.addGatingSequences(processor.getSequence());

         Producer producer = new Producer(ringbuffer);
         exec.submit(producer);
      }
   }

   static class Handler implements EventHandler<Event> {

      @Override
      public void onEvent(Event event, long sequence, boolean endOfBatch) throws Exception {
         System.out.println("Handling event " + event.id);
      }

   }

   public static void main(String[] args) throws Exception {

      Consumer c = new Consumer();
      while (true) {
         c.processor.run();
      }
   }
}