TCP网络编程实战：从协议设计到聊天室并发实现

简介：电子科技大学通信与信息工程学院网络软件设计项目是一套面向计算机相关专业学生的课程设计与毕业设计参考资源，覆盖需求分析、系统设计、编码实现与测试等完整流程，能够帮助学习者将理论知识与实际开发相结合。压缩包共50个文件，大小约4.99MB，包含C#工程源代码、XAML界面定义、Word/PPT/Markdown文档，以及配置文件、工程设置文件、版本忽略文件等；源码和文档按目录分开存放，便于阅读与复用。已有85人学习/下载。除项目源码外，还提供软件设计方案、测试计划、测试报告、编调记录、教学文档模板等过程性材料，以及说明文件、开源许可和辅助脚本。整套材料从设计到测试形成完整学习闭环，既适合初学者作为入门参考，也能支撑中高级学习者深入分析网络通信、数据结构与数据库集成等扩展问题。

1. 网络软件设计这门课到底在做什么：先分清课程目标与工程目标

第一次打开“电子科技大学通信与信息工程学院网络软件设计项目”这个课题文档时，很多人以为这是一门网页开发课，结果第一周就在 TCP 的粘包问题上翻车。网络软件设计的课程目标不是写一个能点的页面，而是让你用 socket 把一套基于 TCP 的小型应用从零搭起来：自己设计报文、自己处理并发、自己维护长连接。它能解决的问题很具体——把教材上的三次握手、流量控制、TIME_WAIT 变成代码里的日志和行为；适合大三、大四通信与计算机方向的学生，也适合准备后端面试的人补足网络功底。下面这套方案不是抄作业，而是把“做一个 TCP 聊天室”这条主线完整跑通，中间踩过的坑我一并标出来，照着做就行。

2. 设计先行：TCP 并发模型、自定义协议与心跳参数怎么定

2.1 为什么选 TCP + 多线程：课程验收真正想看到的考点

技术选型是这个项目第一个决策点。常见路线有两条：TCP 还是 UDP，多线程还是 select/epoll。我基本不会纠结，直接选 TCP 加每连接一个线程。

TCP 的理由很直接：课程验收重点在“可靠传输、长连接、状态维护”这三件事，UDP 只能练到收发报文，把可靠性、粘包、半包、连接状态这些核心概念全部躲过去了。你要是选 UDP 做一个“聊天室”，服务器只需要 recvfrom 加广播，代码量少一半，但答辩时基本聊不到深度问题上，分数上限被锁死。TCP 天然有序、可靠、面向连接，报文边界要靠自己设计，这才是网络软件设计真正要训练的部分。

并发模型上，select 能讲清楚事件驱动的思想，但把状态机和业务塞在一起写，新手很容易在十几个 fd 的 set 里绕晕。epoll 更贴近生产环境，可对一次课程设计来说，代码复杂度会掩盖协议设计的亮点。多线程是最直白的实现方式：主线程 accept，每个连接一个线程，客户端表用锁保护。连接数在 500 以内时，线程开销完全可接受，课程 demo 的规模也到不了性能瓶颈。你可以在答辩时口述对比一下 epoll 和线程模型的差异，这本身就是加分项。

2.2 自定义应用层协议：魔数、版本、类型、长度的四段式

选好传输层，接下来是设计应用层协议。这一步是通信学院课程设计和普通后端开发最大的分水岭：你不是直接套 HTTP，而是自己定义报文格式。

协议报文分两部分：固定头部 6 字节，加可变长度的载荷区。

类型码的具体含义：0x01 认证请求，0x02 认证响应，0x03 心跳，0x04 消息广播，0x05 退出。长度字段上限 65535 字节，对聊天业务绰绰有余；如果以后要传文件，单个报文超过这个值就要自己分片，这是第二版协议的事。

这一层如果不想每次重写，就单独放一个 protocol.py：

# protocol.py：网络软件设计中约定的报文格式 import struct MAGIC = 0xABCD HEADER_LEN = 6 TYPE_AUTH = 0x01 TYPE_AUTH_ACK = 0x02 TYPE_HEARTBEAT = 0x03 TYPE_MSG = 0x04 TYPE_QUIT = 0x05 def pack(type_: int, payload: bytes) -> bytes: # !H 表示网络字节序的无符号短整型，HBBH 正好 6 字节 return struct.pack("!HBBH", MAGIC, 1, type_, len(payload)) + payload def unpack(buf: bytes): if len(buf) < HEADER_LEN: return None, buf magic, ver, type_, length = struct.unpack("!HBBH", buf[:HEADER_LEN]) if magic != MAGIC: # 魔数不对，丢弃一个字节，重新寻找报文起点 return None, buf[1:] if len(buf) < HEADER_LEN + length: # 载荷未到齐，留给下一次 recv 数据拼接后再解析 return None, buf payload = buf[HEADER_LEN:HEADER_LEN + length] return (type_, payload), buf[HEADER_LEN + length:]

协议头设计有两个关键点。第一，魔数必须用两字节的固定值而不是一字节，一字节 256 种可能，随机字节流里出现相同值的概率更高，误同步概率大；两字节 0xABCD 基本能保证在乱序数据里找回同步。第二，长度字段必须用struct.pack("!H")转成网络字节序，本地是小端字节序的机器上直接打包，双方对不上长度，接收方会把整个报文当成垃圾数据处理。后面 3.1 的解析循环里，这两个细节直接决定服务端稳不稳。

2.3 心跳机制参数：间隔与超时阈值为什么按 1:3 配

长连接服务端必须处理一个问题：客户端掉线了，服务器怎么知道。TCP 有自带的 keepalive，但系统默认参数是两小时起步，对课程 demo 来说等于没有。自己在业务层做心跳是通用做法：客户端每隔一段时间发一个 0x03 空包，服务端记录最近一次心跳时间，超过阈值没收到就直接清理连接。

为什么是 3 倍而不是 1 倍或 10 倍？1 倍太敏感，网络抖动超过几秒就把正常用户踢下线；10 倍太迟钝，客户端拔线后服务器要继续保持几分钟的僵尸连接，占着 fd 和内存。3 倍余量能抗住大部分局域网和移动网络的抖动。注意一个细节：心跳只是兜底，任何业务消息到达时都应该同步刷新 last_heartbeat，不要只认心跳包。用户连续发消息，心跳断了，但业务还在，不该判定离线。

实现层还有一个选择：心跳检查用一个独立线程，每 30 秒遍历一次客户端表，比在 recv 里等数据更省事。核心逻辑在 3.1 的服务端代码里，先跑通主流程再回头看这个线程也不迟。

3. 把代码跑起来：服务端、客户端与心跳的最小实现

3.1 服务端主流程：bind、listen、accept 与连接线程

服务端可以分四个部分理解：socket 初始化、客户端连接管理、报文解析循环、心跳监控线程。下面是完整可运行的 server.py，去掉了多余的打印，保留了最核心的分支：

# server.py import socket import threading import struct import json import time MAGIC = 0xABCD HEADER_LEN = 6 HOST, PORT = "0.0.0.0", 8888 HEARTBEAT_INTERVAL = 30 HEARTBEAT_TIMEOUT = 90 clients = {} # conn -> {"name": str, "last_heartbeat": float} clients_lock = threading.Lock() server_running = True def build_packet(type_: int, payload: bytes) -> bytes: return struct.pack("!HBBH", MAGIC, 1, type_, len(payload)) + payload def parse_packets(buf: bytes): packets = [] while len(buf) >= HEADER_LEN: magic, ver, type_, length = struct.unpack("!HBBH", buf[:HEADER_LEN]) if magic != MAGIC: buf = buf[1:] continue if len(buf) < HEADER_LEN + length: break payload = json.loads(buf[HEADER_LEN:HEADER_LEN + length]) packets.append((type_, payload)) buf = buf[HEADER_LEN + length:] return packets, buf def broadcast(sender, message): with clients_lock: targets = [(c, info["name"]) for c, info in clients.items() if c is not sender] for conn, _ in targets: try: conn.sendall(build_packet(0x04, json.dumps(message).encode())) except OSError: remove_client(conn) def remove_client(conn): with clients_lock: cls = clients.pop(conn, None) if cls: broadcast(None, {"from": "system", "text": f"{cls['name']} 已退出"}) try: conn.close() except OSError: pass def handle_conn(conn, addr): print(f"{addr} 已接入") buf = b"" conn.settimeout(10) with clients_lock: clients[conn] = {"name": f"user-{addr[1]}", "last_heartbeat": time.time()} try: while True: try: data = conn.recv(4096) except socket.timeout: data = b"" except OSError: break if not data: break buf += data packets, buf = parse_packets(buf) for type_, payload in packets: if type_ == 0x01: name = payload.get("name", "anon") with clients_lock: clients[conn]["name"] = name conn.sendall(build_packet(0x02, json.dumps({"status": "ok"}).encode())) elif type_ == 0x03: with clients_lock: clients[conn]["last_heartbeat"] = time.time() elif type_ == 0x04: broadcast(conn, {"from": clients[conn]["name"], "text": payload["text"]}) elif type_ == 0x05: return finally: remove_client(conn) def heartbeat_monitor(): while server_running: time.sleep(HEARTBEAT_INTERVAL) now = time.time() with clients_lock: expired = [c for c, info in clients.items() if now - info["last_heartbeat"] > HEARTBEAT_TIMEOUT] for conn in expired: print("心跳超时，断开连接") remove_client(conn) 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) threading.Thread(target=heartbeat_monitor, daemon=True).start() while True: conn, addr = server.accept() threading.Thread(target=handle_conn, args=(conn, addr), daemon=True).start() if __name__ == "__main__": main()

几个参数需要解释清楚。listen(128)是连接等待队列长度，128 对课程实验甚至小规模压测都够，设太大反而掩盖了系统 fd 上限的问题。recv(4096)不是一次收 4096 字节就代表一个完整报文，TCP 是字节流，一次 recv 可能拿到半个包、也可能拿到三个包，缓冲区是必须的，4096 只是单次系统调用能搬回用户态的最大值。settimeout(10)让 recv 每 10 秒被唤醒一次，即使一直没有数据，线程也有机会检查连接是否被其他逻辑清理。

broadcast里采用的“锁内取快照，锁外发送”是这里最容易被忽略的设计：如果把sendall放进锁里面，某个客户端发送阻塞时，整个客户端表会被这个锁卡死，其他线程全部堵在with clients_lock上，这就是 4.4 节要展开的并发坑。

3.2 客户端结构：发送线程、接收线程与心跳线程

客户端比服务端难写的地方在输入阻塞。input()是阻塞的，如果你在主线程里输入文字，就没法及时接收别人的消息；如果收消息写在主线程，input()又会被消息刷掉。解决方案是把网络接收和心跳放进后台线程，主线程只负责读键盘。

# client.py import socket import struct import json import threading import time MAGIC = 0xABCD HEADER_LEN = 6 HOST, PORT = "127.0.0.1", 8888 HEARTBEAT_INTERVAL = 30 def build_packet(type_: int, payload: bytes) -> bytes: return struct.pack("!HBBH", MAGIC, 1, type_, len(payload)) + payload def send_json(sock, type_: int, obj): sock.sendall(build_packet(type_, json.dumps(obj).encode())) def recv_loop(sock): buf = b"" while True: try: data = sock.recv(4096) except (OSError, ConnectionError): print("连接已断开") return if not data: return buf += data while len(buf) >= HEADER_LEN: magic, ver, type_, length = struct.unpack("!HBBH", buf[:HEADER_LEN]) if magic != MAGIC: buf = buf[1:] continue if len(buf) < HEADER_LEN + length: break payload = json.loads(buf[HEADER_LEN:HEADER_LEN + length]) buf = buf[HEADER_LEN + length:] if type_ == 0x02: print("[服务器] 认证成功") elif type_ == 0x04: print(f"[{payload['from']}] {payload['text']}") def heartbeat_loop(sock, stop_event): while not stop_event.is_set(): time.sleep(HEARTBEAT_INTERVAL) try: send_json(sock, 0x03, {}) except OSError: stop_event.set() break def main(): with socket.create_connection((HOST, PORT), timeout=5) as sock: stop_event = threading.Event() name = input("输入昵称: ").strip() send_json(sock, 0x01, {"name": name}) threading.Thread(target=recv_loop, args=(sock,), daemon=True).start() threading.Thread(target=heartbeat_loop, args=(sock, stop_event), daemon=True).start() while True: text = input() if text == "/quit": send_json(sock, 0x05, {}) break send_json(sock, 0x04, {"text": text}) stop_event.set() if __name__ == "__main__": main()

客户端的三个线程分工要分明：主线程input()负责业务发送，recv_loop负责接收，heartbeat_loop负责保持活性。create_connection与socket.connect的区别是它内部已经处理了 DNS 解析和重试，对课程 demo 更省心。

心跳线程里用了stop_event，这个事件是主线程退出时通过set()通知后台线程停下来的。如果不做这个控制，客户端命令行退出了，心跳线程还会试图往已关闭的 socket 上发包，抛出一堆OSError把你的终端刷满异常堆栈。

3.3 联调跑通：两个终端的最小验证流程

代码写完后，第一次联调别加太多功能，按这三步走：

# 终端 1 python server.py # 终端 2 python client.py # 输入昵称后回车，看到 [服务器] 认证成功 # 再开一个终端 2 python client.py # 两个客户端互相发消息验证广播

按顺序能看到这些现象：服务端打印“已接入”，客户端打印“认证成功”，两个客户端之间发的消息能被广播。如果某一步断了，看抛出的异常类型：ConnectionRefusedError说明服务端没启动或端口不对，BrokenPipeError说明对端已经关闭连接，JSONDecodeError说明协议字段顺序或长度字段对不上。

第一次跑通不代表就稳了。关掉一个客户端，服务端如果没有立即打印“已退出”，不是 bug，而是 TCP 连接关闭通知需要等一次 recv 返回空数据才触发；正常拔掉网线的场景里，这一等可能等到天荒地老，那就是第 4 章要解决的问题了。

4. 网络程序设计常见坑与排查：粘包、僵尸连接与端口复用

4.1 粘包/拆包导致 JSON 解析失败：先看现象再给解法

现象：客户端 A 发了两条消息，客户端 B 的终端报错JSONDecodeError: Expecting value，或者看到一条消息里把两条消息的文字粘在一起。

原因：TCP 是字节流，没有报文边界。发送端调用两次send，内核可能在一次send里把两个小包拼成一个 TCP 段发出去；接收端一次recv拿到的数据里，就可能包含第一个包的完整内容和第二个包的前半截。这是 TCP 的天然行为，不是玄学。

解决：必须靠协议头里的长度字段做拆包。3.1 的parse_packets里已经有完整逻辑：先把收到的字节拼进 buf，循环检查长度字段，如果len(buf) < HEADER_LEN + length说明包还没到齐，退出循环等下一次 recv；如果长度够，就按 length 切出完整载荷并消费掉这一段。关键点是“消费”的写法：buf = buf[HEADER_LEN + length:]，漏掉这行，同一个包会被反复解析，客户端就会反复打出同一条消息。

排查时不要只盯着 JSON 报错看，把原始字节打出来定位：

# 在 recv 之后、解析之前打印，确认粘包的原始形态 bad = buf[:64].hex() print(f"buf head: {bad}")

如果看到头部不是abcd 0101 xxxx开头的规律，多半是协议字段顺序写错了，去对一下struct.pack("!HBBH", ...)的格式串。

4.2 客户端拔线后服务器毫无反应：僵尸连接的成因与处理

现象：客户端直接拔网线或Ctrl+C强杀进程，服务端日志里没有任何变化，clients表里还留着这个连接，往它发消息也不报错。

原因：TCP 断开有四种情况，正常关闭会收到 FIN，进程崩溃内核会自动发 RST，但拔网线、断 WiFi 这类物理断连，链路中间的路由器不会主动通知你，服务器上这个 socket 既不触发可读事件也不报错，操作系统默认要等 TCP keepalive 的两小时才清理。

解决：业务层心跳是唯一靠谱的方案。服务端已经实现了heartbeat_monitor，每 30 秒检查一次last_heartbeat，超过 90 秒就把连接清理掉。这个线程要在main()里启动，注意daemon=True，否则主循环意外退出时线程不结束，终端会卡住。另外可以在 socket 上开启系统级 keepalive 作为辅助：

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)

这只是第二道保险，默认参数下它不会比你的 90 秒阈值更早触发，别指望它。

4.3 重启服务端报 Address already in use：TIME_WAIT 的典型场景

现象：服务端Ctrl+C后立刻重启，报OSError: [Errno 98] Address already in use。

原因：主动关闭连接的一端会进入TIME_WAIT状态，持续 2 个 MSL（约 2 分钟）。服务端 accept 到连接后又主动 close，那服务端就是主动关闭方，这个端口上的连接会留在 TIME_WAIT，导致新进程无法 bind 同一个端口。这是 TCP 协议设计上保证旧连接的迟到的数据包不会污染新连接的机制，不是系统 bug。

解决：在 bind 之前加上SO_REUSEADDR，让新进程能复用处于 TIME_WAIT 的端口。

server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

如果加了还是报错，说明端口被一个活着的进程占着，查一下再决定杀不杀：

lsof -i :8888 ps -ef | grep server.py | grep -v grep

lsof看端口被哪个 pid 占用，ps确认是不是上次的服务端没退干净。开发期我习惯在 server.py 里先把这两行写死，省得每次重启都踩一遍。

4.4 多线程改同一张客户端表：Lock 没加对位置

现象：压测或多人联调时，偶尔出现某个客户端收到别人的消息、或者服务端抛RuntimeError: dictionary changed size during iteration，还有概率直接卡死。

原因：clients这个字典被多个连接线程同时读写——A 线程在clients.items()里遍历广播，B 线程在remove_client里 pop 同一个 key，Python 的 dict 在遍历时被修改就会抛异常。更隐蔽的是只加锁但没有加对位置，比如广播时全程锁住sendall，一个慢客户端就会把整张表锁死。

解决：锁的粒度要拆分。操作字典的快照遍历加锁，真正的网络 IO 放锁外：

def broadcast(sender, message): with clients_lock: targets = [(c, info["name"]) for c, info in clients.items() if c is not sender] for conn, _ in targets: try: conn.sendall(build_packet(0x04, json.dumps(message).encode())) except OSError: remove_client(conn)

注意先拿目标连接列表，再在锁外逐个 sendall。这样即使某个连接发送阻塞，也只影响它自己的线程，不会连累其他人抢占锁。这算是血泪经验：并发程序的锁，加宽了性能崩，加窄了数据崩，拿快照再发 IO 是课程规模下最均衡的写法。

5. 进阶验证：抓包、并发压测与功能扩展

5.1 用 tcpdump 抓到自己的协议包：确认三次握手与心跳

代码跑通只是第一步，答辩时最有效的展示场景是把抓包结果摆出来。课程实验的客户端和服务端都在本机，回环网卡。

sudo tcpdump -i lo -XX -nn port 8888

三个参数的作用：-i lo监听回环网卡，因为本机通信不会经过物理网卡；-XX同时显示十六进制和 ASCII 内容，能直接看到报文里的\xab\xcd魔数；-nn不做域名和端口名解析，避免把 8888 显示成某个服务的名字。

抓包时需要盯四个东西：TCP 三次握手的 SYN、SYN-ACK、ACK 三个报文；认证请求里跟随在 TCP 头后面的abcd 0101 ...魔数；心跳包周期性地出现，间隔和代码里配置的 30 秒一致；客户端退出时能看到 FIN 报文。如果你的协议头在网络层完全可见，说明封装没问题；如果只看到abcd出现在奇怪的位置，回去查一下是不是多套了一层 bytes 编解码。

用 Wireshark 会更直观，过滤表达式tcp.port == 8888，把 Wireshark 的抓包截图放进报告里，比贴十行代码更有说服力。

5.2 200 个连接压一小时：稳定吗，看什么指标

课程项目不用上压测工具，自己写一个连接脚本就够了。目标是确认：连接数上去之后，服务端不崩、心跳不误杀、fd 不泄漏。

# load_test.py：简单并发连接测试 import socket import json import struct import time PORT = 8888 def send(sock, type_: int, obj): payload = json.dumps(obj).encode() sock.sendall(struct.pack("!HBBH", 0xABCD, 1, type_, len(payload)) + payload) def open_conn(i): sock = socket.create_connection(("127.0.0.1", PORT), timeout=3) send(sock, 0x01, {"name": f"load-{i}"}) return sock conns = [open_conn(i) for i in range(200)] print(f"opened {len(conns)} conns, wait 60s...") time.sleep(60) alive = [] for i, sock in enumerate(conns): try: send(sock, 0x03, {}) alive.append(i) except OSError: pass print(f"alive: {len(alive)} / 200")

观察点三个。第一，脚本打印的 alive 数量应该是 200，如果有连接被判离线，说明你的心跳阈值太紧，或者 recv 超时设置挡住了数据。第二，压测期间在另一终端执行lsof -i :8888 | wc -l，数量应该稳定在 200 多一点，如果持续增长，说明有连接没被回收，去掉这条连接负责 remove 的日志看谁漏了。第三，压测结束后等 120 秒，再 lsof 一次，所有连接应该被心跳监控线程清干净，剩下的数量回到个位数。这步过了，才敢说服务端的连接管理是闭环的。

5.3 从聊天室扩到文件传输：二进制负载与分片设计

聊天室跑通后，想往上加一个文件传输功能，最规范的扩展是新增两种协议类型：0x06 文件分片，0x07 文件完成。注意不能再把整个文件塞进 JSON 里一次性发送，json.dumps 会把二进制内容做一次不对等的编码，大文件的内存峰值直接爆。正确做法是直接在载荷区放二进制：

def build_file_chunk(seq: int, total: int, data: bytes) -> bytes: ext = struct.pack("!HHI", 0x06, seq, total) # 扩展头 8 字节 return ext + data

接收方根据 seq 和 total 做数据重组，每片建议 4KB 或 16KB，不要超过协议头里 length 字段的 65535 上限。TCP 本身已经保证顺序，你只需要把分片按 seq 排好写到磁盘，不需要做重传。这步一旦做完，项目就从“聊天室”升级成了“带文件传输的多人协作工具”，课程设计的高度直接拉一个档次。

6. 验收答辩的三个技巧：让现场演示少翻车

6.1 先演“故障恢复”，再演主流程

答辩现场最怕的不是代码有 bug，而是演示到一半连接断了。与其祈祷服务器不崩，不如主动把故障场景作为第一个演示动作：启动两个客户端，当着老师的面Ctrl+C杀掉一个，服务端日志里打印“已退出”，第二个客户端正常收广播。再演示拔网线和重启服务端，30 秒内心跳监控日志打出“心跳超时，断开连接”。这个顺序比直接演示聊天效果更能打动评委，因为它展示的是你对异常路径的控制力。

6.2 抓包截图放进报告

一份课程报告如果只有代码，说服力不够。把第 5 章的 tcpdump 输出截一张清楚的三次握手图，标注出 SYN 和 ACK；再截一张心跳报文的时序图，把 30 秒间隔圈出来。口头答辩时，评委问到“你怎么证明你的协议是可靠的”，直接指这张图。

6.3 用刚才的压测数据代替“我测过了”

200 连接 60 秒存活率 100%，这个数据随口说说和拿终端截图出来完全是两个杀伤力。压测脚本跑完后，把终端的 alive 输出和lsof的计数存成一个 log 文件，答辩时贴到 PPT 里。老师说“你这个可能不稳定”，你把 200 连接的心跳存活率往上一摆，大部分疑问都会停下来。

我当年交这个项目的前一天才把僵尸连接的问题修好。答辩时老师只问了一个问题：客户端拔线之后，服务器怎么保证不崩。正是心跳监控线程那一页代码救了我。后来我养成了习惯，任何网络程序落地前，先定义协议字段，再写代码，最后抓包验证，按这个顺序做能少加三个班。希望帮到你。

本文还有配套的精品资源，点击获取