1. 简介
本文将介绍 RSocket 协议以及如何在 Kotlin 中使用它。RSocket 是一个支持响应式流语义的网络通信协议，非常适合构建异步、响应式的网络应用。我们将使用 Kotlin 的 Ktor 框架和 Kotlinx Coroutines 来实现 RSocket 的客户端与服务端。
2. 什么是 RSocket？
RSocket 定义为“提供响应式流语义的应用层协议”。它支持多种语言的实现，允许我们建立各种类型的响应式网络流。这使得我们可以编写出能够以多种方式进行通信的网络应用。
作为应用层协议，RSocket 允许我们用不同的编程语言编写客户端和服务端，并确保它们之间可以正确通信。
本文中我们使用的 Kotlin 实现基于 Ktor 和 Kotlinx Coroutines。
3. 准备工作
在使用 RSocket 之前，需要先完成一些准备工作，包括引入正确的依赖项，并在代码中设置基础组件。
3.1 设置服务端
服务端需要引入以下依赖（以 JVM 为目标运行时）：
<dependency>
<groupId>io.rsocket.kotlin</groupId>
<artifactId>rsocket-core</artifactId>
<version>0.15.4</version>
</dependency>
<dependency>
<groupId>io.ktor</groupId>
<artifactId>ktor-server-netty-jvm</artifactId>
<version>2.3.7</version>
</dependency>
<dependency>
<groupId>io.ktor</groupId>
<artifactId>ktor-server-cio-jvm</artifactId>
<version>2.3.7</version>
</dependency>
<dependency>
<groupId>io.rsocket.kotlin</groupId>
<artifactId>rsocket-ktor-server-jvm</artifactId>
<version>0.15.4</version>
</dependency>
接下来，我们需要创建一个 Ktor 服务，并启用 WebSocket 和 RSocket 支持：
embeddedServer(Netty, port = 9000) {
install(io.ktor.server.websocket.WebSockets)
install(io.rsocket.kotlin.ktor.server.RSocketSupport)
routing {
// 添加 RSocket 接口
}
}.start(wait = true)
⚠️ 注意：必须先安装 WebSockets 扩展，再安装 RSocketSupport，否则会出错。
3.2 设置客户端
客户端需要引入以下依赖：
<dependency>
<groupId>io.rsocket.kotlin</groupId>
<artifactId>rsocket-core</artifactId>
<version>0.15.4</version>
</dependency>
<dependency>
<groupId>io.ktor</groupId>
<artifactId>ktor-client-cio-jvm</artifactId>
<version>2.3.7</version>
</dependency>
<dependency>
<groupId>io.rsocket.kotlin</groupId>
<artifactId>rsocket-ktor-client-jvm</artifactId>
<version>0.15.4</version>
</dependency>
然后创建一个 RSocket 客户端：
val client = HttpClient {
install(io.ktor.client.plugins.websocket.WebSockets)
install(io.rsocket.kotlin.ktor.client.RSocketSupport)
}
同样地，必须先安装 WebSockets，再安装 RSocketSupport。
4. Fire and Forget（单向通信）
Fire and Forget 是 RSocket 提供的最基础操作之一。客户端发送请求后不等待响应，适用于事件通知等场景。
服务端处理
rSocket("/rsocket/fireAndForget") {
RSocketRequestHandler {
fireAndForget { request: Payload ->
val text = request.data.readText()
println("收到请求: $text")
}
}
}
客户端调用
val rSocket: RSocket = client.rSocket(host = "localhost", port = 9000, path = "/rsocket/fireAndForget")
rSocket.fireAndForget(buildPayload { data("Hello") })
✅ 客户端发送消息时不需要等待响应。
5. Request/Response（请求-响应）
有时我们需要服务端返回结果，这就需要使用 Request/Response 模式。
服务端处理
rSocket("/rsocket/requestResponse") {
RSocketRequestHandler {
requestResponse { request: Payload ->
val text = request.data.readText()
println("收到请求: $text")
delay(Duration.ofSeconds(5)) // 模拟耗时处理
buildPayload { data(text.reversed()) }
}
}
}
客户端调用
val rSocket: RSocket = client.rSocket(port = 9000, path = "/rsocket/requestResponse")
val response = rSocket.requestResponse(buildPayload { data("Hello") })
val text = response.data.readText()
println("收到响应: $text")
⚠️ 注意：如果客户端使用 fireAndForget 调用 requestResponse 类型的接口，服务端会抛出异常。
6. Request Stream（请求-流）
除了单次请求响应，RSocket 还支持从服务端向客户端发送数据流。
服务端处理
rSocket("/rsocket/requestStream") {
RSocketRequestHandler {
requestStream { request: Payload ->
val text = request.data.readText()
println("收到请求: $text")
flow {
processData(text)
}
}
}
}
辅助函数：
suspend fun FlowCollector<Payload>.processData(text: String) {
for (i in 0..10) {
val data = "data: ($text)$i"
println("发送: $data")
emitOrClose(buildPayload { data(data) })
delay(Duration.ofMillis(500))
}
}
客户端调用
val rSocket: RSocket = client.rSocket(port = 9000, path = "/rsocket/requestStream")
val stream = rSocket.requestStream(buildPayload { data("Hello") })
stream.onEach { frame ->
val text = frame.data.readText()
println("收到数据: $text")
}.launchAndForget()
✅ 客户端使用 Flow 接收服务端推送的流数据。
7. Request Channel（双向流）
RSocket 还支持双向流通信，即客户端和服务端都可以发送流数据。
服务端处理
rSocket("/rsocket/requestChannel") {
RSocketRequestHandler {
requestChannel { request: Payload, payloads: Flow<Payload> ->
val text = request.data.readText()
println("收到初始请求: $text")
payloads.onEach { frame ->
val payloadText = frame.data.readText()
println("客户端发来: $payloadText")
}.launchIn(this)
flow {
processData(text)
}
}
}
}
客户端调用
val rSocket: RSocket = client.rSocket(port = 9000, path = "/rsocket/requestChannel")
val stream = rSocket.requestChannel(buildPayload { data("Hello") }, flow { produceData() })
stream.onEach { frame ->
val text = frame.data.readText()
println("收到服务端消息: $text")
}.launchAndForget()
辅助函数：
suspend fun FlowCollector<Payload>.produceData() {
for (i in 1..5) {
val data = "来自客户端的数据 $i"
emitOrClose(buildPayload { data(data) })
delay(Duration.ofMillis(300))
}
}
✅ 双向流允许客户端和服务端各自独立地发送数据流。
8. 小结
本文介绍了如何在 Kotlin 中使用 RSocket 实现多种通信模式，包括：
- Fire and Forget（单向通知）
- Request/Response（请求-响应）
- Request Stream（服务端流）
- Request Channel（双向流）
RSocket 结合 Ktor 和 Kotlinx Coroutines 提供了强大的响应式网络通信能力。下次需要构建异步、流式通信的应用时，不妨尝试一下 RSocket。
所有示例代码可在 GitHub 上找到。