我正在尝试使用RxJava retryWhen 运算符。互联网上关于它的信息很少,唯一值得一提的是this。但这也没有探索我想了解的各种用例。我还加入了异步执行和带退避的重试,以使其更加真实。
我的设置很简单:我有一个类
我期望的是
附注:完全透明 - 我也在 RxJava GitHub 上发布了这个问题。
我的设置很简单:我有一个类
ChuckNorrisJokesRepository
,它从JSON文件返回随机数量的Chuck Norris笑话。我要测试的类是ChuckNorrisJokesService
,如下所示。我感兴趣的用例如下:
- 第一次尝试成功(无需重试)
- 重试一次后失败
- 尝试重试3次,但在第二次成功,因此不会重试第三次
- 第三次重试成功
注意:该项目可在我的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 上发布了这个问题。
range
立即完成,因此跳过了onNext
并直接进入onCompleted
。正如我在回答中所说的那样,我已经重写了代码,不再使用range
。 - Abhijit Sarkarrange
为什么会立即完成?那是因为numRetries
被设置为零了吗(你忘记设置它了)? - Yaroslav StavnichiynumRetries=0
,这是有意为之的(问题中的用例#1)。numRetries=0
表示“我不想重试,我要么接受第一次调用的结果,要么失败”。 - Abhijit Sarkar