在java中使用lmax Disruptor（3.0）处理数百万文档

Question

我有以下用例：

当我的服务启动时，它可能需要在尽可能短的时间内处理数百万个文档。将有三个数据来源。

我已设置以下内容：

    /* batchSize = 100, bufferSize = 2^30
    public MyDisruptor(@NonNull final MyDisruptorConfig config) {
        batchSize = config.getBatchSize();
        bufferSize = config.getBufferSize();
        this.eventHandler = config.getEventHandler();
        ThreadFactory threadFactory = createThreadFactory("disruptor-threads-%d");
        executorService = Executors.newSingleThreadExecutor(threadFactory);
        ringBuffer = RingBuffer.createMultiProducer(new EventFactory(), bufferSize, new YieldingWaitStrategy());
        sequenceBarrier = ringBuffer.newBarrier();
        batchEventProcessor = new BatchEventProcessor<>(ringBuffer, sequenceBarrier, eventHandler);
        ringBuffer.addGatingSequences(batchEventProcessor.getSequence());
        executorService.submit(batchEventProcessor);
    }

    public void consume(@NonNull final List<Document> documents) {
        List<List<Document>> subLists = Lists.partition(documents, batchSize);
        for (List<Document> subList : subLists) {
            log.info("publishing sublist of size {}", subList.size());
            long high = ringBuffer.next(subList.size());
            long low = high - (subList.size() - 1);
            long position = low;
            for (Document document: subList) {
                ringBuffer.get(position++).setEvent(document);
            }
            ringBuffer.publish(low, high);
            lastPublishedSequence.set(high);
        }
    }

我的每个源调用consume都使用Guice来创建一个单例破坏者。

我的eventHandler例程是

    public void onEvent(Event event, long sequence, boolean endOfBatch) throws Exception {
        Document document = event.getValue();
        handler.processDocument(document); //send the document to handler
        if (endOfBatch) {
            handler.processDocumentsList(); // tell handler to process all documents so far.
        }
    }

我在日志中看到，制作人（消费）有时会停滞不前。我假设这是当ringBuffer已满，eventHandler无法足够快地处理时。我看到eventHandler正在处理文档（来自我的日志），过了一段时间，生产者开始向环形缓冲区发布更多文档。

问题：

• 我使用的是正确的Disruptor模式吗？我看到有很多方法可以使用它。我选择使用batchEventProcess，这样它就会发出endOfBatch的信号。
• 如何提高EventHandler的效率？进程文档列表可能很慢。
• 我应该使用并行EventHandler吗？lmax用户指南提到这是可能的，常见问题解答对此有一个问题。但是如何将其与batchEventProcess一起使用？它只需要一个eventHandler。




Answer 1

您的处理程序是否有状态？如果没有，则可以使用多个并行事件处理程序来处理文档。您可以实现一个基本的分片策略，其中只有一个处理程序处理每个事件。

endOfBatch通常用于通过优化从批处理中获益的IO操作来加快处理速度。E、 g.在每个事件上写入文件，但仅在endOfBatch上刷新。

如果不知道文档处理器中发生了什么，就很难给出更多建议。