Reactor - 如何在不丢失元素的情况下重试热助焊剂？

Question

我有无限的热数据流。我即将对流中的每个元素执行一个操作，每个元素都返回一个 Mono，它将在有限的时间后（以一种或另一种方式）完成。

这些操作可能会引发错误。如果是这样，我想重新订阅热通量而不会丢失任何内容，重试抛出错误时正在处理的元素（即任何未成功完成的元素）。

我在这里做什么？我可以容忍对相同元素的重复操作，但不能完全从流中丢失元素。

我尝试使用 ReplayProcessor 来处理这个问题，但是如果不重复很多可能已经成功的操作（使用非常保守的超时），或者由于丢失元素，我看不到让它工作的方法覆盖缓冲区中旧元素的新元素（如下所示）。

测试用例：

    @Test
    public void fluxTest() {


        List<String> strings = new ArrayList<>();
        strings.add("one");
        strings.add("two");
        strings.add("three");
        strings.add("four");

        ConnectableFlux<String> flux = Flux.fromIterable(strings).publish();

        //Goes boom after three uses of its method, otherwise 
        //returns a mono. completing after a little time
        DangerousClass dangerousClass = new DangerousClass(3);

        ReplayProcessor<String> replay = ReplayProcessor.create(3);

        flux.subscribe(replay);

        replay.flatMap(dangerousClass::doThis)
              .retry(1)
              .doOnNext(s -> LOG.info("Completed {}", s))
              .subscribe();

        flux.connect();

        flux.blockLast();
    }


    public class DangerousClass {

        Logger LOG = LoggerFactory.getLogger(DangerousClass.class);

        private int boomCount;
        private AtomicInteger count;

        public DangerousClass(int boomCount) {
            this.boomCount = boomCount;
            this.count = new AtomicInteger(0);
        }

        public Mono<String> doThis(String s) {
            return Mono.fromSupplier(() -> {
                LOG.info("doing dangerous {}", s);
                if (count.getAndIncrement() == boomCount) {
                    LOG.error("Throwing exception from {}", s);
                    throw new RuntimeException("Boom!");
                }
                return s;
            }).delayElement(Duration.ofMillis(600));
        }
    }

打印出来：

doing dangerous one
doing dangerous two
doing dangerous three
doing dangerous four
Throwing exception from four
doing dangerous two
doing dangerous three
doing dangerous four
Completed four
Completed two
Completed three

一个永远不会完成。

Answer 1

错误（至少在上面的示例中）只能发生在 flatMap(dangerousClass::doThis) 调用中 - 因此重新订阅根 Flux 并在此 flatMap() 调用失败时重播元素似乎有点奇怪，并且（可能）不是你想做的。

相反，我建议放弃 ReplayProcessor，而只在内部 flatMap() 调用上调用 retry，这样你最终会得到类似的结果：

ConnectableFlux<String> flux = Flux.range(1, 10).map(n -> "Entry " + n).publish();

DangerousClass dangerousClass = new DangerousClass(3);

flux.flatMap(x -> dangerousClass.doThis(x).retry(1))
        .doOnNext(s -> System.out.println("Completed " + s))
        .subscribe();

flux.connect();

这将为您提供以下内容，所有条目均已完成且无需重试：

doing dangerous Entry 1
doing dangerous Entry 2
doing dangerous Entry 3
doing dangerous Entry 4
Throwing exception from Entry 4
doing dangerous Entry 4
Completed Entry 2
Completed Entry 1
Completed Entry 3
Completed Entry 4