java - cyclops-react 和 async-retry：如何在超时时重试？

Question

我开始将 cyclops-react 与异步重试一起使用。我还是有点迷茫。

我正在使用 SimpleReact 并模拟来自服务器的超时，但我从未收到类似这样的超时：

private List<Object> executeParallel() {
    List<Object> result = new SimpleReact(mainThreadPool)
            .of(getSupplier())
            .withRetrier(new AsyncRetryExecutor(retryThreadPool)
                    .abortIf((t) -> !TimeoutException.class.isAssignableFrom(t.getClass()))
            )
            .retry(retrySupplier())
            .block()
            .collect(Collectors.toList());
    return result;
}

private Supplier getSupplier() {
    return () -> someOperationThatTimesOut();
}

private Function<Supplier, Object> retrySupplier() {
    return supplier -> supplier.get();
}

那里缺少什么？

Answer 1

这个版本有效

 AtomicInteger count = new AtomicInteger(0);

@Test
public void executeParallel() {
    List<Object> result = new SimpleReact(Executors.newFixedThreadPool(1))
            .of(getSupplier())
            .withRetrier(new AsyncRetryExecutor(Executors.newScheduledThreadPool(1))
            .retry(Supplier::get)
            .block()
            .collect(Collectors.toList());
    System.out.println(result);
}

private Supplier<String> getSupplier() {
    return () -> {
        System.out.println("Attempt " + count.incrementAndGet());
        if(count.get()<4)
            throw ExceptionSoftener.throwSoftenedException(new TimeoutException());
        return "success";
    };
}

它会打印出来

Attempt 1
Attempt 2
Attempt 3
Attempt 4
[success]

我怀疑您不需要异步重试器上的 abortIf，而且我不确定 someOperationThatTimesOut() 内部发生了什么 - 这可能是这里的关键。