当前位置: 首页 > news >正文

Python Socket 常用代码汇总|TCP/UDP 服务端、客户端基础模板

前言

很多初学网络编程的小伙伴,弄懂Socket、TCP、UDP理论之后,最头疼的问题:到底有哪些通用代码模板?
网上代码零散,版本不一,经常出现端口占用、连接断开、编码报错、Windows/Linux兼容性问题。
本文整理Python标准库socket最常用、工程上高频使用的基础代码,全部基于Python3原生socket,无需第三方库,复制即可运行。

前置知识点:
TCP:面向连接、可靠、流式传输
UDP:无连接、不可靠、数据包传输
所有示例默认使用 IPv4 (AF_INET)

一、通用基础知识点

导入模块

importsocket

核心参数说明:

  • socket.AF_INET:IPv4
  • socket.SOCK_STREAM:TCP
  • socket.SOCK_DGRAM:UDP
  • bind((ip, port)):绑定地址端口
  • listen():开启监听(仅TCP)
  • accept():阻塞等待客户端连接(仅TCP)
  • recv(buffer):接收数据
  • send()/sendall():发送数据

⚠️ 重点:网络传输只能发送bytes字节流,字符串必须encode();收到字节需要decode()


二、TCP 全套常用模板

1. TCP 服务端(基础版)

持续监听,接收客户端消息,回传数据

importsocketdeftcp_server():# 创建socket对象 TCPserver_sock=socket.socket(socket.AF_INET,socket.SOCK_STREAM)# 解决服务端重启端口立即复用,避免 WinError 10048 端口占用server_sock.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)# 绑定地址 0.0.0.0 监听本机所有网卡addr=("0.0.0.0",8888)server_sock.bind(addr)# 开启监听,参数为等待连接队列大小server_sock.listen(5)print(f"TCP服务端启动成功,监听{addr}")whileTrue:# 阻塞等待客户端接入client_conn,client_addr=server_sock.accept()print(f"新客户端连接:{client_addr}")try:whileTrue:# 最多接收1024字节data=client_conn.recv(1024)ifnotdata:# 空数据 = 客户端主动断开连接breakmsg=data.decode("utf-8")print(f"收到[{client_addr}]消息:{msg}")# 回复客户端resp=f"服务端已收到:{msg}"client_conn.sendall(resp.encode("utf-8"))exceptExceptionase:print("连接异常",e)finally:client_conn.close()print(f"客户端{client_addr}已断开")if__name__=="__main__":tcp_server()

2. TCP 客户端(基础版)

主动连接TCP服务端,发送消息并接收返回

importsocketdeftcp_client():client_sock=socket.socket(socket.AF_INET,socket.SOCK_STREAM)server_addr=("127.0.0.1",8888)# 连接服务端client_sock.connect(server_addr)print("成功连接服务端")send_msg="Hello Socket TCP"client_sock.sendall(send_msg.encode("utf-8"))# 接收响应recv_data=client_sock.recv(1024)print("服务端返回:",recv_data.decode("utf-8"))client_sock.close()if__name__=="__main__":tcp_client()

3. TCP 客户端设置超时(非常常用)

防止网络阻塞,recv、connect永久卡死

importsocketdeftcp_client_timeout():client_sock=socket.socket(socket.AF_INET,socket.SOCK_STREAM)client_sock.settimeout(5)# 设置超时5秒try:client_sock.connect(("127.0.0.1",8888))client_sock.sendall(b"test timeout")data=client_sock.recv(1024)print(data.decode())exceptsocket.timeout:print("连接/接收超时!")exceptConnectionRefusedError:print("无法连接,服务端未启动")finally:client_sock.close()if__name__=="__main__":tcp_client_timeout()

TCP注意事项:

  1. TCP是流式协议,存在粘包问题,大批量传输需要自定义数据包长度协议;
  2. 优先使用sendall()send()不一定一次性发送完所有数据;
  3. recv返回空字节代表对方关闭连接。

三、UDP 全套常用模板

1. UDP 服务端

UDP无连接,不需要listen、accept

importsocketdefudp_server():udp_sock=socket.socket(socket.AF_INET,socket.SOCK_DGRAM)udp_sock.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)udp_sock.bind(("0.0.0.0",9999))print("UDP服务端启动,端口9999")whileTrue:data,addr=udp_sock.recvfrom(1024)msg=data.decode("utf-8")print(f"来自{addr}:{msg}")# 回包reply=f"UDP收到消息:{msg}"udp_sock.sendto(reply.encode("utf-8"),addr)if__name__=="__main__":udp_server()

2. UDP 客户端

importsocketdefudp_client():udp_sock=socket.socket(socket.AF_INET,socket.SOCK_DGRAM)server_addr=("127.0.0.1",9999)msg="Hello UDP Socket"udp_sock.sendto(msg.encode("utf-8"),server_addr)# 等待接收返回data,addr=udp_sock.recvfrom(1024)print("服务端响应:",data.decode("utf-8"))udp_sock.close()if__name__=="__main__":udp_client()

UDP重点提醒:

  1. 无连接,就算服务端没启动,sendto也不会报错;
  2. 不存在连接断开的概念,不会感知对方是否在线;
  3. 不存在粘包,数据包独立;
  4. 不保证送达、不保证顺序。

四、高频实用工具代码片段

1. 获取本机内网IP

写socket程序经常需要获取本机地址

defget_local_ip():try:s=socket.socket(socket.AF_INET,socket.SOCK_DGRAM)# 随便连接一个公网地址,不真实发包,用来获取本机网卡IPs.connect(("8.8.8.8",80))ip=s.getsockname()[0]s.close()returnipexceptException:return"127.0.0.1"print(get_local_ip())

2. 判断端口是否被占用

调试socket必备工具

importsocketdefis_port_used(host="127.0.0.1",port=8888):sock=socket.socket(socket.AF_INET,socket.SOCK_STREAM)sock.settimeout(1)result=Falsetry:sock.connect((host,port))result=TrueexceptConnectionRefusedError:result=Falsefinally:sock.close()returnresultprint(is_port_used(port=8888))

3. 简单十六进制收发(二进制通信常用)

很多硬件通信、内网隧道都是传输二进制数据

# 字符串转字节发送raw_data=bytes.fromhex("01 02 03 0A FF")# sock.sendall(raw_data)# 收到字节打印十六进制# data = sock.recv(1024)# print(data.hex())

五、常见踩坑总结(开发必看)

  1. 端口占用 WinError 10048
    解决:添加setsockopt(SO_REUSEADDR,1)
  2. 客户端只能本地访问,外部设备连不上
    原因:服务端绑定127.0.0.1,改为0.0.0.0
  3. UnicodeDecodeError 编码错误
    解决方案:通信双方约定编码(utf-8),不要直接解码未知二进制
  4. recv卡死程序
    解决方案:设置settimeout()超时捕获异常
  5. TCP大量数据收发乱码/拼接在一起
    根源:TCP粘包,需要自定义封包协议(先发送数据包长度)

六、学习路线建议

  1. 先跑通本文TCP客户端+服务端,理解连接生命周期
  2. 再学习UDP,理解无连接通信差异
  3. 解决TCP粘包,实现自定义协议通信
  4. 进阶:多线程处理多客户端连接(上面示例单线程只能处理一个客户端)
  5. 高阶:asyncio异步socket、自研内网穿透、代理隧道

七、拓展方向

掌握以上基础模板后,可以基于这些代码开发:

  • 简易聊天室
  • 内网穿透客户端/服务端
  • 硬件设备TCP/UDP数据采集
  • 简单端口扫描器
  • 自定义TCP代理服务
http://www.cnnetsun.cn/news/3763170.html

相关文章:

  • MinIO IAM Policy配置全解析:从基础概念到高级权限管理实战
  • STC单片机烧录全攻略:从原理到实战,解决常见问题
  • 压敏电阻原理与应用:从防雷卫士到电路保护设计实战
  • C++ OpenCV图像处理:LUT查找表原理与实战性能优化
  • Nginx反向代理HTTPS服务报错排查:The plain HTTP request was sent to HTTPS port
  • 逻辑分析仪阈值电压硬核选购标准全解析|电平判决原理、多电压系统适配、阈值设置避坑实战指南
  • 基于Transformer的风电功率预测MATLAB实现
  • 企业级RAG技术:智能知识库的核心架构与优化实践
  • 终极指南:55项炉石传说插件功能全面解锁游戏新境界
  • 捷米特 JM-RS-WIFI 数传模块,智能仓储 AGV 小车串口无线通讯解决方案
  • 扣子变量传递失效真相(生产环境血泪调试实录):3类隐式类型转换陷阱正在 silently 毁掉你的自动化流程
  • 嵌入式开发实战:STM32按键消抖原理、状态机实现与RTOS应用
  • 大模型应用开发工程师:2026年高薪转行风口,小白也能抓住的机会!收藏必备!
  • BetterGI技术深度解析:5大核心技术构建原神自动化辅助框架
  • 实测:如何检测你的网站是否被 ChatGPT、豆包、Perplexity 引用?附 30 秒免费工具
  • STM32独立看门狗(IWDG)原理、配置与实战设计指南
  • OpenSpeedy终极指南:免费开源游戏加速工具完整教程
  • Satechi Thunderbolt 5 CubeDock 评测:创新 SSD 扩展坞,性能出色但有局限
  • 手动安装 OpenAI Codex Windows 桌面版
  • Arduino开发工具链解析:从图形化编程到专业IDE的进阶指南
  • AI写稿与数据增强:技术挑战与伦理实践
  • 华硕笔记本性能管理终极方案:G-Helper轻量控制工具完全指南
  • 实测允安金属木质围墙护栏:亮点与短板揭秘,适合这些人群!
  • RT-Thread内核与BSP版本不一致:编译报错排查与修复指南
  • 告别重复操作:阴阳师自动化脚本如何让你每天节省3小时游戏时间
  • STM32 GPIO深度解析:从电路原理到标准库实战应用
  • 我的设计banner
  • Android关机重启广播监听:ACTION_SHUTDOWN实现与数据持久化实战
  • 汇编语言寻址方式实战:数据结构设计与循环处理内存数据
  • 3分钟搞定FanControl风扇控制软件中文设置:让你的Windows风扇彻底“说中文“