使用 Webflux 的反应式 Feign答案

【问题标题】：Reactive Feign with Webflux使用 Webflux 的反应式 Feign
【发布时间】：2021-04-12 07:38:42
【问题描述】：

我从这里 (https://github.com/Playtika/feign-reactive) 使用 Reactive Feign 实现，但即使我尝试使用 Spring WebClient，我也遇到了同样的问题。

当我使用 POST 请求调用 URL 并传递 JSON 正文时，流从未完成。线程保持冻结等待完成，并且永远不会发生。

fun post(request: PostRequest): Mono<PostResponse> {   
    val response = myClient.post(request)
        .onErrorResume {
            log.error("Error sending request", it)
            Mono.error<PostResponse>(it)
        }
        .doOnSuccess {
            log.info("Success with response $it")
            it
        }
    response.block() 
    return response
}

但是，当我打电话给subscribe() 时，什么也没发生，请求仍然被冻结。当我调用block() 方法时，它会抛出以下异常：

java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-nio-4

运行调试，我得到了这个日志：

2021-01-06 11:57:42.047 DEBUG 13624 --- [ctor-http-nio-4] r.client.log.DefaultReactiveLogger       : [MyClient#post]--->POST http://localhost:8081/post HTTP/1.1
2021-01-06 11:57:42.048 DEBUG 13624 --- [ctor-http-nio-4] r.n.resources.PooledConnectionProvider   : [id: 0x77f83c34, L:/10.7.8.88:62005 - R:localhost/127.0.0.1:80] Channel acquired, now 1 active connections and 0 inactive connections
2021-01-06 11:57:42.048 DEBUG 13624 --- [ctor-http-nio-4] reactor.netty.ReactorNetty               : [id: 0x77f83c34, L:/10.7.8.88:62005 - R:localhost/127.0.0.1:80] Added decoder [ReadTimeoutHandler] at the end of the user pipeline, full pipeline: [reactor.left.httpCodec, ReadTimeoutHandler, reactor.right.reactiveBridge, DefaultChannelPipeline$TailContext#0]
2021-01-06 11:57:42.048 DEBUG 13624 --- [ctor-http-nio-4] reactor.netty.ReactorNetty               : [id: 0x77f83c34, L:/10.7.8.88:62005 - R:localhost/127.0.0.1:80] Added decoder [WriteTimeoutHandler] at the end of the user pipeline, full pipeline: [reactor.left.httpCodec, ReadTimeoutHandler, WriteTimeoutHandler, reactor.right.reactiveBridge, DefaultChannelPipeline$TailContext#0]
2021-01-06 11:57:42.048 DEBUG 13624 --- [ctor-http-nio-4] r.netty.http.client.HttpClientConnect    : [id: 0x77f83c34, L:/10.7.8.88:62005 - R:localhost/127.0.0.1:80] Handler is being applied: {uri=http://localhost:8081/post, method=POST}
2021-01-06 11:57:42.049 DEBUG 13624 --- [ctor-http-nio-4] r.n.resources.PooledConnectionProvider   : [id: 0x77f83c34, L:/10.7.8.88:62005 - R:localhost/127.0.0.1:80] onStateChange(POST{uri=/post, connection=PooledConnection{channel=[id: 0x77f83c34, L:/10.7.8.88:62005 - R:localhost/127.0.0.1:80]}}, [request_prepared])
2021-01-06 11:57:42.074 DEBUG 13624 --- [ctor-http-nio-4] r.n.resources.PooledConnectionProvider   : [id: 0x77f83c34, L:/10.7.8.88:62005 - R:localhost/127.0.0.1:80] onStateChange(POST{uri=/post, connection=PooledConnection{channel=[id: 0x77f83c34, L:/10.7.8.88:62005 - R:localhost/127.0.0.1:80]}}, [request_sent])

有办法提出这个请求吗？

【问题讨论】：

你不允许在 webflux 应用程序中进行阻塞，这违背了使用像 Webflux 这样的非阻塞框架的全部目的。
@Toerktumlare 但是没有block()，服务永远不会响应成功或错误。如何使用Spring WebClient重新创建这个请求？我试过了，结果一样
好吧，我无法回答我从未使用过 feign-reactive 并且您发布的代码最少，因此您的问题不可重现且无法回答。如果它冻结，它可能是您的网络，但如上所述，没有人可以回答这个问题，因为它是不可重现的。

标签： spring kotlin spring-webflux spring-webclient reactive-feign-client

【解决方案1】：

以这种方式将 .block() 调用替换为 .map()：

fun post(request: PostRequest): Mono<PostResponse> =
    myClient.post(request)
        .onErrorResume {
            log.error("Error sending request", it)
            Mono.error<PostResponse>(it)
        }
        .map {
          log.info("Success with response $it")
          ok(it)
        }

.map() 是同步的，所以它的作用与 subscribe() 相同。应该从客户端调用订阅，而不是从非阻塞管道上的服务器端调用。使用 .map() 你的管道变得同步，但它不会运行，直到客户端 .subscribes()

【讨论】：