本章导读
一个聊天室需要解决三个并发问题：① 如何同时处理多个 TCP 连接（每连接一线程）；② 如何安全地共享在线用户列表（互斥锁）；③ 如何将一条消息广播给所有人（遍历发送）。本章实现一个完整的命令行聊天室：支持昵称、加入/离开通知、在线列表、私聊、消息广播。
本章学习目标
- 设计文本行协议（以 \n 分隔消息）
- 实现线程安全的客户端注册表
- 实现消息广播（排除发送者）
- 实现每连接一线程的并发模型
- 实现客户端的收发分离（读线程 + 主线程写）
- 处理连接断开与资源清理
57.1 架构设计
// 聊天室架构
//
// ┌────────────────────────────────────────────┐
// │ ChatServer │
// │ ┌──────────┐ ┌───────────────────────┐ │
// │ │ TcpServer│ │ Hub (消息中心) │ │
// │ │ :9000 │ │ clients: list │ │
// │ └────┬─────┘ │ mutex: 互斥锁 │ │
// │ │ │ broadcast(msg) │ │
// │ │ accept │ register(client) │ │
// │ ▼ │ unregister(client) │ │
// │ ┌─────────┐ └───────────────────────┘ │
// │ │ Thread-1│──→ ClientHandler │
// │ │ Thread-2│──→ ClientHandler │
// │ │ Thread-N│──→ ClientHandler │
// │ └─────────┘ │
// └────────────────────────────────────────────┘
//
// 每个客户端连接 → 一个独立线程
// 线程职责：读取消息 → 交给 Hub 广播核心组件：TcpServer（监听）、Hub（管理客户端 + 广播）、ClientHandler（每连接线程）。
57.2 协议设计
使用最简单的文本行协议——每条消息以 \n 结尾：
// 协议格式（每条消息一行，\n 分隔）
//
// 客户端 → 服务器：
// "NICK alice\n" 设置昵称
// "MSG hello everyone\n" 发送群聊消息
// "WHO\n" 查询在线列表
// "TELL bob hi\n" 私聊
// "QUIT\n" 退出
//
// 服务器 → 客户端：
// "WELCOME 欢迎来到聊天室\n"
// "JOIN alice 加入了聊天室\n"
// "LEAVE alice 离开了\n"
// "MSG alice: hello everyone\n"
// "WHO alice, bob, charlie\n"
// "TELL bob: hi\n"
// "ERROR 未知命令\n"
// 为什么选择文本行协议？
// 1. 简单——recv() 天然以 \n 为消息边界
// 2. 可调试——telnet/nc 直接连接测试
// 3. 可扩展——加字段只需加前缀
协议设计原则
好的协议应该：① 有明确的消息边界（\n）；② 有消息类型前缀（MSG/WHO/TELL）；③ 人类可读（方便调试）；④ 易于解析（split 即可）。
57.3 客户端结构
// 表示一个已连接的客户端
class ClientInfo
TcpStream conn // TCP 连接
string nick // 昵称
int id // 唯一 ID
func make(TcpStream conn, int id) as ClientInfo
ClientInfo c = ClientInfo()
c.conn = conn
c.nick = "guest" + id.toString()
c.id = id
return c
end
end
// 发送消息给此客户端（加 \n 结尾）
func sendTo(ClientInfo client, string msg)
client.conn.send(msg + "\n")
client.conn.drain(3000)
end每个 ClientInfo 封装了一个 TCP 连接和对应的昵称。所有发送都通过 sendTo 统一处理。
57.4 消息中心（Hub）
Hub 是线程安全的核心——管理所有在线客户端：
class Hub
list<ClientInfo> clients // 在线客户端列表
Mutex lock // 互斥锁保护 clients
int nextId // 自增 ID
func init()
clients = list<ClientInfo>()
lock = Mutex()
nextId = 1
end
// 注册新客户端（线程安全）
func register(TcpStream conn) as ClientInfo
lock.lock()
ClientInfo c = ClientInfo.make(conn, nextId)
nextId += 1
clients.add(c)
lock.unlock()
return c
end
// 注销客户端（线程安全）
func unregister(ClientInfo client)
lock.lock()
int i = 0
while i < clients.len()
if clients[i].id == client.id then
clients.removeAt(i)
// 不 i += 1，因为 removeAt 后元素前移
else
i += 1
end
end
lock.unlock()
end
end关键：所有对 clients 列表的读写都在 lock 保护下进行，避免多线程竞争。
57.5 消息广播
// 广播消息给所有在线客户端（可排除发送者）
func broadcast(Hub hub, string msg, int excludeId)
hub.lock.lock()
int i = 0
while i < hub.clients.len()
ClientInfo c = hub.clients[i]
if c.id != excludeId then
sendTo(c, msg)
end
i += 1
end
hub.lock.unlock()
end
// 广播给所有人（包括发送者）
func broadcastAll(Hub hub, string msg)
broadcast(hub, msg, -1) // -1 表示不排除任何人
end
// 获取在线列表
func whoList(Hub hub) as string
hub.lock.lock()
string result = ""
int i = 0
while i < hub.clients.len()
if i > 0 then result += ", " end
result += hub.clients[i].nick
i += 1
end
hub.lock.unlock()
return result
end
// 按昵称查找客户端
func findByNick(Hub hub, string nick) as ClientInfo
hub.lock.lock()
ClientInfo found = nil
int i = 0
while i < hub.clients.len()
if hub.clients[i].nick == nick then
found = hub.clients[i]
end
i += 1
end
hub.lock.unlock()
return found
end广播时持锁遍历——简单可靠。对于高并发场景（万级连接），可以改用 channel 或无锁队列，但本章的规模下互斥锁足够。
57.6 连接处理线程
每个客户端连接由独立线程处理：
// 处理单个客户端的完整生命周期
func handleClient(Hub hub, TcpStream conn)
// 1. 注册
ClientInfo me = hub.register(conn)
sendTo(me, "WELCOME 你的临时昵称是 " + me.nick)
sendTo(me, "输入 NICK 名字 设置昵称，MSG 内容 发消息，WHO 查看在线")
// 2. 消息循环
bool running = true
while running
string line = conn.recv(60000) // 60 秒超时
if line.len() == 0 then
running = false // 连接断开
else
line = string.trim(line)
if line.len() > 0 then
running = processCommand(hub, me, line)
end
end
end
// 3. 清理
broadcast(hub, "LEAVE " + me.nick + " 离开了聊天室", me.id)
hub.unregister(me)
conn.close()
end生命周期三阶段：注册（加入 + 欢迎）→ 循环（读消息 + 处理命令）→ 清理（通知 + 注销 + 关闭）。
57.7 服务器主循环
func startServer(int port)
Hub hub = Hub()
hub.init()
TcpServer server = TcpServer(tcp.listen(uint16(port)))
println("聊天室服务器启动，端口 " + port.toString())
println("等待连接...")
// 主循环：接受连接，每连接启动一个线程
while true
point accepted = server.accept(30000) // 30 秒超时等待
if accepted.valid() then
TcpStream conn = TcpStream(accepted)
// 启动新线程处理此连接
thread.start(func()
handleClient(hub, conn)
end)
println("新连接: " + conn.peerAddr())
end
end
end
// 启动
startServer(9000)主线程只做一件事：accept → 创建线程。所有业务逻辑都在子线程中，主线程永远不会阻塞。
57.8 客户端程序
客户端需要同时做两件事：① 读取服务器消息并显示；② 读取用户输入并发送。用两个线程实现：
func startClient(string host, int port)
TcpStream conn = TcpStream(tcp.connect(host, uint16(port), 5000))
println("已连接到 " + host + ":" + port.toString())
// 接收线程：持续读取服务器消息并打印
thread.start(func()
while true
string msg = conn.recv(60000)
if msg.len() == 0 then
println("\n[服务器断开连接]")
os.exit(0)
end
println("\r" + msg) // \r 覆盖当前行再输出
end
end)
// 主线程：读取用户输入并发送
while true
string input = io.readLine()
if input.len() > 0 then
conn.send(input + "\n")
conn.drain(3000)
if string.startsWith(input, "QUIT") then
println("已退出")
break
end
end
end
conn.close()
end
// 启动客户端
startClient("127.0.0.1", 9000)收发分离是网络客户端的标准模式：读线程阻塞在 recv，主线程阻塞在 readLine，互不干扰。
57.9 命令系统
// 解析并执行命令，返回 false 表示退出
func processCommand(Hub hub, ClientInfo me, string line) as bool
// 分割命令和参数
int spaceIdx = string.indexOf(line, " ")
string cmd = line
string arg = ""
if spaceIdx >= 0 then
cmd = string.substring(line, 0, spaceIdx)
arg = string.substring(line, spaceIdx + 1)
end
if cmd == "NICK" then
// 设置昵称
if arg.len() == 0 then
sendTo(me, "ERROR 用法: NICK 名字")
else
string oldNick = me.nick
me.nick = arg
broadcast(hub, "JOIN " + oldNick + " 改名为 " + me.nick, me.id)
sendTo(me, "OK 昵称已设为 " + me.nick)
end
elif cmd == "MSG" then
// 群聊消息
if arg.len() == 0 then
sendTo(me, "ERROR 用法: MSG 内容")
else
broadcast(hub, "MSG " + me.nick + ": " + arg, me.id)
end
elif cmd == "WHO" then
// 在线列表
string list = whoList(hub)
sendTo(me, "WHO " + list)
elif cmd == "TELL" then
// 私聊: TELL nick message
int sp = string.indexOf(arg, " ")
if sp < 0 then
sendTo(me, "ERROR 用法: TELL 昵称 内容")
else
string targetNick = string.substring(arg, 0, sp)
string message = string.substring(arg, sp + 1)
ClientInfo target = findByNick(hub, targetNick)
if target == nil then
sendTo(me, "ERROR 用户 " + targetNick + " 不在线")
else
sendTo(target, "TELL " + me.nick + ": " + message)
sendTo(me, "OK 已发送给 " + targetNick)
end
end
elif cmd == "QUIT" then
return false // 触发退出
else
sendTo(me, "ERROR 未知命令: " + cmd)
end
return true // 继续运行
end命令解析用 indexOf + substring 分割——简单高效。每个命令对应一个分支。
57.10 完整实现
将以上所有组件组合为完整程序：
// ═══════════════════════════════════════
// chat_server.xl — 聊天室服务器
// ═══════════════════════════════════════
class ClientInfo
TcpStream conn
string nick
int id
func make(TcpStream conn, int id) as ClientInfo
ClientInfo c = ClientInfo()
c.conn = conn
c.nick = "guest" + id.toString()
c.id = id
return c
end
end
func sendTo(ClientInfo client, string msg)
client.conn.send(msg + "\n")
client.conn.drain(3000)
end
class Hub
list<ClientInfo> clients
Mutex lock
int nextId
func init()
clients = list<ClientInfo>()
lock = Mutex()
nextId = 1
end
func register(TcpStream conn) as ClientInfo
lock.lock()
ClientInfo c = ClientInfo.make(conn, nextId)
nextId += 1
clients.add(c)
lock.unlock()
return c
end
func unregister(ClientInfo client)
lock.lock()
int i = 0
while i < clients.len()
if clients[i].id == client.id then
clients.removeAt(i)
else
i += 1
end
end
lock.unlock()
end
func broadcast(string msg, int excludeId)
lock.lock()
int i = 0
while i < clients.len()
if clients[i].id != excludeId then
sendTo(clients[i], msg)
end
i += 1
end
lock.unlock()
end
func whoList() as string
lock.lock()
string result = ""
int i = 0
while i < clients.len()
if i > 0 then result += ", " end
result += clients[i].nick
i += 1
end
lock.unlock()
return result
end
func findByNick(string nick) as ClientInfo
lock.lock()
ClientInfo found = nil
int i = 0
while i < clients.len()
if clients[i].nick == nick then
found = clients[i]
end
i += 1
end
lock.unlock()
return found
end
end
func processCommand(Hub hub, ClientInfo me, string line) as bool
int spaceIdx = string.indexOf(line, " ")
string cmd = line
string arg = ""
if spaceIdx >= 0 then
cmd = string.substring(line, 0, spaceIdx)
arg = string.substring(line, spaceIdx + 1)
end
if cmd == "NICK" then
if arg.len() == 0 then
sendTo(me, "ERROR 用法: NICK 名字")
else
string oldNick = me.nick
me.nick = arg
hub.broadcast("JOIN " + oldNick + " 改名为 " + me.nick, me.id)
sendTo(me, "OK 昵称已设为 " + me.nick)
end
elif cmd == "MSG" then
if arg.len() > 0 then
hub.broadcast("MSG " + me.nick + ": " + arg, me.id)
end
elif cmd == "WHO" then
sendTo(me, "WHO " + hub.whoList())
elif cmd == "TELL" then
int sp = string.indexOf(arg, " ")
if sp < 0 then
sendTo(me, "ERROR 用法: TELL 昵称 内容")
else
string targetNick = string.substring(arg, 0, sp)
string message = string.substring(arg, sp + 1)
ClientInfo target = hub.findByNick(targetNick)
if target == nil then
sendTo(me, "ERROR 用户 " + targetNick + " 不在线")
else
sendTo(target, "TELL " + me.nick + ": " + message)
sendTo(me, "OK 已发送给 " + targetNick)
end
end
elif cmd == "QUIT" then
return false
else
sendTo(me, "ERROR 未知命令: " + cmd)
end
return true
end
func handleClient(Hub hub, TcpStream conn)
ClientInfo me = hub.register(conn)
sendTo(me, "WELCOME 欢迎来到聊天室! 临时昵称: " + me.nick)
sendTo(me, "命令: NICK 名字 | MSG 内容 | WHO | TELL 昵称 内容 | QUIT")
hub.broadcast("JOIN " + me.nick + " 加入了聊天室", me.id)
bool running = true
while running
string line = conn.recv(120000)
if line.len() == 0 then
running = false
else
line = string.trim(line)
if line.len() > 0 then
running = processCommand(hub, me, line)
end
end
end
hub.broadcast("LEAVE " + me.nick + " 离开了聊天室", me.id)
hub.unregister(me)
conn.close()
end
func main()
Hub hub = Hub()
hub.init()
TcpServer server = TcpServer(tcp.listen(9000))
println("═══ 聊天室服务器 ═══")
println("端口: 9000")
println("等待连接...")
while true
point accepted = server.accept(30000)
if accepted.valid() then
TcpStream conn = TcpStream(accepted)
println("[+] 新连接: " + conn.peerAddr())
thread.start(func()
handleClient(hub, conn)
end)
end
end
end
main()
测试方法
开 3 个终端：一个运行服务器，两个运行客户端。在客户端 A 输入 NICK alice，客户端 B 会收到 JOIN guest1 改名为 alice。输入 MSG hello 广播消息，WHO 查看在线列表。
深入细节
1. 为什么用每连接一线程而不是异步 I/O？
线程模型代码最直观——每个连接的逻辑是顺序的"读→处理→写"。异步模型（epoll/kqueue）性能更好但代码复杂度高。对于百级并发，线程完全够用。xlang 的 acceptAsync/recvAsync/sendAsync 提供了异步选项，但聊天室场景下线程更清晰。
2. 锁的粒度
本章用一把大锁保护整个 clients 列表。更细粒度的方案：读写锁（读多写少）、每客户端独立锁（广播时不阻塞注册）。但对于 10-100 个客户端，大锁的竞争概率极低。
3. 消息边界问题
TCP 是字节流——recv 可能返回半条消息或多条消息。xlang 的 recv 以 \n 为分隔符自动处理了这个问题（返回一行）。如果实现二进制协议，需要自己处理粘包（长度前缀 + 缓冲区拼接）。
4. 优雅关闭
生产环境需要：① 服务器收到 SIGINT 时停止 accept；② 通知所有客户端"服务器关闭"；③ 等待所有线程结束；④ 关闭监听 socket。本章简化为无限循环。
⚠️ 常见错误
错误 1：广播时不持锁
// ✗ 危险：遍历 clients 时其他线程可能修改列表
func broadcastWrong(Hub hub, string msg)
int i = 0
while i < hub.clients.len() // 无锁！
sendTo(hub.clients[i], msg)
i += 1
end
end
// ✓ 正确：持锁遍历
func broadcastRight(Hub hub, string msg)
hub.lock.lock()
int i = 0
while i < hub.clients.len()
sendTo(hub.clients[i], msg)
i += 1
end
hub.lock.unlock()
end错误 2：忘记 drain
// ✗ send 后不 drain，数据可能滞留在缓冲区
conn.send(msg + "\n")
// ✓ send + drain 确保数据发出
conn.send(msg + "\n")
conn.drain(3000)错误 3：客户端断开后继续发送
// ✗ 对已关闭的连接发送会失败（静默丢弃或报错）
// ✓ 在 handleClient 中，recv 返回空串即退出循环
// 退出后 unregister 从列表移除，不再被广播到错误 4：死锁——持锁时调用可能再次加锁的函数
// ✗ broadcast 持锁 → sendTo → 如果 sendTo 内部也加锁 → 死锁
// ✓ 本章的 sendTo 不加锁，安全
// 如果需要更复杂逻辑，先复制列表再释放锁，最后遍历副本发送本章小结
- 架构：TcpServer + Hub（注册表 + 广播）+ 每连接线程
- 协议：文本行协议，\n 分隔，命令前缀（NICK/MSG/WHO/TELL/QUIT）
- 并发：主线程 accept，子线程处理连接，Mutex 保护共享状态
- 广播：持锁遍历 clients 列表，排除发送者
- 生命周期：注册 → 消息循环 → 清理（通知 + 注销 + 关闭）
- 客户端：收发分离（读线程 + 主线程写）
复习题
- 为什么聊天室服务器用"每连接一线程"模型？它的优缺点是什么？
- Hub 的
lock保护什么？如果不加锁会发生什么？ - 文本行协议的消息边界靠什么保证？如果改用二进制协议，如何处理粘包？
- 客户端为什么需要两个线程？如果只用一个线程会怎样？
broadcast函数为什么在持锁状态下发送？有什么潜在风险？如何优化？
编程练习
练习 1：消息历史
在 Hub 中维护最近 50 条消息的列表。新客户端加入时自动发送历史记录。提示：用 list<string> 存储，超过 50 条时 removeAt(0)。
练习 2：频道系统
支持多频道：JOIN #general、JOIN #tech。消息只广播给同频道的用户。提示：ClientInfo 增加 channel 字段，broadcast 时过滤。
练习 3：心跳检测
客户端每 10 秒发送 PING，服务器回复 PONG。如果 30 秒没收到 PING，服务器主动断开。提示：ClientInfo 记录 lastPing 时间戳， Hub 增加定时检查线程。