0

当我运行这段代码

    public class Test {
      public static void main(String[] args) {
        Disruptor<MyEvent> disruptor = new Disruptor<MyEvent>(new EventFactoryImpl<MyEvent>(),
    Executors.newFixedThreadPool(2), new MultiThreadedClaimStrategy(32), new                    BusySpinWaitStrategy());

        MyEventHandler myEventHandler1 = new MyEventHandler("1");
        MyEventHandler myEventHandler2 = new MyEventHandler("2");

        disruptor.handleEventsWith(myEventHandler1, myEventHandler2);
        RingBuffer<MyEvent> ringBuffer = disruptor.start();

        ByteBuffer bb = ByteBuffer.allocate(8);

        for (long l = 0; l < 2; l++) {
          bb.putLong(0, l);
          long sequence = ringBuffer.next();

          try {
             MyEvent event = ringBuffer.get(sequence);
             event.set(bb.getLong(0));
          }
          finally {
            ringBuffer.publish(sequence);
          }
        }
      }
    }
    public class MyEvent {
      private long value;

      public void set(long value) {
        this.value = value;
      }

      public long get() {
        return value;
      }
    }

    public class MyEventHandler implements EventHandler<MyEvent> {
       private String id;

       public MyEventHandler(String id) {
          this.id = id;
       }

       public void onEvent(MyEvent event, long sequence, boolean endOfBatch) {
           System.out.println("id: " + id + ", event: " + event.get() + ", sequence: " + sequence +                  "," + Thread.currentThread().getName());
       }
    }

    public class EventFactoryImpl<T> implements EventFactory<T> {
       @SuppressWarnings("unchecked")
       public T newInstance() {
         return (T) new MyEvent();
       }
    }

我得到这个输出

id：1，事件：0，序列：0，pool-1-thread-1
id：1，事件：1，序列：1，pool-1-thread-1
id：2，事件：0，序列：0，pool-1-thread-2
id：2，事件：1，序列：1，pool-1-thread-2

但我希望每个事件只由一个单独的线程处理一次。我怎样才能让它发生？

4

2 回答 2

1

使用 Disruptor，每个EventHandler订阅环形缓冲区的人都会读取每条消息一次。

如果你想让多个线程在消息从环形缓冲区出来时处理它们，有几个选项。第一个也是最好的选择是为每个读取器线程设置一个单独Disruptor的线程，并让写入器以循环方式在缓冲区之间交替。如果您必须使用单个环形缓冲区（可能对事件进行排序），那么您可以将应该处理每个事件的线程 ID 设置到事件本身（再次以交替方式），并让线程与该 ID 不匹配丢弃该事件。

于 2014-03-26T22:58:41.450 回答
1

每个EventHandler人都会处理每个事件。您可以选择链接EventHandlers，以便它们作用于链中先前处理程序的已处理状态。

破坏者还提供了一个WorkerPool旨在将事件传播到一组工作人员的工具。

于 2014-04-01T04:19:21.910 回答