简介这是一套基于Python与Socket开发的局域网即时通讯与文件传输聊天软件面向企业内网通信与校园实训场景适合作为毕业设计、课程设计或工程实训项目。功能覆盖用户管理注册登录、人脸识别、IP绑定、好友与群组管理、实时聊天、文件传输、表情包与漂流瓶等并内置32款趣味小游戏支持双界面风格切换与多界面跳转。压缩包包含85个文件大小约13.41MB以Python脚本为核心搭配PyQt5界面图片、SQLite数据库、TensorFlow模型等资源服务端与客户端代码分离便于模块化学习。目前已有381人浏览学习。读者可获取完整项目源码、界面素材、数据库及功能模块参考既能理解Socket通信与GUI开发整体流程也能针对人脸识别、情感识别、敏感词检测等特色功能专项练习适合有一定Python基础并希望进阶的开发者。1. 局域网即时通讯的起点把Socket绑到网卡而不是127.0.0.1刚接触Python socket网络编程的人第一次写聊天程序往往 bind(127.0.0.1, 9000)然后发现只有本机能连同事那台电脑却连不上问题就出在绑错了地址。基于Python和socket做局域网即时通讯核心是“让内网其他机器能连上你的端口”消息和文件都走TCP长连接省掉HTTP轮询和外置数据库适合会议室、机房的内部工具开发也适合课程设计做小范围演示。要跑完这个项目单靠教科书里的Echo示例不够还要处理粘包、多客户端并发、文件分块、登录名单和防火墙放行。这里先从协议定义讲到可运行的代码段最后给一条局域网排错链。目标读者是准备自己写内部沟通或传输工具的开发和运维环境只需要Python 3.8和标准库。2. 自定义协议设计先解决粘包和半包再写Python Socket收发TCP是字节流不是消息流。同一次send可能和前后一次send一起被对端recv到也可能一次recv只读到半个应用消息这就是粘包和半包的来源。对重点数据我一般先在应用层定义一套“包头包体”再把Python socket的收发全部收敛到一组函数里避免业务代码里到处处理残缺数据。2.1 聊天和文件传输共用一条TCP连接的理由先想清楚为什么不用UDP。UDP适合局域网设备发现、心跳探测这类丢一个包也不影响全局的场景但聊天要保证到达顺序文件更不能静默丢分块。所以这里只用一条TCP长连接把聊天、文件、在线名单全部封装成消息。这样做的好处是客户端只需要记住一个IP和一个端口防火墙规则也只开一条端口坏处是服务端代码要分类型路由但这个复杂度可以接受。2.2 一个7字节的自定义包头magic、类型、长度我习惯把包头定成固定7字节用Python的struct以网络字节序写进去前2字节是magic用来识别自家协议中间1字节是类型最后4字节是后续payload长度。这样对端先读7字节再按长度读完整包体就能避免粘包导致的错位。字段类型字节数说明magicunsigned short2固定0xC7A1用于过滤垃圾连接typeunsigned char10心跳、1聊天、2文件元数据等payload_lengthunsigned int4payload的字节数最大不能超过64KB类型值不一定要设计得很复杂0到6已经够用心跳、聊天、文件元数据、文件数据、文件结束、登录、在线用户列表。文件数据的payload限制在64KB以内既避免一次recv分配超大内存又方便做分块传输。代码如下import socket import struct MAGIC 0xC7A1 MAX_PAYLOAD 64 * 1024 TYPE_HEARTBEAT 0 TYPE_CHAT 1 TYPE_FILE_META 2 TYPE_FILE_DATA 3 TYPE_FILE_END 4 TYPE_LOGIN 5 TYPE_ONLINE_USERS 6 def recv_exact(sock: socket.socket, n: int) - bytes | None: buf b while len(buf) n: chunk sock.recv(n - len(buf)) if not chunk: return None buf chunk return buf def send_packet(sock: socket.socket, typ: int, payload: bytes) - None: sock.sendall(struct.pack(!HBI, MAGIC, typ, len(payload)) payload) def recv_packet(sock: socket.socket) - tuple[int, bytes] | None: header recv_exact(sock, 7) if header is None: return None magic, typ, length struct.unpack(!HBI, header) if magic ! MAGIC: raise ValueError(fbad magic: {magic:#x}) if length MAX_PAYLOAD: raise ValueError(fpayload too large: {length}) payload recv_exact(sock, length) return (typ, payload) if payload is not None else None参数说明!HBI中的!表示网络字节序H占2字节B占1字节I占4字节拼起来正好是7字节包头。recv_exact循环调用sock.recv直到读满n字节或连接断开这样代码就不会被半包卡死。send_packet里的sendall会保证写入完整字节流但它不负责语义上的“消息边界”边界只由包头里的length决定。2.3 为什么要单独限制MAX_PAYLOAD如果不限制长度一个恶意或错误的连接发送一个“length99999999”的包头客户端就会一直等下去甚至申请大内存。MAX_PAYLOAD不是协议强制而是实现层的安全网。这里设64KB是因为文件数据块本身就按64KB切分聊天消息和元数据几乎不会超过。真的有大文件要传时会分成很多个64KB的文件数据包而不是把整个文件塞进一个payload。3. 多客户端并发与在线名单Socket服务端别用循环等待用Threading写局域网聊天服务端不能一次只服务一个客户端。最简单的做法是accecpt到一个连接就开一个线程连接内部用阻塞recv读消息。Python的GIL对这里影响很小因为线程大部分时间阻塞在socket recv上互不争抢CPU。3.1 一个连接一个线程每条消息在循环里分发每个客户端一个线程生命周期和TCP连接一致。线程内部循环调用上一章的recv_packet收到消息后按type分发。不需要为每条消息创建线程那样线程切换成本太高而且逻辑会乱。常见做法是import threading import socket clients: dict[str, socket.socket] {} clients_lock threading.Lock() def broadcast(sender: str, payload: bytes) - None: with clients_lock: targets list(clients.items()) for name, sock in targets: if name sender: continue try: send_packet(sock, TYPE_CHAT, payload) except OSError: with clients_lock: clients.pop(name, None) def client_loop(sock: socket.socket, addr): name None try: typ, payload recv_packet(sock) if typ ! TYPE_LOGIN: sock.close() return name payload.decode(utf-8) with clients_lock: clients[name] sock broadcast(name, f{name} 加入了聊天室.encode(utf-8)) while True: typ, payload recv_packet(sock) if typ is None: break elif typ TYPE_HEARTBEAT: send_packet(sock, TYPE_HEARTBEAT, b) elif typ TYPE_CHAT: broadcast(name, f{name}: {payload.decode(utf-8)}.encode(utf-8)) finally: if name: with clients_lock: clients.pop(name, None) broadcast(name, f{name} 已下线.encode(utf-8)) sock.close() server socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind((0.0.0.0, 9000)) server.listen(128) while True: conn, addr server.accept() threading.Thread(targetclient_loop, args(conn, addr), daemonTrue).start()核心逻辑说明进入client_loop后先读一条登录消息把昵称和socket放进clients字典。broadcast在遍历字典时先复制一份快照避免在遍历过程中被其他线程改动。finally块保证连接异常断开时在线名单里不会残留脏数据。3.2 在线名单的并发安全设计这里有几个状态需要单独说明状态项类型保护方式说明clientsdictclients_lock键是昵称值是socket对象serversocket单线程accept只在一个线程里accept每个连接socket独立线程阻塞在recv_packet里昵称相同时后面的客户端会覆盖前面的socket原连接会无法收消息。这种“强制上线”在内部工具里可以接受但更稳妥的做法是发现重名时服务端主动给自己返回一条错误包并关闭连接。我一般会给错误包单独定义一个type例如用TYPE_CHAT里的一个系统昵称回传避免客户端误把错误当聊天显示。3.3 客户端怎么维持在线状态服务端需要知道客户端是否还活着。最简单的心跳是客户端每30秒发一条TYPE_HEARTBEAT服务端收到后原样回一条如果连续90秒没有收到服务端在该连接的recv超时后主动清理。设置socket超时可以用sock.settimeout(90)但注意超时后recv_packet会抛TimeoutError需要在client_loop里捕获并退出。如果并不想做完整心跳也可以在客户端关闭窗口时发送一个type为TYPE_CHAT、payload为“bye”的消息服务端再清理。这个方案简单但面对断电断网会留下僵尸连接所以生产级的内部工具还是建议加入心跳。4. 在同一Socket通道里传文件分块发送、服务端中转与MD5校验聊天在同一个TCP连接上跑通后文件传输不需要新开端口。只要在协议里加文件元数据、文件数据、文件结束三个类型服务端把收到的文件数据原样转发给目标客户端。下面这套流程我用过多次适合课程设计和中小规模内网工具。4.1 发送方发元数据服务端建立一条虚拟转移路径发送方先发送TYPE_FILE_METApayload是JSON字符串包含目标昵称、文件名、文件大小。服务端解析后查找目标客户端如果目标在线就记录发送方对应的“当前传输上下文”并把元数据转发给接收方。这里每一步的字节序如下方向内容说明A - ServerJSON元数据to, filename, sizeServer - B元数据发送方昵称让B知道是谁传来的A - Server64KB文件数据块每个包独立Server - B原样转发文件数据块不落盘A - Server文件结束MD5表示传输完成Server - B文件结束MD5B对账服务端只做半中继收到文件数据包后不保存整个文件直接转发。这样即使公司局域网里两台机器不能直连也能通过服务端中转。代价是下行带宽多占一倍但对内网文件传输来说通常不是瓶颈。4.2 服务端转发文件数据块的代码实现文件数据块是典型的重复事件我单独写一个handle_file_meta和handle_file_data避免把业务都塞在client_loop里。import json import hashlib TRANSFER {} TRANSFER_LOCK threading.Lock() CHUNK_SIZE 64 * 1024 def handle_file_meta(sender: str, sock: socket.socket, payload: bytes) - None: meta json.loads(payload.decode(utf-8)) target_name meta.get(to) with clients_lock: target_sock clients.get(target_name) if target_sock is None: send_packet(sock, TYPE_CHAT, f用户 {target_name} 不在线.encode(utf-8)) return with TRANSFER_LOCK: TRANSFER[sender] { target: target_name, filename: meta[filename], size: meta[size], recv: 0, md5: hashlib.md5(), } send_packet(target_sock, TYPE_FILE_META, json.dumps({ sender: sender, filename: meta[filename], size: meta[size], }).encode(utf-8)) def handle_file_data(sender: str, sock: socket.socket, payload: bytes) - None: with TRANSFER_LOCK: context TRANSFER.get(sender) if not context: return context[recv] len(payload) context[md5].update(payload) with clients_lock: target_sock clients.get(context[target]) if target_sock is None: return send_packet(target_sock, TYPE_FILE_DATA, payload)参数说明TRANSFER字典的key是发送方昵称value是当前传输上下文。每次收到文件数据包服务端更新已接收字节数和MD5摘要再原样转发给目标。这样接收方不用知道发送方地址只要维护自己的连接即可。要注意多个文件同时传输时TRANSFER里一个发送方只有一个上下文因此“一个连接同一时刻只传一个文件”是前提否则就要额外引入文件ID来区分。4.3 客户端发送文件时的分块读取客户端不能把整个文件一次性read()进内存一定要循环读取固定大小的块。下面这段就是发送端最简写法def send_file(sock: socket.socket, target: str, path: str) - None: size os.path.getsize(path) meta json.dumps({to: target, filename: os.path.basename(path), size: size}).encode(utf-8) send_packet(sock, TYPE_FILE_META, meta) sent 0 with open(path, rb) as f: while True: chunk f.read(CHUNK_SIZE) if not chunk: break send_packet(sock, TYPE_FILE_DATA, chunk) sent len(chunk) progress sent / size print(f\r进度: {progress:.1%}, end, flushTrue) send_packet(sock, TYPE_FILE_END, b)这里CHUNK_SIZE保持64KB和协议层的MAX_PAYLOAD一致。如果机器内存小可以降到16KB但网络小包数量会变多。传输完成后再发一条空的TYPE_FILE_END接收端看到它才知道文件已经写完了。接收端对应的逻辑是收到TYPE_FILE_META后以wb模式打开文件收到TYPE_FILE_DATA就write一个分块收到TYPE_FILE_END就关闭文件并做MD5对账。MD5建议在接收端重新计算整个文件后和元数据里的值比对服务端计算的那个值只作为转发过程中的中间校验。5. 局域网部署排查bind失败、防火墙拦截、第一次连接就超时代码写完后真实局域网里最常见的问题不是语法错而是“服务端起来了但别人连不上”。这个章节直接从网络层面排查。5.1 绑定0.0.0.0而不是127.0.0.1并用IP查询命令确认地址127.0.0.1是回环地址只有本机能访问。要让局域网其他电脑连接服务端需要绑定0.0.0.0意思是监听本机所有网卡。写成代码就是server.bind((0.0.0.0, 9000))。在Windows上查看本机IPipconfig在Linux上查看hostname -I注意Windows的ipconfig输出里通常有好几段IP其中带“以太网适配器”或“无线局域网适配器”的IPv4地址才是其他机器可访问的地址。不要看127.0.0.1也不要看169.254.x.x这种无效地址。5.2 防火墙入站规则只放行端口不要动“网络发现”局域网聊天软件的端口是9000就需要在防火墙里放行TCP 9000。Windows管理员执行netsh advfirewall firewall add rule namechat-tcp-9000 dirin actionallow protocolTCP localport9000Linux如果用了ufwsudo ufw allow 9000/tcp这里注意不要额外开启“网络发现”或“文件和打印机共享”那和socket聊天不是一回事。排查步骤命令/操作期望输出不通过的常见原因1本机netstat -anofindstr :9000出现LISTENING记录2对端ping 服务端IP有回包IP地址填错不在同一网段3对端telnet 192.168.x.x 9000连接成功防火墙拦截、服务端没监听4杀毒软件/专用网络类型允许专用网络访问网络配置文件选了“公用”但没过规则“公用网络”有时会强制走更严格的防火墙策略这时要再把入站规则作用范围改成“专用网络”或者临时把网络配置文件切到“专用”。我自己在机房测试时通常先把防火墙临时关闭一次等确认端口通再重新打开防火墙逐条加规则这样能快速定位是系统防火墙还是商杀毒软件拦截。5.3 反复重启服务端时的bind: only one usage of each socket address经常改代码的人会遇到一个报错格式类似error: listen tcp 127.0.0.1:11434: bind: only one usage of each socket address这里报的是端口被占用。原因通常是上一次的服务端进程没有完全退出或者连接处于TIME_WAIT状态。排查方式就是上面的netstat -ano | findstr :9000拿到PID后在任务管理器里结束进程。代码层面最好的预防是绑定前设置SO_REUSEADDR让处于TIME_WAIT的旧连接不会阻塞新的监听server socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)参数说明SO_REUSEADDR在Python标准库的socket模块里是一个套接字选项它允许同一端口在旧连接未完全清理时重新绑定。注意在Windows和Linux上这个选项的行为略有不同但对我们这种局域网工具来说设置它基本只有好处。还有一个细节服务端要有优雅关闭逻辑尽量不要让用户在任务管理器里强杀进程否则端口占用排查会反复出现。6. 上线前验证技巧抓包看c7a1再用断点续传收尾所有功能写完后不要急着发给同事先用抓包和校验把协议落地验证清楚。6.1 用tcpdump一眼认出自家协议在Linux服务端执行tcpdump -i any -X -s0 tcp port 9000然后在客户端登录并发一条消息。抓到网络包后按ASCII和Hex混排展示前两个字节应该能看到c7 a1后面紧接着是消息类型和长度。这个过程能确认三层网络是通的协议栈也是通的比在代码里打日志更直观。如果用的是Wireshark过滤条件写tcp.port 9000然后在“流”里找前7字节的规律。6.2 让文件传输支持断点续传的偏移设计很多内网传输场景里几十MB的文件传到一半断线重新发一遍很浪费。常见做法是在元数据里增加offset字段接收方收到文件数据时记录已经写入文件的字节数下次发送文件前告诉发送方“我已经有前N个字节”发送方就从文件的第N字节开始读。核心代码是在发送端打开文件时seekoffset meta.get(offset, 0) with open(path, rb) as f: f.seek(offset) while True: chunk f.read(CHUNK_SIZE) if not chunk: break send_packet(sock, TYPE_FILE_DATA, chunk)接收端要做的配合是在收到文件元数据后先用ab模式打开文件或者先seek到已有文件大小再写。断点续传的价值不是省那一两次重传而是让大文件传输在任何一段网络闪断后都能从最后中断的位置继续。局域网里带宽大不及时但机房上传几百MB日志时这个参数能省大量时间。如果发现接收端文件大小和元数据里的size不一致就把已收到的字节数作为下一次请求的offset继续传。本文还有配套的精品资源点击获取