java - Project Reactor 超时处理

Question

我有三个与 Project Reactor 相关的问题，我将在下面提出。从我拥有的代码开始(它将被简化以便更容易理解问题)。

Mono<Integer> doWithSession(Function<String, Mono<Integer>> callback, long timeout) {
  return Mono.just("hello")
        .compose(monostr -> monostr
            .doOnSuccess(str -> System.out.println("Suppose I want to release session here after all")) //(1)
            .doOnCancel(() -> System.out.println("cancelled")) //(2)
            .then(callback::apply)
            .timeoutMillis(timeout, Mono.error(new TimeoutException("Timeout after " + timeout)))
        );
}

然后测试:

@Test
public void testDoWithSession2() throws Exception {
  Function<String, Mono<Integer>> fun1 = str -> Mono.fromCallable(() -> {
    System.out.println("do some long timed work");
    try {
      Thread.sleep(5000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    System.out.println("work has completed");
    return str.length();
  });

  StepVerifier.create(doWithSession(fun1,1000))
    .verifyError(TimeoutException.class);
}

所以和问题:

1. 如何中断调用fun1并立即返回错误？ (也许我做错了什么，但看起来错误不是在超时后而是在所有回调调用之后返回)
2. 为什么 doOnSuccess和 doOnCancel同时调用？ (我预计将调用 (1) 或 (2) 但不会同时调用两者)
3. 以及如何处理以下情况:
   • 假设在代码中Mono.just("hello")正在获取连接；
   • 在callback我正在做一些有联系的事情并得到一些结果(在我的例子中是 Mono<Integer>)；
   • 最后(成功或失败)我想释放 session (我尝试在 (1) 中这样做)。

Answer 1

1) 如您所见，使用 .publishOn(Schedulers.single()) .这将确保可调用对象在另一个线程上被调用并且只阻塞所述线程。此外，它还允许取消可调用对象。

2) 链的顺序很重要。你把.doOnSuccess在compose的开头(顺便说一句，你并不真正需要那个特定的例子，除非你想提取那个 compose 函数以供以后重用)。所以这意味着它从 Mono.just 获取通知基本上，并在查询源时立即运行，甚至在您的处理发生之前...与 doOnCancel 相同.取消来自 timeout触发...

3) 有一个工厂可以从资源中创建一个序列并确保资源被清理:Mono.using .所以它看起来像这样:

public <T> Mono<T> doWithConnection(Function<String, Mono<T>> callback, long timeout) {
    return Mono.using(
            //the resource supplier:
            () -> {
                System.out.println("connection acquired");
                return "hello";
            },
            //create a Mono out of the resource. On any termination, the resource is cleaned up
            connection -> Mono.just(connection)
                              //the blocking callable needs own thread:
                              .publishOn(Schedulers.single())
                              //execute the callable and get result...
                              .then(callback::apply)
                              //...but cancel if it takes too long
                              .timeoutMillis(timeout)
                              //for demonstration we'll log when timeout triggers:
                              .doOnError(TimeoutException.class, e -> System.out.println("timed out")),
            //the resource cleanup:
            connection -> System.out.println("cleaned up " + connection));
}

返回 Mono<T>可调用的 T 值。在生产代码中，您将订阅它来处理该值。在测试中，StepVerifier.create()将为您订阅。

让我们用你的长时间运行的任务来证明这一点，看看它输出什么:

@Test
public void testDoWithSession2() throws Exception {
    Function<String, Mono<Integer>> fun1 = str -> Mono.fromCallable(() -> {
        System.out.println("start some long timed work");
        //for demonstration we'll print some clock ticks
        for (int i = 1; i <= 5; i++) {
            try {
                Thread.sleep(1000);
                System.out.println(i + "s...");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("work has completed");
        return str.length();
    });

    //let two ticks show up
    StepVerifier.create(doWithConnection(fun1,2100))
                .verifyError(TimeoutException.class);
}

这个输出:

connection acquired
start some long timed work
1s...
2s...
timed out
cleaned up hello

如果我们将超时设置为超过 5000，我们将得到以下结果。 (存在断言错误，因为 StepVerifier 需要超时):

connection acquired
start some long timed work
1s...
2s...
3s...
4s...
5s...
work has completed
cleaned up hello

java.lang.AssertionError: expectation "expectError(Class)" failed (expected: onError(TimeoutException); actual: onNext(5)