30、实战项目二：构建实时聊天系统（WebSocket、消息持久化、在线状态管理）

终于到了最后一个实战项目。说实话，实时聊天系统是后端开发里非常经典的一个场景。它不像CRUD那样简单，也不像支付系统那样敏感，但它把网络通信、状态管理、数据一致性这几个硬骨头都揉在了一起。

我个人习惯把这类项目叫做「后端试金石」——能把这个系统做稳了，你的Ktor功底基本就过关了。今天我们就手撸一个轻量级的实时聊天系统，核心围绕三个点：WebSocket通信、消息持久化、在线状态管理。

WebSocket 路由的搭建

Ktor 对 WebSocket 的支持非常原生。你不需要像 Spring 那样配一堆拦截器，直接在路由里写 webSocket 块就行。

// 服务端 WebSocket 路由
fun Application.chatModule() {
    install(WebSockets) {
        pingPeriod = Duration.ofSeconds(15)
        timeout = Duration.ofSeconds(30)
        maxFrameSize = Long.MAX_VALUE
        masking = false
    }

    routing {
        webSocket("/chat/{roomId}") {
            val roomId = call.parameters["roomId"] ?: "default"
            val userId = call.request.queryParameters["userId"] ?: "anonymous"

            // 加入房间
            ChatRoomManager.join(roomId, userId, this)

            try {
                for (frame in incoming) {
                    if (frame is Frame.Text) {
                        val text = frame.readText()
                        val message = ChatMessage(
                            userId = userId,
                            roomId = roomId,
                            content = text,
                            timestamp = Clock.System.now()
                        )
                        // 持久化 + 广播
                        MessageRepository.save(message)
                        ChatRoomManager.broadcast(roomId, message, exclude = null)
                    }
                }
            } catch (e: Exception) {
                log.error("WebSocket error for user $userId: ${e.message}")
            } finally {
                ChatRoomManager.leave(roomId, userId)
            }
        }
    }
}

这里有个细节我提一下：pingPeriod 我设了 15 秒。为什么？因为很多移动端网络不稳定，如果心跳间隔太长，断线了服务端要很久才能感知到。我曾经在生产环境遇到过因为心跳间隔设了 60 秒，导致用户明明掉线了，其他客户端还显示「在线」——那体验真的很糟糕。

消息持久化：别让聊天记录丢了

实时聊天系统最怕什么？消息丢了。用户发了一条消息，对方没收到，服务端也没存——这种事故一旦发生，用户信任感瞬间崩塌。

我用 Exposed 做 ORM，H2 做嵌入式数据库。当然生产环境你可以换成 PostgreSQL，但开发阶段 H2 足够了。

// 消息表定义
object Messages : Table("messages") {
    val id = long("id").autoIncrement()
    val roomId = varchar("room_id", 64)
    val userId = varchar("user_id", 64)
    val content = text("content")
    val timestamp = long("timestamp")

    override val primaryKey = PrimaryKey(id)
}

// 数据访问层
class MessageRepository(private val db: Database) {
    suspend fun save(message: ChatMessage) = dbQuery {
        Messages.insert {
            it[roomId] = message.roomId
            it[userId] = message.userId
            it[content] = message.content
            it[timestamp] = message.timestamp.toEpochMilliseconds()
        }
    }

    suspend fun getRecent(roomId: String, limit: Int = 50) = dbQuery {
        Messages.selectAll()
            .where { Messages.roomId eq roomId }
            .orderBy(Messages.timestamp, SortOrder.DESC)
            .limit(limit)
            .map { row ->
                ChatMessage(
                    userId = row[Messages.userId],
                    roomId = row[Messages.roomId],
                    content = row[Messages.content],
                    timestamp = Instant.fromEpochMilliseconds(row[Messages.timestamp])
                )
            }
            .reversed()
    }
}

你想想看，用户进入聊天室时，看到的应该是历史消息，而不是空白一片。所以 getRecent 这个方法在用户连接成功后就要调用一次，把最近 50 条消息推送给客户端。

在线状态管理：谁在？谁走了？

在线状态管理说白了就是维护一个「谁在哪个房间」的映射表。我用了 ConcurrentHashMap 套 CopyOnWriteArrayList，因为 WebSocket 连接是并发的，普通 HashMap 会有线程安全问题。

object ChatRoomManager {
    // roomId -> (userId -> WebSocketSession)
    private val rooms = ConcurrentHashMap<String, ConcurrentHashMap<String, WebSocketSession>>()

    fun join(roomId: String, userId: String, session: WebSocketSession) {
        val room = rooms.getOrPut(roomId) { ConcurrentHashMap() }
        room[userId] = session
        broadcast(roomId, StatusMessage(userId, "online"))
    }

    fun leave(roomId: String, userId: String) {
        rooms[roomId]?.remove(userId)
        if (rooms[roomId]?.isEmpty() == true) {
            rooms.remove(roomId)
        }
        broadcast(roomId, StatusMessage(userId, "offline"))
    }

    suspend fun broadcast(roomId: String, message: Any, exclude: String? = null) {
        val room = rooms[roomId] ?: return
        val json = Json.encodeToString(message)
        room.forEach { (userId, session) ->
            if (userId != exclude) {
                try {
                    session.send(Frame.Text(json))
                } catch (e: Exception) {
                    // 发送失败，说明这个 session 已经断了
                    leave(roomId, userId)
                }
            }
        }
    }

    fun getOnlineUsers(roomId: String): List<String> {
        return rooms[roomId]?.keys?.toList() ?: emptyList()
    }
}

这里有个坑我曾经踩过：session.send 可能会抛异常，比如客户端已经断开但服务端还没感知到。如果你不 catch 这个异常，整个协程会挂掉，导致其他正常用户也被踢出房间。所以我在 broadcast 里加了 try-catch，一旦发送失败就主动调用 leave。

心跳检测与断线重连

Ktor 自带的 WebSocket 支持 ping-pong 帧。服务端每 15 秒发一个 ping，客户端如果 30 秒内没回 pong，连接就被判定为断开。

但光靠这个还不够。我建议客户端也做一层心跳：每隔 10 秒发一个自定义的 {"type": "heartbeat"} 消息。为什么？因为有些代理服务器（比如某些云厂商的负载均衡）会过滤掉 WebSocket 的 ping 帧，但不会过滤业务消息。

// 客户端心跳示例（伪代码）
fun startHeartbeat(session: WebSocketSession) {
    CoroutineScope(Dispatchers.Default).launch {
        while (isActive) {
            delay(10_000)
            try {
                session.send(Frame.Text("""{"type":"heartbeat"}"""))
            } catch (e: Exception) {
                // 心跳失败，触发重连逻辑
                reconnect()
                break
            }
        }
    }
}

说实话，实时聊天系统看起来功能不多，但要把「消息不丢、状态不错、连接不断」这三个点同时做好，真得花不少心思。我当年第一个版本上线后，光是断线重连的逻辑就改了四版。嗯，但正是这些坑，才让系统越来越稳。

如果你跟着这个项目走下来，你会发现 Ktor 在 WebSocket 这块的体验其实比 Spring 要清爽很多——没有复杂的配置，没有隐式的线程模型，一切都在协程的掌控之中。这就是我喜欢 Kotlin 的原因。

---

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