第81章 WebSocket：Ktor WebSocket客户端/服务端、实时通信、心跳机制

WebSocket，说白了就是让客户端和服务端之间开一条「长连接」。

不像HTTP那样一问一答，WebSocket允许双方随时推送数据。我最早接触它是在做一个股票行情App的时候，那时候轮询HTTP接口，每秒请求一次，服务器差点没被我们打挂。后来换成WebSocket，整个世界清净了。

WebSocket 是什么？为什么需要它？

传统的HTTP通信，客户端发请求，服务端返回响应。如果你想实时获取数据，只能轮询——每隔几秒发一次请求。这效率太低了。

WebSocket 解决了这个问题。它建立一次连接后，双方可以随时互发消息。你想想看，聊天、推送、游戏、实时监控……这些场景天生就该用 WebSocket。

Ktor 中的 WebSocket 支持

Ktor 是 JetBrains 出品的 Kotlin 异步框架。它原生支持 WebSocket，服务端和客户端都有对应的插件。

我个人习惯用 Ktor 做 WebSocket 项目，因为它的 DSL 风格写起来很舒服，而且和 Kotlin 协程无缝集成。

服务端实现

先看服务端。我们需要引入 Ktor 的 WebSocket 插件。

// build.gradle.kts
implementation("io.ktor:ktor-server-websockets:$ktorVersion")
implementation("io.ktor:ktor-server-netty:$ktorVersion")

服务端代码长这样：

fun main() {
    embeddedServer(Netty, port = 8080) {
        install(WebSockets) {
            pingPeriod = 15.seconds  // 心跳间隔
            timeout = 15.seconds
            maxFrameSize = Long.MAX_VALUE
        }

        routing {
            webSocket("/chat") {
                send("欢迎连接！")
                for (frame in incoming) {
                    when (frame) {
                        is Frame.Text -> {
                            val text = frame.readText()
                            println("收到: $text")
                            send("你说的是: $text")
                        }
                    }
                }
            }
        }
    }.start(wait = true)
}

嗯，这里要注意 pingPeriod 这个参数。它决定了服务端多久发一次心跳包。我刚开始做的时候没配这个，结果客户端那边连接经常莫名其妙断开。后来才发现是防火墙把空闲连接给掐了。

客户端实现

客户端同样需要引入依赖：

implementation("io.ktor:ktor-client-websockets:$ktorVersion")
implementation("io.ktor:ktor-client-cio:$ktorVersion")

客户端代码：

suspend fun main() {
    val client = HttpClient(CIO) {
        install(WebSockets)
    }

    client.webSocket("ws://localhost:8080/chat") {
        send("你好，服务端！")

        for (frame in incoming) {
            when (frame) {
                is Frame.Text -> {
                    println("服务端回复: ${frame.readText()}")
                }
            }
        }
    }
}

你看，客户端和服务端的写法几乎对称。这也是 Ktor 设计得好的地方——你不需要学两套 API。

心跳机制：为什么需要它？

WebSocket 连接虽然叫「长连接」，但网络环境并不总是可靠的。

我曾经遇到过一个问题：客户端显示连接正常，但服务端早就把连接回收了。原因就是中间的路由器或防火墙把空闲连接给断了，但客户端不知道。

心跳机制就是用来解决这个问题的。双方定期发送一个很小的数据包，告诉对方「我还活着」。如果一段时间没收到心跳，就可以认为连接断了，主动重连。

Ktor 中的心跳配置

Ktor 服务端通过 pingPeriod 配置心跳。客户端呢？

客户端其实不需要额外配置，因为 Ktor 的 WebSocket 客户端会自动响应服务端的 Ping 帧。但如果你想在客户端主动发心跳，可以这样：

client.webSocket("ws://localhost:8080/chat") {
    // 启动一个协程定期发心跳
    launch {
        while (true) {
            delay(10.seconds)
            send(Frame.Ping(ByteArray(0)))
        }
    }

    for (frame in incoming) {
        // 处理消息
    }
}

实时通信的完整流程

我们来看一个完整的实时聊天示例。服务端维护所有连接的客户端列表，收到消息后广播给所有人。

val clients = Collections.synchronizedSet<WebSocketSession>()

fun main() {
    embeddedServer(Netty, port = 8080) {
        install(WebSockets) {
            pingPeriod = 15.seconds
        }

        routing {
            webSocket("/chat") {
                clients.add(this)
                try {
                    for (frame in incoming) {
                        if (frame is Frame.Text) {
                            val msg = frame.readText()
                            // 广播给所有客户端
                            clients.forEach { it.send(msg) }
                        }
                    }
                } finally {
                    clients.remove(this)
                }
            }
        }
    }.start(wait = true)
}

客户端这边，我们用一个协程接收消息，另一个协程发送消息：

client.webSocket("ws://localhost:8080/chat") {
    // 接收消息
    launch {
        for (frame in incoming) {
            if (frame is Frame.Text) {
                println("收到: ${frame.readText()}")
            }
        }
    }

    // 发送消息
    launch {
        while (true) {
            val input = readLine() ?: break
            send(input)
        }
    }
}

SVG 架构图：WebSocket 实时通信流程

断连重连：生产环境必备

真实项目中，网络不可能永远稳定。客户端需要实现断连重连逻辑。

我一般这样写：

suspend fun connectWithRetry(client: HttpClient, url: String) {
    var retryCount = 0
    while (true) {
        try {
            client.webSocket(url) {
                retryCount = 0  // 连接成功，重置重试次数
                for (frame in incoming) {
                    // 处理消息
                }
            }
        } catch (e: Exception) {
            retryCount++
            val delay = minOf(retryCount * 2L, 30L)  // 指数退避，最多30秒
            println("连接断开，${delay}秒后重试...")
            delay(delay.seconds)
        }
    }
}

总结

WebSocket 让实时通信变得简单。Ktor 的实现很优雅，服务端和客户端用同一套 DSL，学习成本低。

关键点就三个：

• 建立连接后，用 incoming 和 send 收发消息
• 心跳机制保活，防止连接被意外断开
• 断连重连用指数退避，别把服务器搞崩

你想想看，如果不用 WebSocket，实时推送得写多少轮询代码？Ktor 帮你把这些都封装好了，你只需要关注业务逻辑就行。

---

公众号:蓝海资料掘金营，微信deep3321