本文基于Python原生socket库，无需第三方依赖，从基础概念、核心区别、TCP双向通信、UDP无连接通信、异常处理、粘包解决、生产避坑全方位实战讲解。所有代码均经过本地实测可直接运行，适合零基础自学、面试备考、项目开发落地，同时适配搜索引擎收录规则，关键词布局合理。

一、Socket核心基础概念

1. 什么是Socket

Socket本质是网络通信的端点，是操作系统提供的网络编程接口。程序通过Socket绑定IP和端口，实现跨设备、跨网络的数据收发。

简单理解：IP地址定位设备，端口定位程序，Socket实现两个程序的数据通道。

Python内置socket标准库，无需pip安装，直接导入即可使用，兼容Python3.6+所有版本。

2. TCP与UDP核心区别（面试+实战重点）

对比维度	TCP（传输控制协议）	UDP（用户数据报协议）
连接特性	面向连接，三次握手建立通道	无连接，无需建立通道，直接发包
可靠性	可靠传输，丢包重传、有序到达	不可靠，无重传机制，可能丢包、乱序
传输速度	较慢，有校验、握手开销	极速，无额外校验开销
数据限制	流式传输，无数据包边界	数据报传输，单次数据包有限制
适用场景	文件传输、网页、聊天通信、登录接口（要求稳定）	直播、语音视频、游戏、物联网上报（要求速度）

3. Socket核心参数说明

创建Socket固定语法：socket.socket(family, type)

• AF_INET：IPv4协议（99%场景使用）

• SOCK_STREAM：TCP流式套接字

• SOCK_DGRAM：UDP数据报套接字

二、TCP通信完整实战（面向连接、稳定通信）

TCP通信必须遵循服务端先启动、客户端后连接的规则，流程：创建套接字→绑定端口→监听连接→三次握手建立通道→双向收发数据→关闭连接。

1. TCP服务端代码（可直接运行）

功能：监听本地端口，接收客户端连接，双向收发消息，异常自动容错

# tcp_server.py
import socket

# 1. 创建TCP套接字
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 2. 解决端口复用问题，避免重启提示端口被占用
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

# 3. 绑定IP和端口，空IP代表监听本机所有网卡
HOST = '0.0.0.0'
PORT = 8888
server_socket.bind((HOST, PORT))

# 4. 开启监听，参数为最大等待连接数
server_socket.listen(5)
print(f"TCP服务端启动成功，监听端口：{PORT}")

while True:
    # 阻塞等待客户端连接
    client_conn, client_addr = server_socket.accept()
    print(f"收到客户端连接：{client_addr}")

    try:
        # 循环收发数据
        while True:
            # 单次最大接收1024字节数据
            recv_data = client_conn.recv(1024)
            if not recv_data:
                # 客户端主动断开连接
                print(f"客户端{client_addr}断开连接")
                break
            # 字节解码为字符串
            recv_msg = recv_data.decode("utf-8")
            print(f"接收{client_addr}消息：{recv_msg}")

            # 服务端回复消息
            send_msg = f"服务端已收到：{recv_msg}"
            client_conn.send(send_msg.encode("utf-8"))
    except Exception as e:
        print(f"通信异常：{e}")
    finally:
        # 关闭当前客户端连接
        client_conn.close()

2. TCP客户端代码（可直接运行）

# tcp_client.py
import socket

# 1. 创建TCP套接字
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 2. 连接服务端
HOST = '127.0.0.1'
PORT = 8888
client_socket.connect((HOST, PORT))

try:
    # 发送消息
    send_msg = "Hello TCP Socket"
    client_socket.send(send_msg.encode("utf-8"))

    # 接收服务端回复
    recv_data = client_socket.recv(1024)
    print("服务端回复：", recv_data.decode("utf-8"))
except Exception as e:
    print("连接异常：", e)
finally:
    # 关闭套接字
    client_socket.close()

3. 运行测试步骤

1. 单独打开终端，运行服务端：python tcp_server.py

2. 再打开新终端，运行客户端：python tcp_client.py

3. 双向数据收发成功，即为通信正常

4. TCP核心问题：粘包现象与解决

什么是粘包？

TCP是流式协议，无数据边界，连续发送多条短数据时，会被系统合并为一次接收，导致数据粘连、解析错乱，是TCP开发最常见的坑。

粘包解决方案（生产通用）

• 固定报文长度：每次收发固定字节数数据

• 首尾分隔符：自定义特殊符号分割数据包

• 头部声明长度：报文头部携带数据长度，精准接收

三、UDP通信完整实战（无连接、高速通信）

UDP无需建立连接，不用listen、accept，直接通过sendto发送数据、recvfrom接收数据，代码更简洁，传输效率更高。

1. UDP服务端代码

# udp_server.py
import socket

# 创建UDP套接字 SOCK_DGRAM
udp_server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
udp_server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

# 绑定端口
HOST = "0.0.0.0"
PORT = 9999
udp_server.bind((HOST, PORT))
print(f"UDP服务端启动成功，监听端口：{PORT}")

while True:
    # 接收数据+客户端地址
    recv_data, client_addr = udp_server.recvfrom(1024)
    recv_msg = recv_data.decode("utf-8")
    print(f"收到{client_addr}消息：{recv_msg}")

    # 原路回复数据
    send_msg = f"UDP服务端响应：{recv_msg}"
    udp_server.sendto(send_msg.encode("utf-8"), client_addr)

2. UDP客户端代码

# udp_client.py
import socket

# 创建UDP套接字
udp_client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

HOST = "127.0.0.1"
PORT = 9999

# 发送数据到服务端
send_msg = "Hello UDP Socket"
udp_client.sendto(send_msg.encode("utf-8"), (HOST, PORT))

# 接收服务端响应
recv_data, _ = udp_client.recvfrom(1024)
print("服务端回复：", recv_data.decode("utf-8"))

udp_client.close()

3. UDP核心特性

• 无连接：客户端无需connect，随时发包

• 自带地址：接收数据时可直接获取客户端IP和端口

• 无粘包问题：UDP是数据报协议，数据包独立边界清晰

• 可能丢包：网络波动时数据会丢失，不适合关键数据传输

四、Socket通用高频问题与解决方案

1. 端口被占用报错

报错信息：Address already in use

解决方法：添加端口复用配置，放在bind之前

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

2. 局域网无法通信

原因：服务端绑定127.0.0.1仅本机访问，防火墙拦截端口

解决方法：绑定0.0.0.0监听所有网卡，关闭防火墙或放行对应端口

3. recv接收不到数据、程序卡死

原因：recv是阻塞方法，无数据时会一直等待

解决方法：设置超时时间，避免永久阻塞

server_socket.settimeout(10)  # 10秒无数据自动超时

4. 中文乱码问题

原因：编码格式不统一

解决方法：统一使用utf-8编码，发送encode、接收decode

5. TCP客户端异常断开，服务端卡死

解决方法：增加try-except异常捕获，断开后自动释放连接，继续监听新客户端

五、TCP与UDP选型实战建议

优先使用TCP的场景

• 文件上传下载、数据同步

• 网页访问、接口请求、账号登录

• 即时聊天、指令下发（需要100%送达）

优先使用UDP的场景

• 视频直播、语音通话、游戏对局

• 物联网设备心跳上报、传感器数据推送

• 高并发低延迟、允许少量数据丢失的场景

六、进阶拓展：多线程TCP服务端（支持多客户端同时连接）

原生单线程TCP服务端同一时间只能处理一个客户端，生产环境需支持多客户端并发，搭配多线程实现高并发通信。

# 多线程TCP服务端（支持多客户端）
import socket
import threading

HOST = "0.0.0.0"
PORT = 8888

# 客户端处理函数
def handle_client(conn, addr):
    print(f"新客户端接入：{addr}")
    while True:
        try:
            data = conn.recv(1024)
            if not data:
                break
            msg = data.decode("utf-8")
            print(f"{addr}：{msg}")
            conn.send(f"已接收：{msg}".encode("utf-8"))
        except:
            break
    conn.close()
    print(f"客户端{addr}断开连接")

if __name__ == "__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(5)
    print("多线程TCP服务端启动成功")

    while True:
        conn, addr = server.accept()
        # 每个客户端开启独立线程
        t = threading.Thread(target=handle_client, args=(conn, addr))
        t.daemon = True
        t.start()