前言

很多初学网络编程的小伙伴，弄懂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、自研内网穿透、代理隧道

七、拓展方向

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

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