订阅带有 projectreactor 阻塞的 Flux Websocket 入站连接？答案

【问题标题】：Subscribe to Flux Websocket inbound connection with projectreactor blocking?订阅带有 projectreactor 阻塞的 Flux Websocket 入站连接？
【发布时间】：2023-03-28 21:27:02
【问题描述】：

在下面的代码中，IntelliJ 警告不应在阻塞范围内调用订阅。不幸的是，订阅似乎是将消费者与入站消息流相关联的最直观的方式，有没有更好的方式？

Kotlin 中的代码 sn-p，基于 projectreactor documentation 中的示例 Java 代码。

我想通过注入的消费者订阅入站消息，或者以其他消费者可以访问和订阅的方式公开入站消息通量，我不希望这被阻塞。

import io.netty.buffer.Unpooled
import io.netty.util.CharsetUtil
import reactor.core.publisher.Flux
import reactor.netty.http.client.HttpClient

fun main() {
    HttpClient.create()
        .websocket()
        .uri("wss://echo.websocket.org")
        .handle { inbound, outbound ->
            inbound.receive()
                .asString()
                .take(1)
                .subscribe(
                    { println(it) },
                    { println("error $it") },
                    { println("completed") }
                )

            val msgBytes = "hello".toByteArray(CharsetUtil.ISO_8859_1)
            outbound.send(Flux.just(Unpooled.wrappedBuffer(msgBytes))).neverComplete()
        }
        .blockLast()
}

【问题讨论】：

标签： intellij-idea websocket nonblocking project-reactor reactor-netty

【解决方案1】：

我们找到了一种非阻塞的订阅替代方案。 then 和 zip。 Kotlin 中的示例。

import io.netty.buffer.Unpooled
import io.netty.util.CharsetUtil.UTF_8
import reactor.core.publisher.Flux
import reactor.netty.http.client.HttpClient

fun main() {
    val outgoingMessagesFlux = Flux.just(Unpooled.wrappedBuffer("hello".toByteArray(UTF_8)))

    HttpClient.create()
        .websocket()
        .uri("wss://echo.websocket.org")
        .handle { inbound, outbound ->
            val thenInbound = inbound.receive()
                .asString()
                .doOnNext { println(it) }
                .then()

            val thenOutbound = outbound.send(outgoingMessagesFlux).neverComplete()
            Flux.zip(thenInbound, thenOutbound).then()
        }.blockLast()
}

这是基于Spring WebFlux Netty websocket client 源代码实现和当前的spring-framework documentation。

【讨论】：