RxJava retryWhen的奇怪行为

Question

我正在尝试使用RxJava retryWhen 运算符。互联网上关于它的信息很少，唯一值得一提的是this。但这也没有探索我想了解的各种用例。我还加入了异步执行和带退避的重试，以使其更加真实。

我的设置很简单：我有一个类ChuckNorrisJokesRepository，它从JSON文件返回随机数量的Chuck Norris笑话。我要测试的类是ChuckNorrisJokesService，如下所示。我感兴趣的用例如下：

1. 第一次尝试成功（无需重试）
2. 重试一次后失败
3. 尝试重试3次，但在第二次成功，因此不会重试第三次
4. 第三次重试成功

注意：该项目可在我的GitHub上获取。

ChuckNorrisJokesService.java:

@Slf4j
@Builder
public class ChuckNorrisJokesService {
    @Getter
    private final AtomicReference<Jokes> jokes = new AtomicReference<>(new Jokes());

    private final Scheduler scheduler;
    private final ChuckNorrisJokesRepository jokesRepository;
    private final CountDownLatch latch;
    private final int numRetries;
    private final Map<String, List<String>> threads;

    public static class ChuckNorrisJokesServiceBuilder {
        public ChuckNorrisJokesService build() {
            if (scheduler == null) {
                scheduler = Schedulers.io();
            }

            if (jokesRepository == null) {
                jokesRepository = new ChuckNorrisJokesRepository();
            }

            if (threads == null) {
                threads = new ConcurrentHashMap<>();
            }

            requireNonNull(latch, "CountDownLatch must not be null.");

            return new ChuckNorrisJokesService(scheduler, jokesRepository, latch, numRetries, threads);
        }
    }

    public void setRandomJokes(int numJokes) {
        mergeThreadNames("getRandomJokes");

        Observable.fromCallable(() -> {
            log.debug("fromCallable - before call. Latch: {}.", latch.getCount());
            mergeThreadNames("fromCallable");
            latch.countDown();

            List<Joke> randomJokes = jokesRepository.getRandomJokes(numJokes);
            log.debug("fromCallable - after call. Latch: {}.", latch.getCount());

            return randomJokes;
        }).retryWhen(errors ->
                errors.zipWith(Observable.range(1, numRetries), (n, i) -> i).flatMap(retryCount -> {
                    log.debug("retryWhen. retryCount: {}.", retryCount);
                    mergeThreadNames("retryWhen");

                    return Observable.timer(retryCount, TimeUnit.SECONDS);
                }))
                .subscribeOn(scheduler)
                .subscribe(j -> {
                            log.debug("onNext. Latch: {}.", latch.getCount());
                            mergeThreadNames("onNext");

                            jokes.set(new Jokes("success", j));
                            latch.countDown();
                        },
                        ex -> {
                            log.error("onError. Latch: {}.", latch.getCount(), ex);
                            mergeThreadNames("onError");
                        },
                        () -> {
                            log.debug("onCompleted. Latch: {}.", latch.getCount());
                            mergeThreadNames("onCompleted");

                            latch.countDown();
                        }
                );
    }

    private void mergeThreadNames(String methodName) {
        threads.merge(methodName,
                new ArrayList<>(Arrays.asList(Thread.currentThread().getName())),
                (value, newValue) -> {
                    value.addAll(newValue);

                    return value;
                });
    }
}

为简洁起见，我只展示了第一个用例的Spock测试案例。请查看我的GitHub获取其他测试案例。

def "succeeds on 1st attempt"() {
    setup:
    CountDownLatch latch = new CountDownLatch(2)
    Map<String, List<String>> threads = Mock(Map)
    ChuckNorrisJokesService service = ChuckNorrisJokesService.builder()
            .latch(latch)
            .threads(threads)
            .build()

    when:
    service.setRandomJokes(3)
    latch.await(2, TimeUnit.SECONDS)

    Jokes jokes = service.jokes.get()

    then:
    jokes.status == 'success'
    jokes.count() == 3

    1 * threads.merge('getRandomJokes', *_)
    1 * threads.merge('fromCallable', *_)
    0 * threads.merge('retryWhen', *_)
    1 * threads.merge('onNext', *_)
    0 * threads.merge('onError', *_)
    1 * threads.merge('onCompleted', *_)
}

这会失败并显示以下错误：

Too few invocations for:

1 * threads.merge('fromCallable', *_)   (0 invocations)
1 * threads.merge('onNext', *_)   (0 invocations)

我期望的是 fromCallable 只被调用一次，成功后先调用 onNext 一次，然后调用 onCompleted。我漏掉了什么？

附注：完全透明 - 我也在 RxJava GitHub 上发布了这个问题。

- Abhijit Sarkar

Answer 1

我在经过数小时的疑难解答和 ReactiveX 成员 David Karnok 的帮助后解决了这个问题。

retryWhen 是一个复杂的、甚至可能有缺陷的操作符。官方文档和至少一个回答中使用了 range 操作符，如果没有重试，则会立即完成。参见我的 discussion 与 David Karnok。

代码可在我的 GitHub 上找到，并包含以下测试用例：

1. 第一次尝试成功（无需重试）
2. 第一次重试失败
3. 尝试重试 3 次，但第二次成功，因此不进行第三次重试
4. 第三次重试成功

- Abhijit Sarkar

---

那种“奇怪的行为”是什么原因引起的？ - Yaroslav Stavnichiy

@YaroslavStavnichiy，您可以查看我的回答中的链接获取详细信息，但简而言之，range立即完成，因此跳过了onNext并直接进入onCompleted。正如我在回答中所说的那样，我已经重写了代码，不再使用range。 - Abhijit Sarkar

我这样问是因为我在你的回答中没有看到这一点，也许你应该更新一下。而且range为什么会立即完成？那是因为numRetries被设置为零了吗（你忘记设置它了）？ - Yaroslav Stavnichiy

@YaroslavStavnichiy，我的答案中已经有了“范围运算符，在没有重试的情况下会失败”的内容。我现在已经重新表述了这个语句，使其更加清晰明了。至于numRetries=0，这是有意为之的（问题中的用例#1）。numRetries=0表示“我不想重试，我要么接受第一次调用的结果，要么失败”。 - Abhijit Sarkar

@Abhijit，你的 Github 链接已经失效了。 - Nathan Schwermann