重试（或）重试时似乎不适用于热通量

Question

我正在尝试为我的工作实现反应堆核心。我被困在发生错误时我们需要执行的重试。以下是添加任何错误之前的示例代码

FluxSink<String> mainSink;
// Create the fulx and get handle to Sink
Flux<String> mainFlux = Flux.create(sink -> {
   mainSink = sink;
}, FluxSink.OverflowStrategy.BUFFER);
// Convert to Hot Flux
ConnectableFlux<String> hotFlux = mainFlux.publish();
// Two operations, add A and B to the input
hotFlux.flatMap(o -> Mono.just(o).map(s -> Mono.just(o + "A")))
       .flatMap(o -> Mono.just(o).map(s -> Mono.just(o + "B")))
       .log()
       .subscribe();
// Activate
hotFlux.connect();
// Publish messages to test
Thread.sleep(5000);
int pendingItems = 25;
while(pendingItems > 0) {
     System.out.println("Publishing " + pendingItems + " item");
     mainSink.next(String.valueOf(pendingItems));
     System.out.println("Published " + pendingItems + " item");
     pendingItems--;
}

当我这样做时。它工作正常。

来到错误案例，假设第二个操作（附加“A”）对某个项目失败。我正在尝试获得以下行为。

1. 我尝试添加“A”的部分必须重试 3 次才能放弃
2. 另外我想让整个 Flux 在放弃之前重试 5 次

想知道我怎样才能做到这一点。

AtomicInteger count = new AtomicInteger(0);
FluxSink<String> mainSink;
// Create the fulx and get handle to Sink
Flux<String> mainFlux = Flux.create(sink -> {
   mainSink = sink;
}, FluxSink.OverflowStrategy.BUFFER);
// Convert to Hot Flux
ConnectableFlux<String> hotFlux = mainFlux.publish();
// Two operations, add A and B to the input
hotFlux.flatMap(o -> Mono.just(o).map(s -> {
                  System.out.println("Processing for adding A : " + o);
                  if(count.incrementAndGet() >= 25) {
                       throw new RuntimeException("More than 25th item.. Boom.. !!!");
                  } else {
                       return Mono.just(o + "A")));
                  }
            }).retry(5)
              .doOnError(throwable -> System.out.println("**** Inner Error"))
       ).flatMap(o -> Mono.just(o).map(s -> Mono.just(o + "B")))
       .log()
       .subscribe();
// Activate
hotFlux.connect();
// Publish messages to test
Thread.sleep(5000);
int pendingItems = 25;
while(pendingItems > 0) {
     System.out.println("Publishing " + pendingItems + " item");
     mainSink.next(String.valueOf(pendingItems));
     System.out.println("Published " + pendingItems + " item");
     pendingItems--;
} 

当我如上所示在第一个 flatMap 中添加 retry(5) 时，它可以正常工作，它会为第 25 个进来的人重试 A 的附加 5 次 - 从日志中可以明显看出

我无法实现完整的助焊剂重试（上述要求中的第 (2) 点）。我尝试在第二个通量之后添加一个 .retry(3) ，认为它会重试整个通量。但它似乎没有重试。有人可以帮忙吗？

AtomicInteger count = new AtomicInteger(0);
FluxSink<String> mainSink;
// Create the fulx and get handle to Sink
Flux<String> mainFlux = Flux.create(sink -> {
   mainSink = sink;
}, FluxSink.OverflowStrategy.BUFFER);
// Convert to Hot Flux
ConnectableFlux<String> hotFlux = mainFlux.publish();
// Two operations, add A and B to the input
hotFlux.flatMap(o -> Mono.just(o).map(s -> {
                  System.out.println("Processing for adding A : " + o);
                  if(count.incrementAndGet() >= 25) {
                       throw new RuntimeException("More than 25th item.. Boom.. !!!");
                  } else {
                       return Mono.just(o + "A")));
                  }
            }).retry(5)
              .doOnError(throwable -> System.out.println("**** Inner Error"))
       ).flatMap(o -> Mono.just(o).map(s -> Mono.just(o + "B")))
       .retry(3)
       .log()
       .subscribe();
// Activate
hotFlux.connect();
// Publish messages to test
Thread.sleep(5000);
int pendingItems = 25;
while(pendingItems > 0) {
     System.out.println("Publishing " + pendingItems + " item");
     mainSink.next(String.valueOf(pendingItems));
     System.out.println("Published " + pendingItems + " item");
     pendingItems--;
} 

Answer 1

所有形式的retry 都通过重新订阅“重试”源来工作。它与冷的Flux 一起创造奇迹，但热的Flux 不太适应。

这里有publish() 转换，不能保证迟到的订阅者：因为重试被认为是迟到的订阅者，所以它什么也看不到，因为publish 已被原始完成错误断开。

您需要一种方法来保留最后一项（可能导致异常的项）并为新订阅者重播（或者更确切地说，重试尝试）。

另一个问题是您使用create 来获取您在外部存储的FluxSink，这不是一个好方法。

好消息是这两个问题都可以通过使用ReplayProcessor 一次性解决：您正确地获得了一个专用的接收器来手动推送数据，如果出现错误retry 将能够得到错误-从历史中触发价值并再次尝试：

@Test
public void test() {
    ReplayProcessor<String> foo =
            ReplayProcessor.create(1);
    FluxSink<String> sink = foo.sink();

    foo.subscribe(System.out::println, System.out::println);

    AtomicInteger transientError = new AtomicInteger(5);
    foo.map(v -> "C".equals(v) && transientError.decrementAndGet() >= 0 ? v + (100 / 0) : v)
            .doOnError(e -> System.err.println("Error, should be retried: " + e))
            .retry(5)
            .subscribe(System.err::println, System.err::println);

    sink.next("A");
    sink.next("B");
    sink.next("C");
    sink.complete();
}

打印出来：

A
B
Error, should be retried: java.lang.ArithmeticException: / by zero
Error, should be retried: java.lang.ArithmeticException: / by zero
Error, should be retried: java.lang.ArithmeticException: / by zero
Error, should be retried: java.lang.ArithmeticException: / by zero
Error, should be retried: java.lang.ArithmeticException: / by zero
A
C
B
C