Spring webflux 反应式 Mono::subscribe

Posted 2023-02-27

【中文标题】Spring webflux 反应式 Mono::subscribe

【英文标题】：Spring webflux reactive Mono::subscribe

【发布时间】：2021-11-18 06:13:22

【问题描述】：

问题陈述：

我正在尝试在请求转发到下游系统之前将会话保存在 redis 中。

 return new OrderedGatewayFilter((exchange, chain) -> exchange.getSession().doOnNext(session -> 
     // some code

).map(WebSession::save).doOnNext(Mono::subscribe).then(chain.filter(exchange)), SAVE_SESSION_FILTER);

但是 Mono::subscribe 是异步的本质，不能保证 session 会先存储在 redis 中，然后再转发到下游。我们在使用并发请求执行性能测试时遇到了这个问题。

有没有办法让它同步。

【问题讨论】：

你为什么叫单声道订阅？调用客户端是订阅者，您的应用程序是客户端订阅的生产者。它应该被完全删除。

【参考方案1】：

您可能正在寻找 flatMap 运算符，这使得 subscribe 变得不必要并且在完成之前不会继续流程：

return new OrderedGatewayFilter((exchange, chain) ->
    exchange.getSession()
        .doOnNext(session -> /* some code*/)
        .flatMap(WebSession::save)
        .then(chain.filter(exchange)), SAVE_SESSION_FILTER);