使用 Reactive MongoDB 进行流式传输并且正在取消计时操作

Stream with Reactive MongoDB and time taking operation is being cancelled

问题介于 Project Reactor 和 Reactive MongoDB(Spring 数据)之间。

当执行包含(按以下顺序)的流时:

  1. Reactive MongoDB 上运行的方法非常快
  2. 需要超过 30 秒的方法

流正在被取消(查看下面的代码和日志)

@GetMapping("/test/{msg}")
public Mono<SomeObject> test(@PathVariable String msg) {
    return repository.findByMessage(msg).log("1")
          .map(someObj -> delaySeconds(someObj, 35)).log("2");
}

如您所见,30 秒后流被取消,但又过了 5 秒(超时为 35 秒),事件 onNext 被执行。

12:59:18.556 [Thread-9] INFO  com.why.temp.TempController - Saved:SomeObject(id=5b604106ef301746a86665f3, message=WHY)
12:59:18.591 [http-nio-8080-exec-2] INFO  1 - | onSubscribe([Fuseable] MonoFlatMap.FlatMapMain)
12:59:18.592 [http-nio-8080-exec-2] INFO  2 - | onSubscribe([Fuseable] FluxMapFuseable.MapFuseableSubscriber)
12:59:18.593 [http-nio-8080-exec-2] INFO  2 - | request(unbounded)
12:59:18.593 [http-nio-8080-exec-2] INFO  1 - | request(unbounded)
12:59:18.612 [Thread-8] INFO  1 - | onNext(SomeObject(id=5b604106ef301746a86665f3, message=WHY))
12:59:49.116 [http-nio-8080-exec-3] INFO  2 - | cancel()
12:59:49.117 [http-nio-8080-exec-3] INFO  1 - | cancel()
12:59:53.612 [Thread-8] INFO  2 - | onNext(SomeObject(id=5b604106ef301746a86665f3, message=WHY))

你能解释一下为什么直播被取消了吗?我该如何处理?

是否有任何应该增加的超时,或者我是否以错误的方式使用了 Project Reactor Stream API 和 MongoDB?

这是我的MongoDB配置

@Bean
public ReactiveMongoTemplate reactiveMongoTemplate() {
    ConnectionString str = new ConnectionString(env.getMongoUri());
    return new ReactiveMongoTemplate(MongoClients.create(str), str.getDatabase());
}

有什么想法吗?如果您有类似问题,请点赞此问题。

这个问题的解决方法很简单,但不是那么优雅:

@GetMapping("/test/{msg}")
public Mono<SomeObject> test(@PathVariable String msg) {
    SomeObj someObj = repository.findByMessage(msg).block();
    return Mono.just(someObj).log("1")
        .map(someObj -> delaySeconds(someObj, 35)).log("2");
}

我有一个类似的问题,当反应式操作链花费超过魔法 30 秒时。在我的例子中是 Spring MVC 请求超时,这里是解决方案:

spring.mvc.async.request-timeout的默认值为30s。

我相信它会有所帮助:)。

干杯!