简介这是一份面向C网络编程初学者与进阶学习者的实战项目资源聚焦多端口并发通信与客户端实时聊天系统开发。项目基于Boost.Asio等标准C网络库实现完整覆盖socket多端口监听、多线程/异步连接管理、消息广播中转、客户端-服务器双向通信等核心知识点适用于课程设计、毕业设计及网络编程能力强化训练。压缩包共8个文件含4个关键源码文件MyServer.cpp、MyClient.cpp等用于理解服务端监听逻辑与客户端通信协议以及4个可直接运行的exe程序Server.exe、Client.exe便于快速验证功能与调试交互流程整体体积仅113KB轻量易部署。目前已有386人学习下载读者可直接运行体验多端口并发聊天效果深入掌握TCP连接生命周期管理、跨客户端消息分发机制及基础错误处理实践。1. 多端口服务器多个客户端相互聊天不是“群聊”而是让每个客户端独占一个通信通道实现隔离式实时对话你手头有个multi-port-chat-server.zip解压后发现里面是 Python 脚本、配置文件和几个.pyc——但运行起来却卡在「连接拒绝」或「消息发不出去」。这不是一个简单的「多人在线聊天室」而是一个刻意用多个 TCP 端口隔离通信流的设计客户端 A 连 8001B 连 8002C 连 8003……彼此不共享 socket也不经由中心广播却又能「看到对方发的消息」。它解决的不是高并发压力而是调试场景下的通信可追溯性、协议兼容性验证、以及多设备异步交互时的信道干扰规避问题。比如你在测试 IoT 设备固件升级流程需要同时模拟 3 台不同型号终端分别连不同端口上报状态又或者你在做 SIP 信令调试每台软电话必须绑定独立端口避免 SDP 冲突。这类项目不追求百万在线但要求「每个连接可独立启停、日志可分端口归档、消息路由不交叉」。适合嵌入式工程师做设备联调、协议栈开发者验证握手逻辑、以及教学场景中让学生直观理解「端口即信道」的本质。它不是替代 WebSocket 或 MQTT 的方案而是当你需要「把网络层控制权抓在自己手里」时最轻量、最透明、最可控的落地选择。2. 为什么非得用多端口单服务多线程/协程不香吗2.1 单端口 vs 多端口本质是「连接上下文隔离粒度」的选择很多人第一反应是“用一个端口 select/epoll/asyncio不就能管成百上千连接了吗”——没错但那是应用层视角的复用。而多端口方案提供的是传输层原生隔离。关键区别在于连接标识唯一性单端口下client_ip:port → server_ip:8000是唯一五元组多端口下client_ip:port → server_ip:8001和client_ip:port → server_ip:8002是两个完全独立的 TCP 连接内核 socket 表里就是两条记录netstat -tnp | grep :800[123]能直接看到分离状态。防火墙/NAT 映射友好某些工业网关只允许白名单端口透传你无法在 8000 端口上动态分配子通道但可以提前开放 8001–8005 五个端口每个设备固定绑定一个。协议协商无干扰比如客户端 A 用 TLS 1.2 连 8001B 用明文 TCP 连 8002C 用自定义二进制协议连 8003 —— 单端口服务必须在accept()后靠读取前几个字节做协议嗅探易出错而多端口天然按端口分流socket.bind((0.0.0.0, 8001))就决定了这个 socket 只处理一种协议握手逻辑。提示这不是性能优化而是运维可控性设计。当某台客户端行为异常如疯狂重连、发送畸形包你kill -9 $(lsof -ti:8002)就能精准杀掉对应进程不影响其他端口服务。单端口方案则需在代码里写连接 ID 管理、心跳超时剔除、甚至引入 Redis 做连接状态同步——复杂度指数上升。2.2 选型依据Pythonsocketserver比asyncio更适合此场景虽然asyncio写高并发服务更现代但本项目核心诉求是「每个端口一个独立服务实例」而非「单进程扛万连」。socketserver.ThreadingTCPServer正好匹配每个端口启动一个ThreadingTCPServer实例主线程只负责监听worker 线程处理具体连接不用操心 event loop 共享、任务调度优先级、协程间变量竞争日志可按端口打到不同文件logging.FileHandler(flog_port_{port}.log)排查时tail -f log_port_8002.log直接聚焦问题代码结构清晰server_8001.py、server_8002.py、server_8003.py三份几乎相同的脚本新人改一个端口逻辑不会误动另一个。我们不用multiprocessing启多个进程资源开销大、IPC 复杂也不用docker run -p 8001:8001启多个容器过度工程。就用最朴素的「一个端口一个ThreadingTCPServer实例」配合主控脚本统一启停——这是实操中故障率最低、交接成本最小、debug 最快的做法。2.3 架构图不是星型而是「并列双通道」模型Client A (192.168.1.10:54321) ────→ [Server:8001] ←───┐ Client B (192.168.1.11:61234) ────→ [Server:8002] ←───┤→ 消息中继模块可选 Client C (192.168.1.12:55678) ────→ [Server:8003] ←───┘注意箭头方向客户端只向对应端口发消息但服务器内部有一个轻量级「跨端口消息中继器」非必须但项目 zip 里通常包含。它不转发原始 socket 数据而是解析应用层协议如MSG|from:A|to:B|text:hello再投递到目标端口的服务实例内存队列中。这样既保持端口隔离又实现「跨客户端可见」——这才是标题里「相互聊天」的真实含义A 发给 B 的消息B 在自己连接的 8002 端口上收到而不是 A 直连 B 的 IP。3. 用 threadingTCPServer 在本地跑通最小可运行版本3.1 核心服务类ChatHandler必须重写handle()而非process_request()很多初学者照搬socketserver文档直接继承BaseRequestHandler并改process_request()结果发现self.request在process_request()里根本没初始化。正确做法是重写handle()方法——它在连接建立后、数据可读时被调用此时self.requestsocket 对象和self.client_address已就绪import socketserver import threading import json import time class ChatHandler(socketserver.BaseRequestHandler): # 全局消息池{port: [msg_list]} message_pool {} lock threading.Lock() def setup(self): self.port self.server.server_address[1] if self.port not in self.message_pool: with self.lock: if self.port not in self.message_pool: self.message_pool[self.port] [] def handle(self): # 设置超时避免客户端断连后线程卡死 self.request.settimeout(30) client_ip self.client_address[0] print(f[Port {self.port}] Client {client_ip} connected) try: while True: try: data self.request.recv(1024).strip() if not data: break # 解析 JSON 消息{type:msg,from:A,to:B,text:hi} msg json.loads(data.decode(utf-8)) if msg.get(type) msg: # 存入本端口消息池供本端口其他客户端拉取 with self.lock: self.message_pool[self.port].append({ timestamp: time.time(), from: msg[from], to: msg[to], text: msg[text] }) # 若 to 是其他端口客户端则触发跨端口中继见 3.2 self.relay_to_other_port(msg) # 回复 ACK self.request.sendall(b{status:ok}) except socket.timeout: continue except json.JSONDecodeError as e: self.request.sendall(b{error:invalid_json}) break except ConnectionResetError: break finally: print(f[Port {self.port}] Client {client_ip} disconnected) def relay_to_other_port(self, msg): 将消息中继到目标端口简化版硬编码映射 target_port_map {A: 8002, B: 8003, C: 8001} # A发给B → 投递到8002端口 target_port target_port_map.get(msg.get(to)) if target_port and target_port ! self.port: # 实际应通过线程安全队列或 Redis 通知目标端口服务 # 此处仅打印示意 print(f[RELAY] {msg[from]}→{msg[to]} → Port {target_port}: {msg[text]})这段代码的关键点setup()中初始化self.port并确保message_pool按端口键存在避免多线程写冲突handle()内settimeout(30)防止客户端异常断开导致线程永久阻塞relay_to_other_port()是「相互聊天」的核心逻辑它不直接send()给目标客户端因为目标客户端连的是另一个端口socket 不同而是触发跨端口通知机制后续章节详述所有print()都应替换为logging.info()生产环境禁用 stdout。3.2 启动三个端口服务用threading.Thread统一管理生命周期不要写三个独立脚本然后手动python server_8001.py python server_8002.py ...。用一个主控脚本start_servers.py启动所有服务并支持 CtrlC 安全退出# start_servers.py import threading import time import signal import sys from socketserver import ThreadingTCPServer # 导入上面定义的 ChatHandler from chat_handler import ChatHandler servers [] ports [8001, 8002, 8003] def start_server(port): server ThreadingTCPServer((0.0.0.0, port), ChatHandler) print(fStarting server on port {port}...) server.serve_forever() def signal_handler(signum, frame): print(\nShutting down servers...) for server in servers: server.shutdown() server.server_close() sys.exit(0) if __name__ __main__: signal.signal(signal.SIGINT, signal_handler) # 启动每个端口服务为独立线程 for port in ports: t threading.Thread(targetstart_server, args(port,), daemonFalse) t.start() servers.append(t) time.sleep(0.1) # 避免端口争抢 print(All servers started. Press CtrlC to stop.) # 主线程不能退出否则 daemonTrue 的线程会直接终止 try: while True: time.sleep(1) except KeyboardInterrupt: signal_handler(None, None)参数说明daemonFalse确保线程不是守护线程server.shutdown()才能生效time.sleep(0.1)避免多个ThreadingTCPServer同时 bind 导致Address already in use尽管端口不同但内核调度可能瞬时冲突signal.signal(signal.SIGINT, ...)捕获 CtrlC逐个调用shutdown()关闭监听 socket比os._exit()安全得多。运行后你会看到Starting server on port 8001... Starting server on port 8002... Starting server on port 8003... All servers started. Press CtrlC to stop. [Port 8001] Client 127.0.0.1:54321 connected [Port 8002] Client 127.0.0.1:54322 connected [RELAY] A→B → Port 8002: hi there!这证明「多端口隔离 跨端口中继」已跑通。4. 跨端口消息中继的三种落地方式从内存队列到 Redis4.1 方案对比为什么不用全局 dict为什么不用 multiprocessing.Queue方案实现难度进程安全跨机器排查难度适用场景全局dictthreading.Lock★☆☆☆☆✅同进程❌★★☆☆☆日志分散单机调试、教学演示multiprocessing.Manager().dict()★★☆☆☆✅跨进程❌★★★☆☆Manager 进程易成瓶颈多进程部署、不跨机器Redis Pub/Sub★★★★☆✅✅✅✅★★★★☆redis-cli monitor实时看流生产环境、需横向扩展你解压的 zip 包里大概率是第一种全局 dict因为它最简单但也是线上翻车率最高的方案——一旦你把三个端口服务拆到不同机器全局 dict 就彻底失效。而multiprocessing.Queue看似合理但它要求所有ThreadingTCPServer实例必须由同一个multiprocessing.Process启动违背了「每个端口独立生命周期」的设计初衷。4.2 推荐方案Redis Pub/Sub零修改适配现有代码只需两处改动就能把内存中继升级为 Redis 中继且不破坏原有端口隔离性在ChatHandler.relay_to_other_port()中把print(...)替换为 Redis 发布import redis # 初始化一次全局或类属性 r redis.Redis(hostlocalhost, port6379, db0) def relay_to_other_port(self, msg): target_port_map {A: 8002, B: 8003, C: 8001} target_port target_port_map.get(msg.get(to)) if target_port and target_port ! self.port: # 发布到频道 port_8002 channel fport_{target_port} payload json.dumps({ from: msg[from], text: msg[text], timestamp: time.time() }).encode(utf-8) r.publish(channel, payload)在每个端口服务的ChatHandler.handle()循环末尾添加 Redis 订阅监听注意必须用redis-py的pubsub模块不能用blpopdef handle(self): # ... 原有 recv 逻辑 ... # 启动本端口订阅线程只启动一次 if not hasattr(self, _sub_thread_started): self._sub_thread_started True sub_thread threading.Thread( targetself._listen_redis_channel, args(fport_{self.port},), daemonTrue ) sub_thread.start() def _listen_redis_channel(self, channel): pubsub r.pubsub() pubsub.subscribe(channel) for message in pubsub.listen(): if message[type] message: try: data json.loads(message[data].decode(utf-8)) # 将消息写入本端口消息池供当前连接的客户端拉取 with self.lock: self.message_pool[self.port].append({ from: data[from], text: data[text], timestamp: data[timestamp], via: redis }) except Exception as e: print(fRedis parse error: {e})这样A 发给 B 的消息 → 8001 端口服务发布到port_8002→ 8002 端口服务的订阅线程收到 → 写入message_pool[8002]→ B 的客户端下次GET /messages就能拿到。所有端口服务仍独立运行只是消息流转经过 Redis 中转完美兼顾隔离性与互通性。注意Redis 必须提前安装sudo apt install redis-server并启动sudo systemctl start redis-server。若用 Dockerdocker run -d --name redis -p 6379:6379 redis:alpine即可。5. 避坑多端口聊天服务的 4 个血泪经验5.1 现象客户端连上后立即断开netstat显示TIME_WAIT爆满原因ThreadingTCPServer默认不启用SO_REUSEADDR每次重启服务旧连接的TIME_WAIT状态会占满端口新bind()失败。解决在ThreadingTCPServer初始化后显式设置 socket 选项server ThreadingTCPServer((0.0.0.0, port), ChatHandler) server.allow_reuse_address True # 关键等价于 setsockopt(SO_REUSEADDR)提示Linux 默认net.ipv4.tcp_fin_timeout 60TIME_WAIT状态持续 60 秒。开启reuse_address后即使有TIME_WAIT新bind()也能成功。5.2 现象A 发消息给 BB 收不到但日志显示RELAY成功原因ChatHandler的message_pool是按端口维护的但客户端拉取消息的逻辑如GET /messages接口可能没实现「清空已读消息」导致重复推送或堆积溢出。解决在handle()中接收消息后不要无限追加要限制长度并轮转MAX_MSGS_PER_PORT 100 with self.lock: self.message_pool[self.port] self.message_pool[self.port][-MAX_MSGS_PER_PORT:]5.3 现象用telnet 127.0.0.1 8001能连上但发 JSON 消息后服务端报JSONDecodeError原因telnet发送的是纯文本回车符是\r\n而 Pythonjson.loads()要求严格格式。telnet输入{type:msg,from:A,to:B,text:hi}实际发过去的是{type:msg,from:A,to:B,text:hi}\r\n\r\n导致 JSON 解析失败。解决在recv()后先strip()再解析data self.request.recv(1024).strip() # 去掉 \r\n if data: msg json.loads(data.decode(utf-8))血泪经验永远不要在telnet里测 JSON 协议用nc或写个 Python 客户端脚本见 6.1。5.4 现象三个端口服务都启动了但ps aux | grep python只看到一个进程原因ThreadingTCPServer.serve_forever()是阻塞调用如果你在start_servers.py里用for port in ports: server ...; server.serve_forever()第二个端口根本启动不了——程序卡死在第一个serve_forever()。解决必须用threading.Thread包裹每个serve_forever()且daemonFalse如 3.2 节所示。切记serve_forever()是「永不返回」的函数不能顺序调用。6. 客户端验证与生产就绪技巧用 Python 写个带心跳的 CLI 客户端6.1 为什么不用浏览器或 Postman 测因为 TCP 层协议需要长连接维持HTTP 是短连接每次GET /messages都要三次握手无法维持「在线状态」。而聊天本质是长连接客户端连上后服务端要能随时send()消息过来。所以必须用原生 socket 客户端。下面是一个带心跳、自动重连、支持多端口切换的 CLI 工具# client.py import socket import json import threading import time import sys class ChatClient: def __init__(self, host, port, client_id): self.host host self.port port self.client_id client_id self.sock None self.running False def connect(self): while not self.running: try: self.sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.connect((self.host, self.port)) self.sock.settimeout(30) print(fConnected to {self.host}:{self.port} as {self.client_id}) self.running True return True except ConnectionRefusedError: print(fConnection refused, retrying in 3s...) time.sleep(3) except Exception as e: print(fConnect error: {e}, retrying...) time.sleep(3) def send_msg(self, to, text): msg { type: msg, from: self.client_id, to: to, text: text } try: self.sock.sendall(json.dumps(msg).encode(utf-8) b\n) except Exception as e: print(fSend failed: {e}) self.running False def recv_loop(self): while self.running: try: data self.sock.recv(1024) if not data: break # 按行解析服务端每条消息后加 \n for line in data.split(b\n): if not line.strip(): continue try: resp json.loads(line.decode(utf-8)) if text in resp and from in resp: print(f[{resp[from]}]: {resp[text]}) elif status in resp: print(f← {resp[status]}) except json.JSONDecodeError: print(f← [raw] {line}) except socket.timeout: continue except ConnectionResetError: print(Server closed connection) self.running False break def start(self): if not self.connect(): return # 启动接收线程 recv_thread threading.Thread(targetself.recv_loop, daemonTrue) recv_thread.start() # 发送心跳每 25 秒发一次空消息防 NAT 超时 heartbeat_thread threading.Thread(targetself.heartbeat, daemonTrue) heartbeat_thread.start() # 主线程读取 stdin try: while self.running: cmd input(f{self.client_id} ).strip() if not cmd: continue if cmd.lower() in [quit, exit]: break # 支持格式to:B hello world if cmd.startswith(to:): parts cmd.split( , 1) if len(parts) 2: to parts[0][3:] # 去掉 to: self.send_msg(to, parts[1]) else: print(Usage: to:id message) except EOFError: pass finally: self.running False if self.sock: self.sock.close() def heartbeat(self): while self.running: try: self.sock.sendall(b{type:ping}\n) time.sleep(25) except: break if __name__ __main__: if len(sys.argv) ! 4: print(Usage: python client.py host port client_id) sys.exit(1) client ChatClient(sys.argv[1], int(sys.argv[2]), sys.argv[3]) client.start()使用方法# 开三个终端 $ python client.py 127.0.0.1 8001 A $ python client.py 127.0.0.1 8002 B $ python client.py 127.0.0.1 8003 C在 A 终端输入to:B hi from AB 终端立刻显示[A]: hi from A。这就是「多端口服务器多个客户端相互聊天」的完整闭环。6.2 生产就绪 checklist不是可选是必须项目检查方式不做的后果端口占用检测启动前lsof -i :8001或netstat -tuln | grep :8001多个服务实例冲突静默失败日志按端口分割logging.FileHandler(flogs/port_{port}.log)出问题时无法定位是哪个端口出错连接数限制ThreadingTCPServer默认无上限加max_children10参数客户端恶意连接耗尽系统线程消息体大小限制recv(1024)改为recv(8192)并校验len(data) 8192大消息截断导致 JSON 解析失败SSL/TLS 封装用ssl.wrap_socket()包装self.request明文传输密码、token 等敏感信息最后说个我踩过的坑曾经上线后发现客户端连上 2 小时就自动断开查了一天发现是云厂商 SLB 默认 900 秒空闲超时而我们的heartbeat()是 25 秒发一次但sendall()失败没重试导致心跳丢失。后来改成def heartbeat(self): while self.running: try: self.sock.sendall(b{type:ping}\n) except Exception as e: print(fHeartbeat send failed: {e}) self.running False break time.sleep(25)加了异常捕获断了立刻退出让主循环触发重连。这种细节文档里不会写但线上真会卡你三天。希望帮到你。本文还有配套的精品资源点击获取