11、WebSocket实时通信：WebSocket配置、消息处理、广播机制、心跳检测

WebSocket 这东西，说白了就是让服务器和客户端之间能「随时说话」。不像 HTTP 那样一问一答，WebSocket 建立连接后，两边都能主动推数据。我在做即时通讯项目时，第一次体会到这种「双向自由」有多爽——用户发消息，服务器能立刻推送给对方，延迟基本感觉不到。

Ktor 对 WebSocket 的支持很原生，不需要额外引入复杂的框架。你只要加一个插件，写几行配置，就能跑起来。嗯，咱们一步步来。

WebSocket 配置：从依赖到路由

首先，你得在 build.gradle.kts 里加上 WebSocket 的依赖：

implementation("io.ktor:ktor-server-websockets:$ktor_version")

然后，在 Application 模块里安装插件：

fun Application.module() {
    install(WebSockets) {
        pingPeriod = Duration.ofSeconds(15)    // 心跳间隔
        timeout = Duration.ofSeconds(15)       // 超时时间
        maxFrameSize = Long.MAX_VALUE          // 最大帧大小
        masking = false                        // 是否掩码
    }
}

这里有个细节：pingPeriod 和 timeout 是配套使用的。服务器每隔 15 秒发一个 ping 帧，如果客户端 15 秒内没回 pong，连接就断了。我刚开始做的时候把 timeout 设成了 60 秒，结果用户网络波动时，连接要等很久才断开，体验很差。后来改成 15 秒，舒服多了。

路由配置也很简单：

routing {
    webSocket("/chat") {
        // 这里处理连接
    }
}

你看，一个 webSocket 函数就搞定了。路径可以随便定义，比如 /ws、/live，看你业务需求。

消息处理：接收与发送

连接建立后，核心就是处理消息。Ktor 的 DefaultWebSocketSession 提供了 incoming 和 outgoing 两个通道。我习惯用 for 循环来监听：

webSocket("/chat") {
    try {
        for (frame in incoming) {
            when (frame) {
                is Frame.Text -> {
                    val text = frame.readText()
                    println("收到消息: $text")
                    // 处理业务逻辑
                    send(Frame.Text("服务器已收到: $text"))
                }
                is Frame.Binary -> {
                    val bytes = frame.readBytes()
                    // 处理二进制数据
                }
                is Frame.Ping -> {
                    // 心跳响应，Ktor 自动处理
                }
                else -> {
                    // 其他帧类型
                }
            }
        }
    } catch (e: Exception) {
        println("连接异常: ${e.localizedMessage}")
    }
}

这里要注意：incoming 是一个 ReceiveChannel，它会阻塞直到有消息到来。如果你在循环里做耗时操作，记得用协程。我曾经踩过一个坑——在消息处理里直接调数据库查询，结果把整个连接线程卡住了。后来改成 launch 异步处理，问题解决。

广播机制：让所有客户端都收到

广播是 WebSocket 的常见需求——比如聊天室里，一个人发言，所有人都能看到。Ktor 没有内置广播功能，但我们可以自己实现。我常用的方案是维护一个连接池：

object ConnectionPool {
    private val connections = Collections.synchronizedSet<DefaultWebSocketSession>()

    fun add(session: DefaultWebSocketSession) {
        connections.add(session)
    }

    fun remove(session: DefaultWebSocketSession) {
        connections.remove(session)
    }

    suspend fun broadcast(message: String) {
        connections.forEach { session ->
            try {
                session.send(Frame.Text(message))
            } catch (e: Exception) {
                // 发送失败，移除该连接
                remove(session)
            }
        }
    }
}

然后在路由里这样用：

webSocket("/chat") {
    ConnectionPool.add(this)
    try {
        for (frame in incoming) {
            if (frame is Frame.Text) {
                val text = frame.readText()
                // 广播给所有人
                ConnectionPool.broadcast("用户说: $text")
            }
        }
    } finally {
        ConnectionPool.remove(this)
    }
}

你想想看，这个模式其实很简单。每个连接进来时注册，断开时注销。广播时遍历所有连接，逐个发送。但要注意并发问题——synchronizedSet 保证了线程安全，但遍历时如果连接被移除，可能会抛异常。所以我加了 try-catch，发送失败就自动清理。

心跳检测：保持连接活跃

WebSocket 连接看起来是长连接，但网络环境复杂。路由器可能超时断开，移动网络可能切换 IP，这些都会导致连接「假死」。心跳检测就是用来发现这些问题的。

Ktor 的 WebSockets 插件已经内置了心跳机制——就是前面配置的 pingPeriod。服务器定期发 Ping 帧，客户端自动回 Pong 帧。如果客户端没响应，Ktor 会关闭连接并触发 onClose 回调。

但有时候，你需要在应用层做更精细的控制。比如，客户端也要主动发心跳，证明自己还活着。我一般这样处理：

webSocket("/chat") {
    // 启动一个协程，定期发送心跳
    val heartbeatJob = launch {
        while (isActive) {
            delay(10_000) // 每10秒发一次
            try {
                send(Frame.Text("{\"type\":\"heartbeat\"}"))
            } catch (e: Exception) {
                break
            }
        }
    }

    try {
        for (frame in incoming) {
            when (frame) {
                is Frame.Text -> {
                    val text = frame.readText()
                    // 如果是心跳响应，忽略
                    if (text.contains("heartbeat")) continue
                    // 处理业务消息
                }
            }
        }
    } finally {
        heartbeatJob.cancel()
        ConnectionPool.remove(this)
    }
}

为什么要在应用层做？因为有些代理服务器会过滤掉 WebSocket 的 Ping/Pong 帧。我在一个企业项目里就遇到过，客户的内网防火墙把 Ping 帧当异常流量拦截了，结果连接一直断。后来改成应用层心跳，用普通文本消息，问题解决。

整体架构图

下面这张图展示了 WebSocket 实时通信的完整流程，从连接建立到消息广播，再到心跳维护：

从图上你能看到，客户端先建立 WebSocket 连接，服务器维护一个连接池。消息进来后，经过处理器，再广播给所有客户端。心跳检测是独立运行的，定期检查连接健康状态。

避坑指南

最后，分享几个我实际项目中踩过的坑：

• 连接泄漏：客户端断开时，服务器没及时清理连接。我曾经在 finally 块里忘了调用 remove，结果连接池越来越大，内存飙升。后来加了监控，每次连接数超过阈值就报警。
• 并发发送：多个协程同时往一个 session 发消息，可能导致帧错乱。Ktor 的 send 是线程安全的，但如果你自己维护队列，记得加锁。
• 大消息处理：WebSocket 默认帧大小有限制。如果你要传大文件，建议分片发送，或者用二进制流。我试过直接传 10MB 的图片，结果连接直接断开。
• Nginx 代理：如果用了 Nginx 做反向代理，记得配置 proxy_set_header Upgrade $http_upgrade 和 proxy_set_header Connection "upgrade"，否则 WebSocket 握手会失败。

WebSocket 实时通信，说白了就是「连接管理 + 消息分发 + 心跳维护」。Ktor 把这套流程简化了很多，但底层的原理你还是要懂。嗯，掌握了这些，你就能做出一个稳定可靠的实时通信系统了。