使用 Reactive MongoDB 进行流式传输，并且正在取消耗时操作答案

【问题标题】：Stream with Reactive MongoDB and time taking operation is being cancelled使用 Reactive MongoDB 进行流式传输，并且正在取消耗时操作
【发布时间】：2019-01-07 19:33:42
【问题描述】：

问题介于 Project Reactor 和 Reactive MongoDB（Spring Data）之间。

执行包含（按以下顺序）的流时：

在Reactive MongoDB上运行的方法非常快
耗时超过 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");
}

【问题讨论】：

在我的脑海中，我想说的是，最有可能的是，并非 msg 的所有字段都被索引。这可能会导致集合扫描，这取决于集合的大小 - 可能需要很长时间。
在这种情况下，repository.findByMessage(msg) 非常快，但如果流中的下一步需要超过 30 秒 (delaySeconds(someObj, 35 ) 在这种情况下)，所有流都被取消。我想某些连接过期了，应该在 findByMessage 操作之后关闭。

标签： spring mongodb reactive-programming project-reactor

【解决方案1】：

我有一个类似的问题，当反应操作链花费了超过 30 秒的时间。在我的情况下，这是 Spring MVC 请求超时，这是解决方案：

Recurring AsyncRequestTimeoutException in Spring Boot Admin log

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

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

干杯！

【讨论】：

你是对的，application.yml 中的配置帮助：spring:mvc:async:request-timeout:180s