多客户端TCP服务器与广播消息：从并发模型到工程实现-北京尧图网络

前段时间有个刚转行做服务端的同学问我一个特别经典的问题他要在一个端口上让几百个客户端同时连上来任何一个客户端发一句话其余客户端都要立刻收到。我说这不就是个聊天室嘛他说聊天室他懂但真自己写就卡住了——多客户端连接好说广播消息到底该怎么发要不要给每个客户端开一个线程发消息的时候锁不锁客户端掉线了怎么处理这个问题我问过不少人很多干了几年后端的朋友一讲到“服务器多客户端连接和广播消息”第一反应都是背几个函数名一旦落到真实场景里就会漏掉一堆细节。今天干脆把这套逻辑从头到尾捋一遍从TCP连接在内核里怎么排队到消息协议怎么设计再到一个能直接跑起来的最小服务器最后聊聊单机广播和服务器集群广播的区别。这一套吃透了不管是做IM、做设备接入还是做消息推送底层思路都一样。1. 服务器多客户端连接难的不是API是并发1.1 你面对的最小可复现场景长什么样先把这个问题的边界说清楚。我们假设要做的不是复杂的分布式IM就是一个最朴素的“TCP聊天室”一个服务端进程监听固定端口比如9000。多个客户端主动接入每个客户端一条独立的TCP长连接。任意客户端输入一行文本服务端把它转发给当前在线的其他所有客户端。客户端上下线时要给其他人发系统通知比如“张三加入了聊天室”。这个场景麻雀虽小五脏俱全。它覆盖了服务器编程里最核心的三件事连接接入、连接保活、消息扩散。你去看现在市面上各种实时通信系统从游戏服务器到直播弹幕本质都是这三件事的变体。很多人容易在这里犯一个概念错误以为“多客户端连接”等于开多个端口或者以为广播消息是往某个特殊地址上写数据。其实服务端只开一个端口所有客户端连的都是同一个端口。内核会根据客户端的IP和端口四元组区分出不同的TCP连接你的程序拿到的是一堆不同的socket。多客户端连接的这个“多”指的是同时服务多条TCP连接而不是开多个监听端口。1.2 accept背后TCP连接在内核里经历了什么要理解多客户端连接先得知道一条连接从客户端发起到服务端代码感知中间发生了什么。很多教学只讲bind、listen、accept三步但这三步中间的细节才是线上翻车的重灾区。服务端调用listen时内核会为这个监听socket维护两个队列半连接队列和全连接队列。客户端发来SYN内核先把连接放进半连接队列等三次握手完成后再移入全连接队列。accept做的不是“创建连接”而是“从全连接队列里取一个已经握好手的连接出来交给应用”。这个机制解释了三个经典问题。第一为什么listen后面要带一个数字比如listen(fd, 128)。这个数字是全连接队列长度的上限不是最大并发连接数。你写1024也并不意味着只能有1024个客户端在线它只影响握手完成后连接在内核队列里排队等待accept的积压能力。如果队列满了新连接会被内核丢弃或延迟响应表现就是客户端能ping通但TCP连接一直建立不起来。第二为什么明明客户端已经发起连接了你的服务端代码里却迟迟没有accept返回可能是因为全连接队列满了也可能是因为你根本没有调用accept。事件驱动模型里监听socket变成可读表示队列里有连接可取这时候才去accept不能把accept当作一个主动轮询连接的动作。第三生产环境里大量客户端同时接入时连接堆积会导致全连接队列溢出。排查方法很直接ss -lnt看监听端口的Send-Q列它显示的往往是队列上限配置值如果Recv-Q长期接近上限说明accept的速度跟不上连接建立的速度。此时要检查是不是在accept之后做了太多耗时操作或者进程的文件描述符上限太低。1.3 并发模型的三种主流玩法怎么选连接怎么接收搞清楚了下一个问题是几百上千个连接同时在那里收发数据程序怎么兼顾得过来历史上主流方案有三条路每条路的取舍完全不同。一是每连接一个线程。思路简单粗暴accept到一个连接就开一个线程去处理。对于刚入门的同学这是最直观的模型写起来也最省心。但问题在于线程是昂贵资源。一个线程默认栈空间可能就要8MB的虚拟内存几千个线程跑起来光是内存就很可观。而且线程切换要交内核调度当线程数远大于CPU核数时大部分时间都耗在上下文切换上。这个模型适合并发量不大、业务逻辑简单的场景比如公司内部几十个设备做监控上报。二是事件驱动模型。把socket设置为非阻塞然后用select、poll、epoll这类系统调用统一等待事件。Linux下的epoll是目前主流高性能服务器的基石Nginx、Netty底层都在用它。它的核心优势是不管你有多少连接内核只告诉你“哪些连接有事件发生”你不需要一个一个去遍历问“你有没有数据”。这个模型写起来比多线程绕因为你的代码必须保持非阻塞、不能一遇到阻塞操作就卡住整个线程但它能支撑的连接数和CPU利用率都远好于每连接一个线程。三是协程模型。最典型的是Go语言的goroutine配合netpoll。它让你写出了“看起来像每连接一个线程”的同步代码底层却由运行时帮你做事件调度。这是目前新项目里很舒服的方案编码心智负担低又能兼顾高并发。适合对团队上手成本敏感、又要规模化的业务。选型不用迷信某个语言或框架。如果你只是临时搭一个给几百个设备用的服务事件驱动或者协程都能轻松胜任。真正要避免的是两种极端用“每连接一个线程”硬扛几万连接或者为了追求技术上的极致性能把简单需求做成一个复杂的Netty项目。先把连接管理模型理解透再根据业务量选择即可。2. 广播消息的完整套路连接池、消息边界、心跳与掉线2.1 业务层广播和网络层广播不是一回事聊到广播消息很多懂一点网络的同学会先想到UDP广播。UDP允许你往255.255.255.255这种广播地址发包交换机收到后会把这个包复制到同一个二层网络里的所有设备上。如果滥用会造成网络广播风暴整个局域网的设备都会受影响。路由器不会转发这种广播包所以在公网上不会遇到全局广播但公司局域网里乱发UDP广播是会被运维骂的。而我们做聊天室或者消息推送时说的广播是应用层的广播。TCP是点对点协议没有“一次发送就送达所有人”的原语。所谓广播就是服务器自己把同样的消息复制多份然后往每个连接的TCP通道里各发一遍。这个区别非常重要。你在设计消息接口的时候面向的“广播”是自己的业务逻辑比如“把这行文本推给当前在线的所有人”而不应该去套网络层的广播概念。不然你会发现UDP广播既不保证可靠又会殃及无辜设备而且根本没有“哪些人在线”的概念完全不适合做可靠的多客户端推送。2.2 服务端维护的在线列表里到底该放什么既然广播要靠服务器一个个发那就必须在内存里有一个“当前有哪些连接”的账簿。这个账簿通常是一个映射表key是socket对象或者连接IDvalue是那条连接的状态信息。至少应该包含这些字段socket或fd真正用来收发数据的句柄对端地址记录IP和端口排查问题、打日志的时候会用到用户身份信息比如user_id、昵称、设备号业务逻辑要靠它识别是谁最近活跃时间做心跳超时判断的基准收发缓冲TCP流式协议需要缓冲半包数据这会在后面讲到。保存的数据结构要支持O(1)的增删查。Python里通常就是字典Java可以用ConcurrentHashMapGo语言用一个带锁的map。广播消息时需要遍历当前所有在线连接为了不因为遍历过程中有连接回调删除自身而报错遍历前最好对键做一次快照或者把要删除的连接先放进列表等遍历完再统一清理。在线列表的更新时机也要小心。新连接accept成功就加入但此时用户可能还没“登录”很多实现会选择先放进一个未认证区等客户端发来身份标识后再正式加入在线列表并全网广播上线通知。如果你accept就广播“某某上线了”等用户身份还没识别完广播消息里可能连名字都拿不到。更稳妥的做法是加入连接池时只初始化基础数据真正广播给其他人是在第一条认证消息解析成功之后。2.3 一条消息到另一台机器上为什么会乱成一团这里必须讲TCP流式传输的坑。TCP没有“消息”的概念它只负责可靠地传输字节流。你调用一次send对方不一定一次recv就能拿到完整数据你连续调用两次send对方也可能把两次的数据合在一次recv里收到。前者叫半包后者叫粘包。举个例子。客户端A先发“你”再发“好”服务端可能先收到“你”也可能一次收到“你好”。如果没有任何边界规则服务端根本不知道该把这两段数据当作两句话还是一句话。解决这个问题是老生常谈的三板斧固定长度每条消息都占固定字节数不足就补齐按长度切分即可。缺点是很浪费业务消息长短不一很难受。分隔符每条消息末尾加一个特殊字符比如换行符\n收到后从缓冲区里按分隔符切出完整消息。HTTP早期从文本协议时代就常用这个方案。缺点是消息内容里如果有分隔符会冲突需要做转义。长度头每条消息前面加4个字节表示后面payload的长度服务端先读4字节再按长度读完整消息体。JSON、Protobuf这类二进制协议常用这种方案是目前生产环境最推荐的方式。看完后面的代码你会发现我特意在demo里用了“一行JSON加一个换行符”的形式不是为了偷懒而是为了让半包和粘包的处理逻辑清楚可见。真实生产环境建议格式改成4字节大端长度一个JSON字符串语义更清晰也不用担心JSON里出现换行符导致解析错乱。2.4 广播发送时最容易踩的暗坑慢客户端与心跳广播这个操作看着简单遍历连接逐个send。可一旦客户端网络状况不好问题立刻出现。TCP有发送缓冲区和接收缓冲区还有一套流量控制机制。当某个客户端接收很慢时它的接收缓冲区会满这时候TCP会回传一个窗口为0的包服务器的发送缓冲区会越积越多。如果你用的是阻塞socket一次send可能直接卡住导致整个广播循环停下来。如果是每连接一个线程的模型这个慢客户端只是卡住自己的线程影响还可控但在事件驱动模型里一个阻塞的socket就能让整个进程的后续事件都无法处理这就是“一颗老鼠屎拖垮整个服务器”。生产级做法是给每条连接维护一个发送队列和可写状态。注册读事件的同时平时不监听可写事件只有当这条连接的发送队列非空时才去监听EVENT_WRITE等socket可写时再冲刷缓冲。如果发送队列积压超过一个阈值比如几MB说明这个客户端已经明显跟不上节奏了这时候主动断开连接可能比带病坚持更合理。再聊心跳。TCP本身没有实时通知对端“我挂了”的机制。如果客户端是正常调用close退出服务端能收到FIN包表现为recv返回0很好处理。但如果客户端断网、机器断电、或者程序直接崩溃对端往往连FIN都发不出来服务端那边看这条连接依然是ESTABLISHED状态。如果代码里是阻塞等待读那这个线程或连接就永远傻等下去。解决方式通常是应用层心跳加超时检测。客户端每隔一段时间发一个心跳包服务端记录每个连接的最后活跃时间然后定期扫描在线列表把超过N秒没有活跃的连接踢掉。有些底层的SO_KEEPALIVE开关也能做探测但默认探测周期太长而且不能携带业务信息一般建议业务自己做心跳。心跳包和正常数据包在消息协议里要区分开。最简单的做法是在JSON里加一个type字段比如{type:ping}。服务端收到ping时可以回一个pong也可以不回只要更新活跃时间即可。注意不要每收到一个包都把整个在线列表反过来广播一遍这样很容易造成心跳风暴。3. 手把手实操写一个能跑的多客户端广播服务器3.1 为什么选Python的selectors做demo讲了一堆理论你现在最需要的是一个能在五分钟内跑起来、能看到广播效果的代码。我这里选Python3配合标准库selectors一个第三方依赖都不需要。选择selectors而不是直接写select或者epoll的原因很实在selectors对不同平台做了封装。你在自己的Mac或者Windows上跑它底层可能用的是select部署到Linux云服务器上它会自动用上epoll。这正好呼应前面讲的“事件驱动模型”代码里只处理事件回调不用关心底层具体是哪个多路复用器适合用来理解概念。在代码里我们用一个全局字典clients维护在线连接。每条连接的信息包括名字、socket、对端地址、接收缓冲区和最后活跃时间。监听socket收到可读事件时说明有客户端来连接执行accept普通socket收到可读事件说明有数据到达读取数据并处理。广播就是遍历clients字典把消息发给每个非发送者。框架非常短但每个环节都能砸中前面的知识点。3.2 服务端完整代码与逐步讲解先放完整代码我建议你直接存成server.py来跑。import json import selectors import socket import time HOST 0.0.0.0 PORT 9000 HEARTBEAT_TIMEOUT 90 sel selectors.DefaultSelector() clients {} # sock - 连接信息 def now_text(): return time.strftime([%H:%M:%S], time.localtime()) def broadcast_message(sender, msg): 把 msg 广播给所有在线连接sender 为 None 时广播给全部人。 line (json.dumps(msg, ensure_asciiFalse) \n).encode(utf-8) dead [] for sock in list(clients.keys()): if sock is sender: continue try: sock.sendall(line) except Exception: # 发送失败说明该连接已不可用收集起来统一关闭 dead.append(sock) for sock in dead: handle_close(sock) def handle_close(sock): try: sel.unregister(sock) except Exception: pass try: sock.close() except Exception: pass info clients.pop(sock, None) if info: name info.get(name) or info[addr] print(f{now_text()} [下线] {name}) broadcast_message(None, {type: system, text: f{name} 离开了当前在线 {len(clients)} 人}) def handle_line(sock, line): 处理一行完整 JSON 消息。 info clients.get(sock) if not info: return if info[name] is None: # 第一条消息是注册消息只有拿到名字后才广播上线 payload json.loads(line.decode(utf-8, ignore)) info[name] payload.get(name, f匿名-{sock.fileno()}) print(f{now_text()} [上线] {info[name]} - {info[addr]}) broadcast_message(None, {type: system, text: f{info[name]} 加入了聊天室当前在线 {len(clients)} 人}) return payload_str line.decode(utf-8, ignore).strip() print(f{now_text()} [{info[name]}] {payload_str}) broadcast_message(sock, {type: chat, name: info[name], content: payload_str}) def handle_read(sock): 一个客户端连接上有数据到达。 info clients.get(sock) if not info: handle_close(sock) return try: data sock.recv(4096) except Exception: data b if not data: # 对端正常关闭或异常断开 handle_close(sock) return info[buffer] data info[last_seen] time.time() # 按换行符切出完整消息余下部分继续留在缓冲区 while True: idx info[buffer].find(b\n) if idx -1: break line info[buffer][:idx] info[buffer] info[buffer][idx 1:].strip() if line: try: handle_line(sock, line) except Exception as e: print(f{now_text()} 消息处理失败: {e}) handle_close(sock) break def handle_accept(server): conn, addr server.accept() conn.setblocking(False) sel.register(conn, selectors.EVENT_READ, handle_read) clients[conn] { name: None, addr: addr, buffer: b, last_seen: time.time(), } print(f{now_text()} [连接] {addr} 进入当前连接数 {len(clients)}) def main(): server socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind((HOST, PORT)) server.listen(128) server.setblocking(False) sel.register(server, selectors.EVENT_READ, handle_accept) print(f聊天服务器已启动监听 {HOST}:{PORT}) while True: events sel.select(timeout1) for key, _ in events: callback key.data callback(key.fileobj) # 定时清理心跳超时的连接 now time.time() for sock in list(clients.keys()): if now - clients[sock][last_seen] HEARTBEAT_TIMEOUT: print(f{now_text()} [心跳超时] {clients[sock][addr]}) handle_close(sock) if __name__ __main__: main()代码核心逻辑其实集中在四个回调里。handle_accept处理新连接handle_read处理已有连接的数据到达handle_close负责统一清理broadcast_message负责把消息扩散给在线集合。心跳检查放在主循环里每select一次就顺手扫一遍这在在线数几千以内是完全够用的。有一点你需要特别注意handle_read里我用了info[buffer]累积数据每次收到数据就按换行符去切。这样即使一个完整的JSON被拆成多个TCP段到达代码也能等到换行符出现后再处理正好解决了上一节说的半包问题。如果一次收到多条消息拼在一起这个循环也会把它们逐条切出来粘包问题也顺带解决了。broadcast_message里用sender参数来决定是否跳过某条连接。当参数是None时系统消息发给所有人当参数是一个具体的socket时这条消息不会回发给说话的人自己。如果你做的是IM单聊可能希望有“自己聊天记录保存在本地”的回执那就需要客户端自己做去重而不是让服务端把自己也发一份。3.3 客户端测试工具与完整验证步骤服务端写完了怎么验证很多教程会叫你去用nc 127.0.0.1 9000连接。但注意我们的服务端约定每条消息是“一行JSON加一个换行符”用nc手工敲JSON太痛苦而且一次只开一个nc窗口看不出广播效果。我建议你写个几十行的Python客户端。存成client.pyimport json import socket import threading HOST 127.0.0.1 PORT 9000 def recv_loop(sock): buf b while True: try: chunk sock.recv(4096) except Exception: print(\n连接已断开) break if not chunk: print(\n连接已断开) break buf chunk while b\n in buf: line, buf buf.split(b\n, 1) if not line.strip(): continue msg json.loads(line.decode(utf-8, ignore)) if msg.get(type) system: print(f【系统】{msg.get(text)}) else: print(f\n{msg.get(name)}: {msg.get(content)}) print( , end, flushTrue) def main(): name input(你的昵称).strip() or 路人甲 s socket.create_connection((HOST, PORT), timeout30) threading.Thread(targetrecv_loop, args(s,), daemonTrue).start() # 先发送注册消息 s.sendall((json.dumps({name: name}) \n).encode(utf-8)) while True: text input( ) if not text: continue if text /quit: break s.sendall((json.dumps({content: text}) \n).encode(utf-8)) s.close() if __name__ __main__: main()验证流程是这样的在Linux服务器上或者本地启动服务端python3 server.py。打开两个以上终端分别运行python3 client.py取不同的昵称比如Alice和Bob。再开一个终端运行第三个客户端叫Carol。在Alice的窗口输入一句话比如“大家好”你会看到Bob和Carol的终端都打印出了Alice: 大家好而Alice自己的窗口不会重复显示这条内容。关掉Carol的客户端Alice和Bob的终端会收到一条系统消息“Carol 离开了当前在线 2 人”。如果这套流程跑通说明连接接入、消息广播、上下线通知这三个核心流程都工作了。这时候你可以在服务器上看连接状态ss -tn state established ( sport :9000 )这条命令能列出当前9000端口上的所有TCP连接每行对应一个客户端。3.4 从demo走向生产前必须补的几个细节这个demo能演示原理但直接拿上生产会出事。第一个要补的是消息体长度限制。现在如果有人故意发一个几百MB的字符串且从不带换行符服务端就会一直往buffer里灌数据内存迟早被撑爆。生产代码里每次往缓冲区追加数据后都要判断长度超过上限直接断开连接并记录日志。第二个要补的是连接数据与业务数据的分离。demo把“收到消息就广播”写在一个回调里真实业务肯定要经过鉴权、反垃圾、内容审核、存储这些环节。简单做法是先把消息解出来放进队列再由后面的工作线程去消费不要在一段回调里把所有业务都干完。第三个要补的是发送队列。我在demo里直接用了非阻塞socket加sock.sendall。消息量小、客户端数量不超过几百的时候这个方案基本不出事。但一旦某个客户端网速极差TCP发送缓冲区被填满这里的sendall就可能抛出BlockingIOError而我的代码逻辑会把这个连接当成异常连接踢掉。更健壮的实现是给每条连接加一个发送缓冲队列广播时只把消息压进队列并注册写事件等到socket可写时真正发送。4. 多客户端服务器经常翻车的几个点4.1 高频问题速查表很多问题其实都是同一个套路。我按经验整理了一个速查表基本覆盖多客户端连接和广播消息的高频雷区。现象可能原因处理方式服务端启动时报Address already in use上一个进程还没退出或端口被占用换端口或用SO_REUSEADDR确认进程占用用lsof -i:9000客户端连接超时服务端没任何日志云服务器安全组/防火墙没放行端口检查安全组入方向规则和本机防火墙不能只改代码连接建立了但消息总是收不全半包导致消息边界错乱按换行或长度头切分把未完整数据留在缓冲区多条消息黏在一起显示粘包问题没有消息边界解析同上服务端一广播就卡死某个客户端阻塞或发送缓冲爆满非阻塞socket、发送队列、踢出慢客户端客户端拔网线后连接一直不被清除TCP没收到FIN服务端不知道连接已死应用层心跳定期清理广播消息有人收到有人收不到遍历发送时有连接抛异常没处理send统一捕获异常并关闭异常连接进程运行几天后无法accept新连接文件描述符耗尽ulimit -n调大检查socket是否正确close服务端日志正常但客户端显示乱码编解码不一致消息统一UTF-8JSON里指定ensure_asciiFalse4.2 三个印象深刻的真实排查记录第一个坑Connection reset by peer。曾经有个服务在Linux跑着跑着整个进程没了日志末尾只有一行“Connection reset by peer”。查下来的原因是客户端崩溃后服务端继续向那条连接写数据触发了TCP的RST在C语言层面这个信号会终止进程。Python和Java的运行时屏蔽了这层信号所以那边的同学不太能遇到但用C/C实现服务器时一定要在代码里忽略SIGPIPE信号或者在send时加上MSG_NOSIGNAL标志否则一个客户端异常断开能让整个服务退出。第二个坑弱网客户端导致的“假广播风暴”。当时某个IM系统上线后群里反映消息有时候延迟好几秒。顺着日志查发现某条连接的客户端在弱网环境下接收很慢

多客户端TCP服务器与广播消息：从并发模型到工程实现

相关推荐

嵌入式开发工具选型：好用与专业之别，目标决定选择

嵌入式Linux进程管理实战：从fork到进程池的核心机制

Claude Code与MiniMax H3：本地部署与组合工作流指南

更多资讯

【python基础】python基础习题练习(一)

PP-LCNet_x0_25_textline_ori_onnx未来展望：轻量级AI模型的发展趋势

5个关键步骤实现WanVideo多模态视频生成：ComfyUI本地部署深度实战指南

Han1mePlus多平台构建指南：Android、iOS、macOS、Windows、Linux打包全流程

PP-LCNet_x0_25_textline_ori_onnx：轻量级文本方向检测模型的终极优势对比

xray与fofa联动（附脚本）

今日更新

淘宝商品详情API开发实战与性能优化指南

Unity Package Manager刷新错误排查与解决方案

Java record 规范构造器能否被反序列化或反射绕过

本周精选

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

本月热门

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

读完文章，想聊聊您的网站？