第十二章 Python之网络编程实战

前言

学网络编程的时候，很多人一开始会觉得自己懂了——IP 地址、端口号、TCP/UDP，这些概念背得滚瓜烂熟。但真到了写代码的时候，才发现到处都是坑：为什么端口被占用？为什么 recv 收不到完整数据？为什么服务只能连一个客户？

这篇文章不会重复教科书式的概念 ，而是在系统梳理原笔记全部知识点的基础上，用大白话把那些"教程一带而过但实际开发必踩的坑"讲透。

1 网络编程概述

1.1 什么是网络编程

网络编程，就是让在不同的电脑上的软件能够进行数据传递，即进程之间的通信。

打个比方：

• 你打开微信给朋友发消息 → 你的微信进程和朋友的微信进程在通信
• 你在浏览器访问百度 → 你的浏览器进程和百度的服务器进程在通信

网络编程的本质：就是写代码让两个进程通过网络"聊天"。

2 网络编程三要素：IP、端口、协议

网络编程有三个核心要素，缺一不可：

类比理解：

• 你要给朋友寄快递 → 需要地址（IP）+ 门牌号（端口）+ 快递怎么寄的规则（协议）
• 少了地址，快递员找不到人；少了门牌号，快递送到大楼不知道给谁；少了规则，快递可能寄丢

3 IP 地址：网络中的门牌号

3.1 什么是 IP

IP 地址是一串数字，用来标识一台电脑在网络中的位置。

当你的电脑连接网络时，运营商会给你分配一个 IP 地址。通过这个地址，网络上的其他设备才能找到你。

IP 地址的两个核心功能：

1. 标识设备：告诉网络"我是谁"
2. 寻址：告诉路由器"数据包该发到哪里"

查看本机 IP：

# Windows 用户
ipconfig

# Linux / macOS 用户
ifconfig
# 或者
ip addr

3.2 子网掩码与 CIDR

3.2.1 为什么需要子网掩码

IP 地址虽然能标识设备，但当网络规模变大时，管理起来就很麻烦。比如一个公司有 1000 台电脑，如果都在一个网络里，广播消息会让整个网络瘫痪。

解决办法：把大网络切成小网络（子网），每个子网独立管理。

3.2.2 子网掩码的作用

子网掩码（Subnet Mask）的作用是：把 IP 地址分成两部分——网络前缀和主机编号。

IP 地址：    192.168.10.2
子网掩码：  255.255.255.0

解读：

• 子网掩码 255.255.255.0 的前 24 位都是 1，表示 IP 地址的前 24 位是网络部分
• 剩下的 8 位是主机部分，标识这个网络里的具体设备

3.2.3 CIDR 表示法

CIDR（Classless Inter-Domain Routing，无类别域间路由）是一种更简洁的表示方法：

192.168.10.2/24

/24 表示 IP 地址的前 24 位是网络部分，等价于子网掩码 255.255.255.0。

为什么 CIDR 更好？

• 更简洁：写 192.168.10.2/24 比写 192.168.10.2 255.255.255.0 方便
• 更灵活：可以任意划分网络大小，不局限于 A/B/C 类

3.2.4 特殊地址

3.3 IPv4 地址的分类

IPv4 地址根据网络规模分为 5 类：

记忆口诀：A 大 B 中 C 小，D 组播 E 保留。

💡 实战提示：127.0.0.1 是个特殊地址，属于 A 类但用于回路测试（Loopback），永远指向本机，即使网卡没配 IP 也能 ping 通。

3.4 公网 IP 与私网 IP

3.4.1 区别

3.4.2 私网 IP 范围

10.0.0.0    ~ 10.255.255.255   （A 类私网，超大型内网）
172.16.0.0  ~ 172.31.255.255   （B 类私网，中型内网）
192.168.0.0 ~ 192.168.255.255 （C 类私网，家庭/办公室）

127.0.0.1 ~ 127.255.255.255：回路测试专用，127.0.0.1 就是 localhost。

3.4.3 NAT（网络地址转换）

问题：IPv4 地址只有约 42 亿个，不够全球每台设备分一个。

解决办法：NAT 技术。

NAT 的原理：

• 家里的路由器有一个公网 IP（比如 123.45.67.89）
• 家里的手机、电脑、平板都用私网 IP（比如 192.168.1.2、192.168.1.3）
• 当手机访问百度时，路由器把手机的私网 IP 替换成公网 IP，百度看到的请求来自 123.45.67.89
• 百度回复时，路由器再把公网 IP 替换回私网 IP，发给手机

好处：

• 多台设备共享一个公网 IP，解决地址短缺
• 外部网络无法直接访问内部设备，有一定安全性

💡 为什么 NAT 流行？ 1990 年代中期 IPv4 地址开始枯竭，NAT 让多台主机共享一个公网 IP，既解决了地址短缺，又成了家庭路由器的标准配置。

3.5 IPv4 与 IPv6

IPv6 有多大？ 可以给地球上每一粒沙子都分配一个 IP 地址。

💡 现状：虽然 IPv6 推广多年，但 IPv4 + NAT 仍然是主流。写代码时建议同时兼容两者。

3.6 实战：获取本机真实 IP

很多新手用 socket.gethostbyname(socket.gethostname()) 获取本机 IP，这在某些系统上会返回 127.0.0.1。靠谱的做法：

import socket

def get_local_ip():
    """获取本机真实 IP（不是 127.0.0.1）"""
    try:
        # 创建一个 UDP 套接字，不需要真正连接
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
            s.connect(("8.8.8.8", 80))  # Google 的 DNS
            ip = s.getsockname()[0]
            return ip
        except Exception:
            return"127.0.0.1"

print(f"本机 IP: {get_local_ip()}")

原理：创建一个 UDP 包发往 8.8.8.8（Google DNS），操作系统会选出正确的网卡和 IP，然后我们通过 getsockname() 获取这个 IP。

4 端口：进程的门牌号

4.1 什么是端口

端口（Port）是计算机中的逻辑概念，用于标识进程（应用软件）。

类比：

• IP 地址 = 大楼的地址
• 端口 = 大楼里的门牌号

一台电脑有多个软件同时运行（微信、浏览器、QQ），数据包到了电脑之后，怎么知道该发给谁？靠的就是端口。

端口号范围：0 ~ 65535，共 65536 个。

查看端口占用：

# Windows
netstat -ano

# Linux / macOS
netstat -tlnp
# 或
ss -tlnp

4.2 端口号的分配

⚠️ 注意：端口 0 是保留端口，不可使用。1~1023 系统保留，只能由 root 用户使用。

4.3 常见端口速查表

💡 实战建议：写服务程序时，别用 8080，这年头比公共厕所还容易被占用。用 8000 以上，或者从配置文件读取。

4.4 实战：检测端口是否可用

import socket

def is_port_in_use(port, host='0.0.0.0'):
    """检测端口是否被占用"""
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        try:
            s.bind((host, port))
            return False
        except OSError:
            return True

def find_free_port(start_port=8000):
    """找一个可用端口"""
    port = start_port
    while is_port_in_use(port):
        port += 1
     return port

print(f"8080 是否可用: {not is_port_in_use(8080)}")
print(f"推荐端口: {find_free_port()}")

5 通信协议：规则决定一切

5.1 什么是通信协议

通信协议是一组规则，规定了不同设备之间如何进行数据交换。

类比：

• 两个人写信，需要约定：用什么语言、信封怎么写、邮票贴哪里
• 网络通信用协议规定：数据怎么打包、怎么传输、出错怎么办

协议定义了通信的各个方面：

• 数据的格式：数据怎么打包？比如一个数据包多大？头部放什么信息？
• 传输的顺序：先发什么？后发什么？比如先发送长度，再发数据
• 错误检查机制：数据出错了怎么发现？比如用校验和检查数据是否被篡改
• 如何处理异常情况：数据丢了怎么办？网络断了怎么重连？

5.2 为什么要分层

问题：网络通信很复杂，涉及硬件、驱动、路由、应用等多个环节。如果所有功能都混在一起，改一个地方就可能影响全局。

解决办法：分层。

类比：公司组织架构

• CEO → 总经理 → 部门经理 → 员工
• 每个层级只负责自己的事情，不越权
• 改一个部门不影响其他部门

分层的好处：

1. 降低复杂度：每层只关注自己的功能
2. 标准化：每层定义好接口，不同厂商可以互操作
3. 易于维护：改一层不影响其他层

5.3 分层网络模型

5.3.1 OSI 七层模型（理论标准）

OSI（Open System Interconnection，开放式系统互联）七层网络模型由国际标准化组织制定。

类比：一个公司，从高到低有 7 个层级：

7. 应用层  —— 直接为用户提供服务（HTTP、FTP、SMTP）
            类比：前台接待员，直接和客户打交道

6. 表示层  —— 数据格式转换、加密、压缩
            类比：翻译官，把数据翻译成接收方能懂的语言

5. 会话层  —— 建立、管理、终止会话
            类比：秘书，负责安排会议、记录会议内容

4. 传输层  —— 端到端的数据传输（TCP、UDP）
            类比：快递公司，负责把数据从一端送到另一端

3. 网络层  —— 路由和寻址（IP、ICMP）
            类比：导航系统，决定数据包走哪条路

2. 数据链路层 —— 帧的传输（以太网、WiFi）
            类比：卡车司机，负责在一段路上运输数据

1. 物理层  —— 物理介质（电缆、光纤、无线电波）
            类比：公路、铁路、航线，实际的物理传输介质

问题：OSI 模型虽然理论完美，但实现过于复杂，且制定周期过长。

5.3.2 TCP/IP 四层模型（实际标准）

在 OSI 标准推出之前，TCP/IP 模型已经在全球广泛使用。它把 7 层精简为 4 层：

4. 应用层  —— 对应 OSI 的 5~7 层（前台+翻译官+秘书的活一起干）
3. 传输层  —— 对应 OSI 的第 4 层（快递公司）
2. 网络层  —— 对应 OSI 的第 3 层（导航系统）
1. 网络接口层 —— 对应 OSI 的 1~2 层（司机+公路一起管）

5.3.3 五层模型（学习中最常用）

由于 TCP/IP 模型没有定义网络接口层的具体内容，实际学习和开发中，通常将其替换为 OSI 的数据链路层和物理层，这就是五层网络模型：

5. 应用层
4. 传输层
3. 网络层
2. 数据链路层
1. 物理层

5.4 TCP/IP 协议族

TCP/IP 协议族得名于两个核心协议：

• TCP（传输控制协议）：面向连接、可靠传输
• IP（网际协议）：负责路由和寻址

类比：

• IP = 导航系统 + 快递公司的地址系统 —— 知道地址和路线
• TCP = 快递公司的保险服务 —— 保证包裹安全送达

常见网络协议（按层级）：

6 Socket 套接字：网络编程的万能接口

6.1 什么是 Socket

Socket（套接字）是同一或不同电脑的进程间通信的工具。

类比：

• Socket 就像电话机——打电话需要电话机，网络通信需要 Socket
• Socket 就像网络文件——读写 Socket 就像读写文件，只不过数据通过网络传输

为什么需要 Socket？

• 进程想要通过网络发送数据，不能直接操作网卡
• Socket 是操作系统提供的抽象接口，进程通过 Socket 间接操作网络
• 就像你打电话不需要知道基站怎么工作，只需要拿起来拨号就行

特点：

• 进程之间想要进行网络通信，必须基于 Socket
• 只要与网络相关的应用程序或者软件，都使用到了 Socket

6.2 Socket 的类型

Socket 主要分两种：

流式套接字（TCP）：

• 先建立连接，再传输数据
• 保证数据不丢、不重复、按顺序到达
• 像打电话：先拨号，确认对方接听，再说话

数据报套接字（UDP）：

• 直接发送数据，不需要建立连接
• 不保证数据一定到达
• 像发短信：直接发，不管对方有没有收到

6.3 创建 Socket

Python 中提供了 socket 模块用于创建套接字。

import socket

# TCP 套接字：面向连接、可靠传输
tcp_socket = socket.socket(
    family=socket.AF_INET,      # IPv4
    type=socket.SOCK_STREAM     # 流式套接字 = TCP
)

# UDP 套接字：无连接、不可靠但快速
udp_socket = socket.socket(
    family=socket.AF_INET,      # IPv4
    type=socket.SOCK_DGRAM      # 数据报套接字 = UDP
)

常用参数解释：

6.4 Socket 地址：IP + 端口

Socket 地址 = IP 地址 + 端口号

类比：

• IP 地址 = 大楼地址
• 端口号 = 房间号
• Socket 地址 = 大楼地址 + 房间号（唯一标识一个进程）

在代码中，Socket 地址表示为元组(IP, 端口)：

server_address = ("127.0.0.1", 8000)  # 本地 IP，端口 8000

6.5 那些教程不提的 Socket 选项

这部分是实战中最容易踩坑的地方。

6.5.1 SO_REUSEADDR：端口复用

问题场景：服务重启时经常遇到 Address already in use 错误。

原因：TCP 连接断开后，端口会进入 TIME_WAIT 状态（默认 60 秒），期间无法重新绑定。

类比：

• 你刚挂断电话，电话就进入"冷却期"
• 在冷却期内，你不能用同一个号码拨打
• 设置 SO_REUSEADDR 就是告诉系统："冷却期不用等，直接复用"

解决办法：

import socket

tcp_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 必须在 bind 之前设置
tcp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
tcp_socket.bind(('0.0.0.0', 8000))

6.5.2 TCP_NODELAY：禁用 Nagle 算法

问题场景：发送小数据包时，TCP 会缓冲一下再发送，导致延迟。

原因：Nagle 算法——TCP 会把多个小数据包合并成一个大包发送，减少网络开销。

类比：

• 你发快递，快递公司会等攒够一车再发
• 但有时候你急着让对方收到，不等攒车，直接发
• 设置 TCP_NODELAY 就是告诉 TCP："别攒了，直接发"

tcp_socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)

6.5.3 SO_SNDBUF / SO_RCVBUF：缓冲区大小

默认缓冲区大小可能不够，高吞吐场景需要手动调整：

# 设置发送缓冲区 64KB
tcp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 65536)
# 设置接收缓冲区 64KB
tcp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 65536)

7 UDP 编程：简单但不"简陋"

7.1 什么是 UDP

UDP（User Datagram Protocol，用户数据报协议）是一个简单的面向数据报的通信协议。

核心特性：

UDP 适合什么场景？

7.2 UDP 编程流程

服务端                              客户端
  |                                  |
  |--- 1. 创建 UDP 套接字 ---------->|
  |--- 2. 绑定 IP 和端口 ---------->|
  |                                  |
  |<---------- 3. 发送数据 -----------|
  |--- 4. 接收数据 ----------------->|
  |                                  |
  |--- 5. 发送响应 ----------------->|
  |<---------- 6. 接收响应 -----------|
  |                                  |
  |--- 7. 关闭套接字 --------------->|

和 TCP 的区别：

• UDP 服务端不需要listen() 和 accept()
• UDP 客户端不需要connect()
• 直接用 sendto() 发送，用 recvfrom() 接收

7.3 UDP 实战案例

服务端

"""udp_server.py"""
import socket

def udp_server(host='0.0.0.0', port=8000):
    # 创建 UDP 套接字
    udp_socket = socket.socket(family=socket.AF_INET, type=socket.SOCK_DGRAM)
    # 设置端口复用
    udp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    # 绑定 IP 和端口
    # 注意：UDP 不需要 listen()，直接绑定就能收数据
    udp_socket.bind((host, port))
    print(f"UDP 服务端启动，监听 {host}:{port}")

    try:
        while True:
            # 接收数据（最大 1024 字节）
            # recvfrom 返回数据和发送方的地址
            recv_data, client_addr = udp_socket.recvfrom(1024)
            client_ip = client_addr[0]
            client_port = client_addr[1]
            print(f"{client_ip}:{client_port} >> {recv_data.decode('utf-8')}")

            # 发送响应
            # sendto 需要指定对方的地址
            udp_socket.sendto("你好".encode('utf-8'), client_addr)
    except KeyboardInterrupt:
        print("\n服务端关闭")
    finally:
        # 关闭套接字
        udp_socket.close()

if __name__ == '__main__':
    udp_server()

客户端

"""udp_client.py"""
import socket

def udp_client(server_ip='127.0.0.1', server_port=8000):
    # 创建 UDP 套接字
    udp_socket = socket.socket(family=socket.AF_INET, type=socket.SOCK_DGRAM)
    # 设置超时，避免无限等待（UDP 不保证到达，所以要设超时）
    udp_socket.settimeout(5.0)

    try:
        while True:
            try:
                # 发送数据
                msg = input(f"{server_ip}:{server_port} << ")
                udp_socket.sendto(msg.encode('utf-8'), (server_ip, server_port))

                # 接收响应
                recv_data, client_addr = udp_socket.recvfrom(1024)
                client_ip = client_addr[0]
                client_port = client_addr[1]
                print(f"{client_ip}:{client_port} >> {recv_data.decode('utf-8')}")
            except socket.timeout:
                print("请求超时，服务端未响应")
    except KeyboardInterrupt:
        print("\n客户端退出")
    finally:
        # 关闭套接字
        udp_socket.close()

if __name__ == '__main__':
    udp_client()

7.4 UDP 的坑

坑 1：UDP 数据包大小限制

UDP 数据包最大理论值是 65535 字节（含头部），但实际受 MTU（通常 1500 字节）限制。超过 MTU 的数据包会被分片，丢一个分片整个包就废了。

💡 实战建议：单个 UDP 数据包控制在 1400 字节以内，避免分片。

坑 2：UDP 不保证顺序

先发的包可能后到，后发的包可能先到。如果你的应用需要保证顺序，要么加序号，要么换 TCP。

坑 3：UDP 没有拥塞控制

疯狂发数据包可能导致网络拥塞。高吞吐场景需要自己做流量控制。

8 TCP 编程：可靠传输背后的复杂世界

8.1 什么是 TCP

TCP（Transmission Control Protocol，传输控制协议）是一种面向连接的、可靠的、基于字节流的传输层通信协议。

核心特性：

TCP 的三个阶段：

1. 连接建立（三次握手）
2. 数据传送
3. 连接终止（四次挥手）

保证可靠性的机制：

8.2 三次握手与四次挥手

8.2.1 三次握手（建立连接）

目的：确认双方都能正常收发数据。

为什么不能两次？

• 两次握手，服务器无法确认客户端能收到自己的消息
• 比如：A 说"你好"，B 说"你好"，但 A 不知道 B 听到了没有
• 三次是确认双方收发能力的最少次数

过程：

客户端                        服务器
  |                             |
  |------- SYN=1, seq=x ------->|  ← 第一次握手：客户端说"你好，能听到吗？"
  |                             |
  |<---- SYN=1, ACK=1,          |  ← 第二次握手：服务器说"能听到，你能听到我吗？"
  |    seq=y, ack=x+1 ---------|
  |                             |
  |------- ACK=1,               |  ← 第三次握手：客户端说"我也能听到"
  |    seq=x+1, ack=y+1 -----> |
  |                             |
  |====== 连接已建立 ============|

8.2.2 四次挥手（断开连接）

目的：双方都确认没有数据要发送了，才能断开。

为什么不是三次？

• TCP 是全双工通信，每个方向需要单独关闭
• 客户端说"我说完了"，服务器确认，但服务器可能还有话要说
• 等服务器也说"我说完了"，客户端确认，才能完全断开

过程：

客户端                        服务器
  |                             |
  |------- FIN=1, seq=u ------->|  ← 第一次挥手：客户端说"我说完了"
  |                             |
  |<------ ACK=1,               |  ← 第二次挥手：服务器说"好的，我知道了"
  |    seq=v, ack=u+1 ---------|
  |                             |
  |<------ FIN=1,               |  ← 第三次挥手：服务器说"我也说完了"
  |    seq=w, ack=u+1 ---------|
  |                             |
  |------- ACK=1,               |  ← 第四次挥手：客户端说"好的，再见"
  |    seq=u+1, ack=w+1 -----> |
  |                             |
  |====== 连接已断开 ============|

8.3 TCP 编程流程

服务端                              客户端
  |                                  |
  |--- 1. 创建 TCP 套接字 ---------->|
  |--- 2. 绑定 IP 和端口 ---------->|
  |--- 3. 设置监听（listen）------->|   ← 服务器进入"等待连接"状态
  |                                  |
  |<-------- 4. 连接请求 ------------|   ← 客户端发起连接（三次握手）
  |--- 5. 接受连接（accept）------>|   ← 服务器接受连接
  |                                  |
  |<-------- 6. 发送数据 ------------|
  |--- 7. 接收数据 ----------------->|
  |                                  |
  |--- 8. 关闭连接 ---------------->|
  |--- 9. 关闭套接字 --------------->|

关键区别：

• TCP 服务端需要 listen() 进入监听状态
• TCP 客户端需要 connect() 发起连接
• TCP 服务端 accept() 接受连接，返回一个新的 Socket 用于和该客户端通信
• 直接用 send() 和 recv() 传输数据，不需要指定地址

8.4 TCP 实战案例

服务端（支持多客户端）

"""tcp_server.py"""
import socket
import threading

def handle_client(client_socket, client_addr):
    """处理单个客户端连接"""
    print(f"新连接: {client_addr[0]}:{client_addr[1]}")
    try:
        while True:
            # 接收数据
            # recv 返回空数据表示客户端断开连接
            recv_data = client_socket.recv(1024)
            if not recv_data:
                print(f"客户端 {client_addr[0]}:{client_addr[1]} 断开")
                break
            print(f"{client_addr[0]}:{client_addr[1]} >> {recv_data.decode('utf-8')}")

            # 发送响应
            client_socket.send("你好".encode('utf-8'))
    except ConnectionResetError:
        print(f"客户端 {client_addr[0]}:{client_addr[1]} 异常断开")
    except Exception as e:
        print(f"处理客户端 {client_addr[0]}:{client_addr[1]} 时出错: {e}")
    finally:
        client_socket.close()

def tcp_server(host='0.0.0.0', port=8000, max_connections=5):
    # 创建 TCP 套接字
    tcp_socket = socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM)
    # 设置端口复用
    tcp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    # 绑定 IP 和端口
    tcp_socket.bind((host, port))
    # 设置监听（最大等待连接数）
    # listen 让服务器进入"等待连接"状态
    tcp_socket.listen(max_connections)
    print(f"TCP 服务端启动，监听 {host}:{port}")

    try:
        while True:
            # 等待客户端连接
            # accept 返回一个新的 Socket 和客户端地址
            client_sock, client_addr = tcp_socket.accept()
            # 为每个客户端创建独立线程
            thread = threading.Thread(
                target=handle_client,
                args=(client_sock, client_addr),
                daemon=True
            )
            thread.start()
            print(f"活跃连接数: {threading.active_count() - 1}")
    except KeyboardInterrupt:
        print("\n服务端关闭")
    finally:
        # 关闭套接字
        tcp_socket.close()

if __name__ == '__main__':
    tcp_server()

客户端

"""tcp_client.py"""
import socket

def tcp_client(server_ip='127.0.0.1', server_port=8000):
    # 创建 TCP 套接字
    tcp_socket = socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM)
    tcp_socket.settimeout(10.0)

    try:
        # 连接服务器（三次握手在这里发生）
        tcp_socket.connect((server_ip, server_port))
        print(f"已连接到 {server_ip}:{server_port}")

        while True:
            try:
                # 发送数据
                msg = input(f"{server_ip}:{server_port} << ")
                tcp_socket.send(msg.encode('utf-8'))

                # 接收数据
                recv_data = tcp_socket.recv(1024)
                if not recv_data:
                    print("服务器断开连接")
                    break
                print(f"{server_ip}:{server_port} >> {recv_data.decode('utf-8')}")
            except socket.timeout:
                print("接收超时")
    except ConnectionRefusedError:
        print(f"无法连接到 {server_ip}:{server_port}，请确认服务是否启动")
    except KeyboardInterrupt:
        print("\n客户端退出")
    finally:
        # 关闭套接字（四次挥手在这里发生）
        tcp_socket.close()

if __name__ == '__main__':
    tcp_client()

8.5 TCP 粘包与拆包：最大的坑

问题根源：TCP 是面向字节流的协议，没有消息边界。

类比：

• TCP 就像水流——你倒三杯水进水管，另一端可能一次接了 2.5 杯
• UDP 就像乒乓球——你发三个球，对方收到三个球（可能丢了一个）

发送方: [消息A][消息B][消息C]
接收方可能收到: [消息A的前半部分][消息A的后半部分+消息B][消息C]

解决方案 1：长度前缀（最常用）

原理：先发 4 字节表示消息长度，再发消息内容。接收方先读 4 字节知道长度，再读对应长度的数据。

import struct

def send_msg(sock, msg):
    """发送带长度前缀的消息"""
    encoded = msg.encode('utf-8')
    # 4 字节长度前缀（网络字节序）
    length_prefix = struct.pack('!I', len(encoded))
    sock.sendall(length_prefix + encoded)

def recv_msg(sock):
    """接收带长度前缀的消息"""
    length_data = recv_exact(sock, 4)
    if not length_data:
        return None
    msg_length = struct.unpack('!I', length_data)[0]
    msg_data = recv_exact(sock, msg_length)
    return msg_data.decode('utf-8')

def recv_exact(sock, n):
    """精确读取 n 字节"""
    data = bytearray()
    while len(data) < n:
        packet = sock.recv(n - len(data))
        if not packet:
            return None
        data.extend(packet)
    return bytes(data)

这是工业界最常用的方案，Redis、HTTP/2 等协议都采用类似机制。

解决方案 2：固定长度消息

def send_fixed(sock, msg, size=64):
    """发送固定长度消息"""
    encoded = msg.encode('utf-8')[:size].ljust(size, b'\0')
    sock.sendall(encoded)

def recv_fixed(sock, size=64):
    """接收固定长度消息"""
    data = recv_exact(sock, size)
    return data.rstrip(b'\0').decode('utf-8') if data else None

缺点：浪费带宽，消息长度受限。

解决方案 3：分隔符

def send_delimited(sock, msg, delimiter=b'\r\n'):
    """发送带分隔符的消息"""
    sock.sendall(msg.encode('utf-8') + delimiter)

def recv_delimited(sock, delimiter=b'\r\n', buffer_size=4096):
    """接收带分隔符的消息"""
    buffer = bytearray()
    while True:
        data = sock.recv(buffer_size)
        if not data:
            return None
        buffer.extend(data)
        if delimiter in buffer:
            line, _, buffer = buffer.partition(delimiter)
            return line.decode('utf-8')

HTTP 和 FTP 用 \r\n 作为分隔符。

8.6 并发处理多客户端

上面的示例用了多线程，但实际项目中还有更好的方案：

asyncio 示例：

import asyncio

async def handle_client(reader, writer):
    addr = writer.get_extra_info('peername')
    print(f"新连接: {addr}")
    try:
        while True:
            data = await reader.read(1024)
            if not data:
                break
            msg = data.decode('utf-8')
            print(f"收到 {addr}: {msg}")
            writer.write(f"已收到: {msg}".encode('utf-8'))
            await writer.drain()
    except Exception as e:
        print(f"处理 {addr} 时出错: {e}")
    finally:
        writer.close()
        await writer.wait_closed()

async def main():
    server = await asyncio.start_server(handle_client, '0.0.0.0', 8000)
    async with server:
            await server.serve_forever()

asyncio.run(main())

8.7 TCP vs UDP 对比

💡 决策树：需要可靠传输？→ TCP。需要实时性能，能容忍丢包？→ UDP。需要广播/组播？→ UDP。不确定？→ 默认选 TCP。

9 HTTP 协议：应用层的王者

9.1 什么是 HTTP

HTTP（HyperText Transfer Protocol，超文本传输协议）是万维网数据通信的基础。

类比：

• HTTP 就像餐厅的服务员——你点菜（请求），服务员把菜单传给厨房，菜做好了服务员端回来（响应）

设计初衷：提供一种发布和接收 HTML 页面的方法。

工作流程：

1. 客户端发起请求，建立到服务器指定端口（默认 80）的 TCP 连接
2. 服务器在该端口监听客户端请求
3. 一旦收到请求，服务器返回状态码（如 HTTP/1.1 200 OK）和内容

9.2 HTTP 消息结构

9.2.1 客户端请求消息

HTTP 请求由四个部分组成：请求行、请求头、空行、请求体。

GET /index.html HTTP/1.1          ← 请求行：我要用什么方法，访问什么资源
Host: example.com                 ← 请求头：补充信息（访问哪个网站）
User-Agent: Python/3.11                  （客户端是什么）
Accept: application/json                  （支持什么格式）
                                  ← 空行：请求头结束
{"name": "test"}                  ← 请求体：POST/PUT 等带的数据

请求行包含：

• 请求方法：GET、POST、PUT、DELETE 等——我要做什么？
• 请求 URI：资源路径——我要访问什么？
• 协议版本：HTTP/1.1 或 HTTP/2——用什么版本的协议？

常见请求头：

9.2.2 服务端响应消息

HTTP 响应由四个部分组成：状态行、消息报头、空行、响应正文。

HTTP/1.1 200 OK                   ← 状态行：请求成功了吗？
Content-Type: application/json    ← 响应头：补充信息（返回什么格式）
Content-Length: 42                        （返回数据多大）
                                  ← 空行：响应头结束
{"id": 123, "name": "张三"}       ← 响应正文：实际返回的数据

9.3 HTTP 请求方法

HTTP/1.1 定义了八种方法，HTTP 服务器至少应该实现 GET 和 HEAD 方法，其他方法都是可选的。

关键概念：

• 幂等性：多次执行结果相同（GET、PUT、DELETE 幂等，POST 不幂等）- 就像按电梯，按一次和按十次效果一样
• 安全性：方法不会修改服务器状态（GET、HEAD、OPTIONS、TRACE 安全）- 就像看菜单，不会把菜单看坏

⚠️ 注意：GET 方法只应用于读取，不应产生"副作用"。因为 GET 可能被网络爬虫随意访问，如果 GET 请求会修改数据，后果很严重。

9.4 HTTP 状态码

状态码分为五类：

9.4.1 1xx（信息性状态码）

表示请求正在处理。

9.4.2 2xx（成功状态码）

表示请求正常处理完毕。

9.4.3 3xx（重定向状态码）

表示需要后续操作才能完成请求。

9.4.4 4xx（客户端错误状态码）

表示请求包含语法错误或无法完成。

9.4.5 5xx（服务器错误状态码）

表示服务器处理请求时发生错误。

💡 记忆技巧：2xx 成功、3xx 重定向、4xx 客户端错、5xx 服务器错。

9.5 HTTPS：HTTP 的安全版

HTTPS = HTTP + TLS/SSL。

类比：

• HTTP 就像寄明信片——任何人都能看到内容
• HTTPS 就像寄挂号信——内容加密，只有收件人能看

核心概念：

• 对称加密：加密解密用同一个密钥，速度快，但密钥分发困难
• 非对称加密：公钥加密、私钥解密，速度慢，但解决了密钥分发问题
• TLS 握手：先用非对称加密交换对称密钥，再用对称加密传输数据

10 实战：发送 HTTP 请求

10.1 基础请求

requests 是 Python 中最常用的 HTTP 库。

import requests

# 一言网的 API 地址
url = 'https://v1.hitokoto.cn/'

# 请求参数
params = {
'c': 'a',        # 类型：a 动画、b 漫画、c 游戏...
'encode': 'json'
}

try:
    print(f"正在发送 GET 请求到: {url}，参数: {params}")
    response = requests.get(url, params=params)
    status_code = response.status_code

    if status_code == 200:
        print(f"请求成功！状态码: {status_code}")
        data = response.json()
        hitokoto = data['hitokoto']
        from_who = data['from_who'] if data['from_who'] else'未知'
        print(f"随机名言: {hitokoto} - {from_who}")
    elif status_code == 404:
        print(f"请求的资源未找到！状态码: {status_code}")
    elif status_code == 500:
        print(f"服务器内部错误！状态码: {status_code}")
    else:
        print(f"发生未知错误，状态码: {status_code}")
except requests.RequestException as e:
    print(f"请求过程中出现错误: {e}")

10.2 生产级请求封装

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_session(
    retries=3,
    backoff_factor=0.3,
    status_forcelist=(500, 502, 503, 504),
    timeout=10
):
    """创建带重试机制的 requests Session"""
    session = requests.Session()

    # 配置重试策略
    retry_strategy = Retry(
        total=retries,
        backoff_factor=backoff_factor,
        status_forcelist=status_forcelist,
        allowed_methods=["GET", "POST", "PUT", "DELETE"]
    )

    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("http://", adapter)
    session.mount("https://", adapter)

    # 设置默认请求头
    session.headers.update({
'User-Agent': 'MyApp/1.0',
'Accept': 'application/json'
    })

    return session

# 使用示例
session = create_session()
try:
    response = session.get(
'https://v1.hitokoto.cn/',
        params={'c': 'a', 'encode': 'json'},
        timeout=10
    )
    response.raise_for_status()
    data = response.json()
    print(f"名言: {data['hitokoto']} —— {data.get('from_who', '未知')}")
except requests.exceptions.RequestException as e:
    print(f"请求失败: {e}")

10.3 一言 API 封装（面向对象版）

import requests
from typing import Optional, Dict

class HitokotoClient:
    """一言网 API 客户端"""

    BASE_URL = 'https://v1.hitokoto.cn/'

    TYPES = {
        'a': '动画', 'b': '漫画', 'c': '游戏',
        'd': '文学', 'e': '原创', 'f': '来自网络',
        'g': '其他', 'h': '影视', 'i': '诗词',
        'j': '网易云', 'k': '哲学', 'l': '抖机灵'
    }

    def __init__(self, timeout: int = 10):
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': 'HitokotoPythonClient/1.0',
            'Accept': 'application/json'
        })
        self.timeout = timeout

    def get_hitokoto(self, type_code: str = 'a') -> Optional[Dict]:
        """获取随机一言"""
        params = {'c': type_code, 'encode': 'json'}
        try:
            response = self.session.get(
                self.BASE_URL, 
                params=params, 
                timeout=self.timeout
            )
            response.raise_for_status()
            data = response.json()
            return {
                'id': data['id'],
                'content': data['hitokoto'],
                'author': data.get('from_who') or'未知',
                'type': self.TYPES.get(type_code, '未知'),
                'source': data.get('from', '')
            }
        except requests.RequestException as e:
            print(f"获取一言失败: {e}")
            return None

    def close(self):
        self.session.close()

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.close()

# 使用示例
with HitokotoClient() as client:
    result = client.get_hitokoto('i')
    if result:
        print(f"「{result['content']}」")
        print(f"—— {result['author']}（{result['type']}）")

11 实战：用 Starlette 构建 Web API

11.1 为什么选 Starlette？

Starlette 是一个轻量级的 Python 异步 Web 框架，专为高性能异步应用设计。

类比：

• Django 就像全家桶餐厅——什么都有（ORM、模板、后台），但可能用不上
• Flask 就像快餐店——简单快速，但复杂功能要自己装插件
• Starlette 就像精品咖啡馆——只做核心功能，但性能极高

特点：

• 轻量：核心功能简洁，没有 ORM、模板等"全家桶"
• 异步：基于 asyncio，适合 IO 密集型场景
• 灵活：可以只用它的一部分，也可以组合其他库
• 高性能：Uvicorn + uvloop，性能接近 Go 的 Web 框架

💡 关系：FastAPI 就是基于 Starlette 构建的，额外加入了数据验证（Pydantic）和自动文档功能。

11.2 技术栈说明

11.3 安装依赖

pip install starlette uvicorn requests

11.4 基础版代码

"""hitokoto_api.py"""
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route
import requests
import uvicorn

HITOKOTO_URL = 'https://v1.hitokoto.cn/'

async def get_hitokoto():
    """获取随机名言"""
    try: 
        params = {'c': 'a', 'encode': 'json'}
        response = requests.get(HITOKOTO_URL, params=params)
        status_code = response.status_code

        if status_code == 200:
            data = response.json()
            hitokoto = data['hitokoto']
            from_who = data['from_who'] if data['from_who'] else'未知'
            return {'hitokoto': hitokoto, 'from_who': from_who}
        else:
            return {'error': f'请求一言网 API 失败，状态码: {status_code}'}
    except requests.RequestException as e:
        return {'error': f'请求过程中出现错误: {str(e)}'}

async def homepage(request):
    """处理根路径请求"""
    result = await get_hitokoto()
    return JSONResponse(result)

app = Starlette(debug=True, routes=[
    Route('/', homepage),
])

if __name__ == "__main__":
    uvicorn.run(app, host='0.0.0.0', port=8000)

11.5 进阶版：异步 HTTP 客户端 + CORS

"""hitokoto_api_advanced.py"""
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route
from starlette.middleware.cors import CORSMiddleware
import httpx
import uvicorn

HITOKOTO_URL = 'https://v1.hitokoto.cn/'

TYPES = {
'a': '动画', 'b': '漫画', 'c': '游戏',
'd': '文学', 'e': '原创', 'f': '来自网络',
'g': '其他', 'h': '影视', 'i': '诗词',
'j': '网易云', 'k': '哲学', 'l': '抖机灵'
}

async def get_hitokoto(type_code: str = 'a'):
    """异步获取随机名言"""
    async with httpx.AsyncClient(timeout=10.0) as client:
        try:
            response = await client.get(HITOKOTO_URL, params={
                'c': type_code,
                'encode': 'json'
            })
            response.raise_for_status()
            data = response.json()
            return {
                'success': True,
                'data': {
                'id': data['id'],
                'content': data['hitokoto'],
                'author': data.get('from_who') or'未知',
                'type': TYPES.get(type_code, '未知'),
                'source': data.get('from', '')
                }
            }
        except httpx.HTTPError as e:
            return {'success': False, 'error': f'请求失败: {str(e)}'}

async def homepage(request):
    """根路径：返回 API 说明"""
    return JSONResponse({
        'message': '一言 API 服务',
        'endpoints': {
            '/': 'API 说明',
            '/hitokoto': '获取随机一言（默认动画类）',
            '/hitokoto/{type}': '获取指定类型的一言',
        },
        'types': TYPES
    })

async def random_hitokoto(request):
    """获取随机一言"""
    return JSONResponse(await get_hitokoto())

async def typed_hitokoto(request):
    """获取指定类型的一言"""
    type_code = request.path_params.get('type', 'a')
    return JSONResponse(await get_hitokoto(type_code))

app = Starlette(
    debug=True,
    routes=[
        Route('/', homepage),
        Route('/hitokoto', random_hitokoto),
        Route('/hitokoto/{type}', typed_hitokoto),
    ]
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=['*'],
    allow_methods=['*'],
    allow_headers=['*'],
)

if __name__ == '__main__':
    uvicorn.run(app, host='0.0.0.0', port=8000)

11.6 运行与测试

pip install starlette uvicorn httpx
python hitokoto_api_advanced.py
curl http://localhost:8000/
curl http://localhost:8000/hitokoto
curl http://localhost:8000/hitokoto/i

11.7 代码说明

⚠️ 注意：虽然 Starlette 支持异步，但 requests 库是同步的。在异步环境中使用 requests 会阻塞事件循环，丧失异步优势。应该用 httpx.AsyncClient 或 aiohttp。

12 网络调试工具箱

写网络程序不会调试等于盲人摸象。这些工具你必须掌握：

12.1 curl

12.1.1 作用

curl 是一个命令行工具，用来发送 HTTP/HTTPS 请求。它可以：

• 测试 API 接口是否正常
• 查看请求和响应的详细信息
• 模拟浏览器发送请求
• 上传/下载文件

类比：curl 就像"命令行版的浏览器"——你输入网址，它帮你访问并显示结果。

12.1.2 使用方式

# 发送 GET 请求（-v 显示详细信息，包括请求头和响应头）
curl -v https://v1.hitokoto.cn/?c=a

# 发送 POST 请求（-X 指定方法，-H 添加请求头，-d 添加请求体）
curl -X POST https://httpbin.org/post \
  -H "Content-Type: application/json" \
  -d '{"name": "Python"}'

# 只查看响应头（-I）
curl -I https://example.com

# 带超时和重试（--connect-timeout 连接超时，--max-time 总超时，--retry 重试次数）
curl --connect-timeout 10 --max-time 30 --retry 3 https://example.com

# 保存响应到文件（-o）
curl -o output.html https://example.com

# 跟随重定向（-L）
curl -L https://example.com

# 添加请求头（-H）
curl -H "Authorization: Bearer token123" https://api.example.com

# 上传文件（-F）
curl -F "file=@/path/to/file.txt" https://example.com/upload

12.1.3 官网及下载

• 官网：https://curl.se/
• 下载：https://curl.se/download.html
• Windows：直接下载 .zip 包，解压后添加到系统 PATH
• macOS：brew install curl
• Linux：sudo apt install curl 或 sudo yum install curl

12.2 netstat / ss

12.2.1 作用

netstat 和 ss 用来查看网络连接状态、端口占用、路由表等信息。

区别：

• netstat：老牌工具，所有系统都支持
• ss：新一代工具，速度更快，推荐 Linux 使用

类比：netstat/ss 就像"网络监控摄像头"——可以看到谁在连接、哪个端口被占用。

12.2.2 使用方式

# 查看所有监听端口（-t TCP，-l 监听，-n 数字显示，-p 显示进程）
netstat -tlnp

# 查看所有连接状态
netstat -an

# 查看 ESTABLISHED 状态的连接
netstat -an | grep ESTABLISHED

# 查看端口 8000 的占用情况
netstat -tlnp | grep 8000

# 查看路由表
netstat -r

# 查看网络统计信息
netstat -s

ss 命令（Linux 推荐）：

# 查看所有监听端口
ss -tlnp

# 查看所有连接
ss -an

# 查看端口 8000 的占用
ss -tlnp | grep 8000

# 查看连接统计
ss -s

# 查看 TIME_WAIT 状态的连接数
ss -o state time-wait | wc -l

12.2.3 官网及下载

• netstat：系统自带，无需安装
• ss：iproute2 包的一部分，Linux 自带
• 文档：man netstat 或 man ss

12.3 Wireshark

12.3.1 作用

Wireshark 是一个图形化抓包工具，可以捕获和分析网络数据包。

功能：

• 实时捕获网络流量
• 解析各种协议（HTTP、TCP、UDP、DNS 等）
• 过滤特定流量
• 查看数据包详细内容

类比：Wireshark 就像"网络显微镜"——可以看到网络中传输的每一个数据包。

12.3.2 使用方式

启动和抓包：

1. 打开 Wireshark
2. 选择要监听的网络接口（如以太网、WiFi）
3. 点击"Start"开始抓包
4. 在浏览器或程序中发送请求
5. 回到 Wireshark 查看捕获的数据包

常用过滤表达式：

# 过滤特定端口
tcp.port == 8000

# 只看 HTTP 协议
http

# 过滤特定 IP
ip.addr == 192.168.1.100

# 过滤源 IP
ip.src == 192.168.1.100

# 只看 SYN 包（TCP 握手）
tcp.flags.syn == 1

# 只看 HTTP GET 请求
http.request.method == "GET"

# 过滤特定域名
http.host == "example.com"

# 组合过滤
tcp.port == 8000 && http

查看 TCP 三次握手：

1. 过滤 tcp.flags.syn == 1
2. 可以看到三个数据包：SYN → SYN-ACK → ACK

12.3.3 官网及下载

• 官网：https://www.wireshark.org/
• 下载：https://www.wireshark.org/download.html
• Windows/macOS：下载安装包直接安装
• Linux：sudo apt install wireshark 或 sudo yum install wireshark

12.4 tcpdump

12.4.1 作用

tcpdump 是命令行抓包工具，功能类似 Wireshark，但没有图形界面。

适用场景：

• 服务器环境（通常没有图形界面）
• 需要脚本自动化抓包
• 轻量级抓包需求

类比：tcpdump 就像"命令行版的 Wireshark"——功能相同，但用命令行操作。

12.4.2 使用方式

# 抓取 8000 端口的流量，保存到文件
tcpdump -i any port 8000 -w capture.pcap

# 查看实时流量（ASCII 显示）
tcpdump -i any port 8000 -A

# 抓取 HTTP 请求和响应（显示内容）
tcpdump -i any port 8000 -A -s 0 'tcp'

# 抓取 100 个数据包后停止
tcpdump -i any -c 100

# 过滤特定主机
tcpdump -i any host 192.168.1.100

# 过滤特定端口范围
tcpdump -i any portrange 8000-9000

# 读取 pcap 文件
tcpdump -r capture.pcap

# 显示更详细的信息（-vvv 最详细）
tcpdump -i any port 8000 -vvv

常用参数：

• -i any：监听所有接口
• -w file.pcap：保存到文件
• -r file.pcap：读取文件
• -A：ASCII 显示
• -s 0：捕获完整数据包（不截断）
• -c N：抓 N 个包后停止
• -n：不解析域名（直接显示 IP）

12.4.3 官网及下载