我正在玩 RxJava重试时间 http://reactivex.io/documentation/operators/retry.html操作员。网上关于它的资料很少,唯一值得一提的是this http://blog.danlew.net/2016/01/25/rxjavas-repeatwhen-and-retrywhen-explained。这也不足以探索我想了解的各种用例。我还加入了异步执行和回退重试,以使其更加现实。
我的设置很简单:我有一堂课ChuckNorrisJokesRepository
从 JSON 文件返回随机数量的 Chuck Norris 笑话。我的测试类是ChuckNorrisJokesService
如下所示。我感兴趣的用例如下:
- 第一次尝试成功(无需重试)
- 重试 1 次后失败
- 尝试重试 3 次,但第二次成功,因此不会重试第三次
- 第三次重试成功
Note: 该项目可以在我的GitHub https://github.com/asarkar/java/blob/master/concurrency-learning/src/main/java/org/abhijitsarkar/rx/ChuckNorrisJokesService.java.
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 https://github.com/asarkar/java/blob/master/concurrency-learning/src/test/groovy/org/abhijitsarkar/rx/ChuckNorrisJokesServiceSpec.groovy对于其他测试用例。
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
。我缺少什么?
P.S.:完全披露 - 我也将这个问题发布在RxJava GitHub https://github.com/ReactiveX/RxJava/issues/4207.