java Reactor retry不调用要重试的方法

Question

我有一个方法总是返回Mono.error

private Mono<String> doSomething() {
        System.out.println("doSomething");
        return Mono.error(new Error());
}

我尝试了三次重试执行

Mono<String> mono =
                myClass.doSomething()
                        .doOnSubscribe(x -> System.out.println("Subscribe"))
                        .retryWhen(companion -> companion
                                .doOnNext(s -> System.out.println(s + " at " + LocalTime.now()))
                                .zipWith(Flux.range(1, 4), (error, index) -> {
                                    if (index < 4) return index;
                                    else throw Exceptions.propagate(error);
                                })
                                .flatMap(index -> Mono.delay(Duration.ofMillis(index * 100)))
                                .doOnNext(s -> System.out.println("retried at " + LocalTime.now()))
                        );

        mono.block();

但是，doSomething只打印一次，而Subscribe打印4次（开始时打印1次，重试3次）

如果不执行该方法，它如何重新订阅

我想做的是在每次重试时执行doSomething

Answer 1

要在每次重试时执行doSomething，可以将^{}运算符与^{}结合使用：

Mono.defer(() -> myClass.doSomething()
                        .doOnSubscribe(x -> System.out.println("Subscribe")))
    .retryWhen(// rest of the code

基本上，^{}操作符通过重新订阅Mono源来工作，^{}操作符在每次有新订阅时重新评估lambda。所以，最终你会得到想要的行为

Answer 2

doSomething()只有一次调用——也就是说，由您显式地检索基于错误的Mono。每次重试时，返回的Mono确实会被再次订阅，但最初返回它的方法不会被再次调用

这里更可取的做法是重构代码，以便在订阅Mono时发生任何副作用，例如：

private Mono<String> doSomething() {
    return Mono.<String>error(new Error()).doOnSubscribe(s -> System.out.println("doSomething"));
}

但是，如果这不可能，那么您可以将doSomething()作为供应商使用，并将结果平展，从而在每次重试时按原样执行。如果你想沿着这条路线走下去，那就换掉：

myClass.doSomething()

。。。与：

Mono.fromSupplier(myClass::doSomething).flatMap(p->p)