当前位置: 代码网 > it编程>前端脚本>Python > Python Socket中常用的代码模板汇总(最新整理)

Python Socket中常用的代码模板汇总(最新整理)

2026年07月31日 Python 我要评论
前言很多初学网络编程的小伙伴,弄懂socket、tcp、udp理论之后,最头疼的问题:到底有哪些通用代码模板?网上代码零散,版本不一,经常出现端口占用、连接断开、编码报错、windows/linux兼

前言

很多初学网络编程的小伙伴,弄懂socket、tcp、udp理论之后,最头疼的问题:到底有哪些通用代码模板?

网上代码零散,版本不一,经常出现端口占用、连接断开、编码报错、windows/linux兼容性问题。

本文整理python标准库socket最常用、工程上高频使用的基础代码,全部基于python3原生socket,无需第三方库,复制即可运行。

前置知识点:

  • tcp:面向连接、可靠、流式传输
  • udp:无连接、不可靠、数据包传输
  • 所有示例默认使用 ipv4 (af_inet)

一、通用基础知识点

导入模块

import socket

核心参数说明:

  • 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 服务端(基础版)

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

import socket

def tcp_server():
    # 创建socket对象 tcp
    server_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}")

    while true:
        # 阻塞等待客户端接入
        client_conn, client_addr = server_sock.accept()
        print(f"新客户端连接:{client_addr}")
        try:
            while true:
                # 最多接收1024字节
                data = client_conn.recv(1024)
                if not data:
                    # 空数据 = 客户端主动断开连接
                    break
                msg = data.decode("utf-8")
                print(f"收到[{client_addr}]消息:{msg}")
                # 回复客户端
                resp = f"服务端已收到:{msg}"
                client_conn.sendall(resp.encode("utf-8"))
        except exception as e:
            print("连接异常", e)
        finally:
            client_conn.close()
            print(f"客户端 {client_addr} 已断开")

if __name__ == "__main__":
    tcp_server()

2. tcp 客户端(基础版)

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

import socket

def tcp_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永久卡死

import socket

def tcp_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())
    except socket.timeout:
        print("连接/接收超时!")
    except connectionrefusederror:
        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

import socket

def udp_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")

    while true:
        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 客户端

import socket

def udp_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程序经常需要获取本机地址

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

print(get_local_ip())

2. 判断端口是否被占用

调试socket必备工具

import socket

def is_port_used(host="127.0.0.1", port=8888):
    sock = socket.socket(socket.af_inet, socket.sock_stream)
    sock.settimeout(1)
    result = false
    try:
        sock.connect((host, port))
        result = true
    except connectionrefusederror:
        result = false
    finally:
        sock.close()
    return result

print(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、自研内网穿透、代理隧道

到此这篇关于python socket中常用的代码模板汇总(最新整理)的文章就介绍到这了,更多相关python socket代码模板内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网!

(0)

相关文章:

版权声明:本文内容由互联网用户贡献,该文观点仅代表作者本人。本站仅提供信息存储服务,不拥有所有权,不承担相关法律责任。 如发现本站有涉嫌抄袭侵权/违法违规的内容, 请发送邮件至 2386932994@qq.com 举报,一经查实将立刻删除。

发表评论

验证码:
Copyright © 2017-2026  代码网 保留所有权利. 粤ICP备2024248653号
站长QQ:2386932994 | 联系邮箱:2386932994@qq.com