如何确保反应堆通量处理提供的所有消息

Question

假设我们想让 Flux 管道来处理从多个线程提供的所有消息。让我们考虑下面的代码：

@Test
public void testFluxCreate() throws InterruptedException {
    EmitterProcessor<String> processor = EmitterProcessor.create();
    CountDownLatch latch = new CountDownLatch(1);

    AtomicLong counter = new AtomicLong();
    AtomicLong batch = new AtomicLong();
    Flux<List<String>> flux = processor
            .doOnSubscribe(ss -> System.out.println(nm() + " : subscribing to + ss))
            .onBackpressureError()
            .buffer(7)
            .publishOn(Schedulers.immediate())
            .doOnNext(it -> {
                counter.addAndGet(it.size());
                System.out.println(batch.incrementAndGet() + " : " + nm() + "Batch: " + it.size());
            })
            ;

    CompletableFuture<Void> producer = CompletableFuture.runAsync(() -> {
        IntStream.range(1, 1001).forEach(it -> {
            //sleep();
            processor.onNext("Message-" + it);
        });
    });

    CompletableFuture<Void> producer2 = CompletableFuture.runAsync(() -> {
        IntStream.range(1, 1001).forEach(it -> {
            //sleep();
            processor.onNext("Message2-" + it);
        });
    });

    CompletableFuture<Void> future = CompletableFuture.allOf(producer, producer2).thenAccept(it -> processor.onComplete());

    flux.doOnComplete(latch::countDown).subscribe();

    future.join();
    latch.await();

    System.out.println("Total: " + counter);
}

计数器告诉我们，每次执行此代码时，实际处理的消息数都是不同的。 这个实现有什么问题？ 我们如何确保在程序结束之前处理所有消息？

Answer 1

这个实现有什么问题？

当我运行代码时，我会在启动后的早期日志中看到以下内容：

18:39:12.590 [ForkJoinPool.commonPool-worker-1] DEBUG reactor.core.publisher.Operators - Duplicate Subscription has been detected
java.lang.IllegalStateException: Spec. Rule 2.12 - Subscriber.onSubscribe MUST NOT be called more than once (based on object equality)
    at reactor.core.Exceptions.duplicateOnSubscribeException(Exceptions.java:162)
    at reactor.core.publisher.Operators.reportSubscriptionSet(Operators.java:502)
    at reactor.core.publisher.Operators.setOnce(Operators.java:607)
    at reactor.core.publisher.EmitterProcessor.onNext(EmitterProcessor.java:245)
    at de.schauder.reactivethreads.demo.StackoverflowQuicky.lambda$null$2(StackoverflowQuicky.java:54)
    at java.util.stream.Streams$RangeIntSpliterator.forEachRemaining(Streams.java:110)
    at java.util.stream.IntPipeline$Head.forEach(IntPipeline.java:557)
    at de.schauder.reactivethreads.demo.StackoverflowQuicky.lambda$main$3(StackoverflowQuicky.java:52)

我不熟悉 EmitterProcessor，但似乎 onNext 不是线程安全的，我强烈怀疑这是导致事件丢失的原因。

我们如何确保在程序结束之前处理完所有消息？

我会使用两个单独的 Producers 和 merge。另外我认为您不需要倒计时闩锁。

public static void main(String[] args) {

    AtomicLong counter = new AtomicLong();
    AtomicLong batch = new AtomicLong();

    EmitterProcessor<String> processor1 = EmitterProcessor.create();
    EmitterProcessor<String> processor2 = EmitterProcessor.create();

    Thread thread1 = constructThread(processor1);
    Thread thread2 = constructThread(processor2);


    Flux<List<String>> flux = processor1.mergeWith(processor2)
            .buffer(7)
            .onBackpressureError()
            .publishOn(Schedulers.immediate())
            .doOnNext(it -> {
                counter.addAndGet(it.size());
                System.out.println(batch.incrementAndGet() + " : Batch: " + it.size());
            }).doOnComplete(() -> {
                System.out.println("Total count: " + counter.get());
            });

    thread1.start();
    thread2.start();

    flux.blockLast();
}

private static Thread constructThread(EmitterProcessor<String> processor) {
    return new Thread(() -> {
        IntStream.range(1, 1001).forEach(it -> {
            processor.onNext("Message2-" + it);
        });
        processor.onComplete();
    });
}

注意我的评论：

onBackpressureError() 导致Flux 在订阅者无法足够快地处理所有事件时发出错误，因此这可以解释不匹配，但您会看到异常。