Python 3 进阶速查

版本：示例以 Python 3.13 为基线，3.14 的关键差异单独标注。每个 Python 代码块可独立运行；进程池示例需保存为 .py 文件。

基础语法、容器和文件操作见 Python 3 基础速查。这里重点复习语言机制、并发选型及网络通信边界，不展开框架部署和完整服务器实现。

1. 闭包：保留外层变量

函数也是对象，可以作为参数、返回值或回调。闭包让内部函数继续访问外层函数的变量，即使外层调用已经结束。

• 读取外层变量：通常不需要额外声明。
• 重新绑定外层变量：使用 nonlocal，目标必须是外层函数中已有的绑定。
• 返回内部函数：是保留闭包的常见方式，不是闭包定义的必要条件。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

# 模块级变量，与 make_counter() 内部的 count 是两个不同的绑定。
count = 100


def make_counter(start=0):
    count = start

    def increment():
        # 更新外层 make_counter() 的 count，而不是模块级的 count。
        nonlocal count
        count += 1
        return count

    return increment


counter = make_counter(10)
print(counter())
print(counter())
print(count)

# 输出：
# 11
# 12
# 100

nonlocal 与 global 改的是哪一个变量？

这里需要让计数器从 10 累加，因此使用 nonlocal，模块级的 count 仍为 100。

• 不声明：本例的 count += 1 会被当作局部赋值，但右侧又先读取这个尚未赋值的局部变量，因此触发 UnboundLocalError。
• 换成 global count：会修改模块级变量，不能用它代替 nonlocal 更新外层函数的状态。

nonlocal 在 Python 3.8～3.10 中同样适用，并非 3.13 的新写法。见 作用域声明。

循环中的闭包为什么容易出错？

先分清：这里的列表装的是三个函数，不是三个计算结果。 列表推导式创建函数时，并没有执行 lambda 的函数体。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

functions = [lambda: i for i in range(3)]
print([function() for function in functions])

# 默认参数在创建函数时求值，用它保存每轮的整数值。
functions = [lambda i=i: i for i in range(3)]
print([function() for function in functions])

# 输出：
# [2, 2, 2]
# [0, 1, 2]

第一种：lambda: i，调用时再读取外层变量。

1. for i in range(3) 让 i 依次取 0、1、2，每轮创建一个无参数函数。
2. lambda: i 表示“调用这个函数时，返回外层的 i”，不是“现在立即返回 i”。
3. 循环结束后才执行 function()；三个函数读取同一个外层变量，它此时是 2，所以得到 [2, 2, 2]。

第二种：lambda i=i: i，用默认参数保存每轮的值。

• = 左边的 i 是函数参数名；右边的 i 是创建函数时读取的当前循环值。
• 三个函数的默认参数分别是 0、1、2；冒号后的 i 返回各自的参数值。
• function() 不传参数，就使用保存的默认值，因此得到 [0, 1, 2]。

回调需要“记住当时的值”时，要显式绑定，不能只看函数写在哪一轮循环里。 见 官方闭包常见问题。

2. 装饰器：包装调用，而不改函数正文

@decorator 可以理解为：函数定义完成后，执行 function = decorator(function)。

装饰器接收被装饰对象，并用返回值重新绑定原名称。它不一定是闭包，也可以由类或可调用实例实现；日常先掌握函数写法即可。见 函数定义与装饰器。

2.1 参数、返回值与元数据都要保留

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

from functools import wraps


def trace(function):
    @wraps(function)
    def wrapper(*args, **kwargs):
        print(f"调用 {function.__name__}")
        return function(*args, **kwargs)

    return wrapper


@trace
def add(left, right=0):
    """返回两个数的和。"""
    return left + right


print(add(2, right=3))
print(add.__name__)
print(add.__doc__)

# 输出：
# 调用 add
# 5
# add
# 返回两个数的和。

• *args, **kwargs：接收并转交位置参数和关键字参数。
• return：把原函数的结果交还调用方；漏写会得到 None。
• @wraps(function)：保留名称、文档等元数据，并通过 __wrapped__ 提供原函数引用；不会自动保证包装逻辑正确。见 functools.wraps。

如果只删除上例中的 @wraps(function)，其余代码不变，输出会变成：

1
2
3
4

# 调用 add
# 5
# wrapper
# None

装饰后，名称 add 实际绑定到 wrapper。不使用 wraps 时，add.__name__ 和 add.__doc__ 展示的是包装函数自己的信息；本例的 wrapper 没写文档字符串，所以文档值是 None。

加法结果仍为 5：wraps 保留的是元数据，不是负责计算的部分。

这个例子面向普通同步函数。若要在异步函数实际执行前后添加逻辑，应使用 async def 包装器并 await 原函数。

2.2 带参数的装饰器与叠加顺序

@label("外层") 先调用 label() 得到装饰器，再把目标函数传给它。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

from functools import wraps


def label(name):
    def decorate(function):
        @wraps(function)
        def wrapper(*args, **kwargs):
            print(f"进入{name}")
            result = function(*args, **kwargs)
            print(f"离开{name}")
            return result

        return wrapper

    return decorate


@label("外层")
@label("内层")
def greet():
    return "你好"


print(greet())

# 输出：
# 进入外层
# 进入内层
# 离开内层
# 离开外层
# 你好

装饰时由内向外组合，相当于 label("外层")(label("内层")(greet))。

调用时的顺序由包装器决定：本例先进入外层，返回时先离开内层。不要把“装饰顺序”和“函数调用顺序”混为一谈。

3. property：让属性访问带上逻辑

property 适合计算属性或需要校验的赋值。普通字段不需要为了形式完整而再写一套 getter/setter。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

class Stock:
    def __init__(self, quantity):
        self.quantity = quantity

    @property
    def quantity(self):
        return self._quantity

    @quantity.setter
    def quantity(self, value):
        if value < 0:
            raise ValueError("库存不能为负数")
        self._quantity = value


stock = Stock(5)
stock.quantity = 3
print(stock.quantity)

try:
    stock.quantity = -1
except ValueError as error:
    print(f"ValueError: {error}")

# 输出：
# 3
# ValueError: 库存不能为负数

• 读取 stock.quantity：调用被 @property 修饰的方法，不加括号。
• 赋值 stock.quantity = 3：调用对应 setter；本例初始化也走同一条校验路径。
• 只有 getter：不能通过该属性赋值，但不代表整个对象不可变。

_quantity 是内部存储的命名约定，不是访问权限机制。上例仅演示非负校验，不代表已经验证所有输入类型。见 property。

4. 上下文管理器：明确资源何时释放

with 的能力取决于对象实现的协议，不是所有 with 都表示“关闭文件”。

__enter__() 成功后，正常退出或因异常离开代码块都会调用 __exit__()。不要无意中返回 True 吞掉异常。 见 官方 with 语句说明。

用 contextmanager 简化成对操作

@contextmanager 用于基于生成器的上下文管理，不是给任意函数加上它就能配合 with 使用。

• 被装饰函数调用后需要返回生成器迭代器；通常写成下面这种含 yield 的生成器函数。
• 普通只 return 一个字符串、列表等值的函数，不能直接套用这个装饰器。
• 正常进入并退出上下文时，生成器必须恰好 yield 一次，不能零次或多次。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

from contextlib import contextmanager
from io import StringIO


@contextmanager
def text_reader(text):
    # StringIO 是内存中的文本流，本例不创建磁盘文件。
    reader = StringIO(text)
    try:
        yield reader
    finally:
        reader.close()


with text_reader("hello") as reader:
    print(reader.read())

print(reader.closed)

# 输出：
# hello
# True

按执行顺序看这个例子：

1. 进入 with：执行 yield 前的代码，创建文本流。
2. yield reader：把流交给 as reader，暂停生成器，执行 with 中的读取操作。
3. 退出 with：恢复生成器，执行 finally，关闭文本流，所以最后输出 True。

这是协议演示；实际只使用 StringIO 时，可以直接 with StringIO(...) as reader。

如果捕获块内异常只是为了记录，应再次抛出，而不是静默结束。 见 contextlib.contextmanager。

5. 生成器：按需组合数据

生成器适合逐项处理，避免一次性构造全部中间结果。它不保证每种任务都更快，也不会自动缓存已产生的数据。

5.1 逐步读取，耗尽后不会重来

下面的输入有三组：[1, 2]、[3]、[]。这个函数按组依次取出数字，只展开一层。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

def flatten(groups):
    for group in groups:
        # 依次产出当前 group 的元素；空列表没有元素可产出。
        yield from group


numbers = flatten([[1, 2], [3], []])
# 只取第一个数字，生成器暂停在这里。
print(next(numbers))
# 从暂停处继续，将剩余数字收集成列表。
print(list(numbers))
# 已经读完，不会自动回到开头。
print(list(numbers))

# 输出：
# 1
# [2, 3]
# []

1. flatten(...) 创建生成器；next(numbers) 开始执行，从第一组取出 1。
2. 第一次 list(numbers) 继续取出第一组剩余的 2，再取第二组的 3；第三组为空，所以得到 [2, 3]。
3. 第二次 list(numbers) 时已经耗尽，得到 []。如需重新遍历，要重新调用 flatten(...) 创建生成器。

这里先理解“读取位置会向前推进”。yield 与 yield from 的完整差异留给后续易混淆点文章，不在此展开。见 生成器官方说明。

5.2 数据流与资源生命周期分开

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

from io import StringIO


def nonempty_lines(lines):
    for line in lines:
        text = line.strip()
        if text:
            yield text


with StringIO("第一行\n\n第二行\n") as source:
    for text in nonempty_lines(source):
        print(text)
        break

print(source.closed)

# 输出：
# 第一行
# True

本例由调用方的 with 管理文本流，生成器只处理数据；换成 open(..., encoding="utf-8") 时也是这个思路。

循环 break 不等于关闭生成器内部持有的文件。 如果资源是在生成器内部创建的，需要明确关闭生成器或安排上下文管理，不要依赖垃圾回收时机。

6. 赋值、浅拷贝与深拷贝

先看结构：original = [[1, 2], [3]] 是一个外层列表，里面装着两个子列表。

浅拷贝也会创建新列表。区别在于：里面的子列表是否仍与原对象共享。

下面用 import copy 导入模块，copy.copy() 是浅拷贝，copy.deepcopy() 是深拷贝；对于本例的普通列表，original.copy() 也能做浅拷贝。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

import copy

original = [[1, 2], [3]]
alias = original
shallow = copy.copy(original)
deep = copy.deepcopy(original)

# shallow[0] 与 original[0] 是同一个子列表，两边都能看到新增的 9。
shallow[0].append(9)
# 只给 shallow 的外层列表加一项，不会给 original 的外层加项。
shallow.append([4])

print(alias is original)
print(shallow is original)
print(shallow[0] is original[0])
print(deep[0] is original[0])
print(original)
print(shallow)
print(deep)

# 输出：
# True
# False
# True
# False
# [[1, 2, 9], [3]]
# [[1, 2, 9], [3], [4]]
# [[1, 2], [3]]

对应上面的两次修改：

• shallow[0].append(9) 修改共享的子列表，所以 original 和 shallow 都出现 9。
• shallow.append([4]) 修改的是新的外层列表，所以只有 shallow 多出 [4]。
• deep 的外层列表和子列表都与原列表分开，因此仍保持 [[1, 2], [3]]。

把深拷贝理解为“为这组嵌套列表建立独立副本”是合适的；但不能只概括成“重新申请一块内存”，否则无法区分浅拷贝，也容易误以为所有内部对象都必然重新创建。

是否深拷贝，取决于是否允许共享，而不是“有嵌套就必须复制”。 深拷贝有额外成本，也不保证每个对象都换成新对象；文件、socket 等资源不能靠它复制。见 copy。

Python 3.13：只替换部分字段

copy.replace() 适用于数据类、命名元组和实现 __replace__() 的类型；创建同类型的新对象，替换指定字段。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

from copy import replace
from dataclasses import FrozenInstanceError, dataclass


@dataclass(frozen=True)
class Config:
    host: str
    port: int


original = Config("localhost", 8000)
try:
    # frozen=True 禁止修改这个已有实例的字段。
    original.port = 9000
except FrozenInstanceError as error:
    print(f"FrozenInstanceError: {error}")

# 创建修改了 port 的新实例，原实例保持不变。
updated = replace(original, port=9000)
print(original)
print(updated)

# 输出：
# FrozenInstanceError: cannot assign to field 'port'
# Config(host='localhost', port=8000)
# Config(host='localhost', port=9000)

frozen=True 阻止常规字段赋值；replace() 则创建新实例。字段替换不是深拷贝，也不是任意对象都支持。 见 copy.replace。

dataclass 带不带括号有什么区别？

不传选项时，有无括号不会改变行为；真正影响能否赋值的是 frozen。 这属于 dataclass 提供的两种用法，不能推广为所有装饰器都能省略括号。见 dataclass 的等价写法。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

from dataclasses import dataclass


@dataclass
class Config1:
    host: str
    port: int


# 实例化新定义的 Config1，不要误用前面冻结的 Config。
original1 = Config1("localhost", 8000)
original1.port = 9000
print(original1)

# 输出：Config1(host='localhost', port=9000)

如果定义的是 Config1，却写 Config(...)，创建的仍是 Config 的实例：

• 前面保留了冻结的 Config：修改字段会得到 FrozenInstanceError。
• 没有定义过 Config：实例化时就会得到 NameError，还没执行到修改字段。

因此要先核对实际实例化的类名。上例的 @dataclass 换成 @dataclass() 也能正常赋值；这些写法在 Python 3.8～3.10 中已经可用，copy.replace() 才是这里的 3.13 新 API。

7. 正则表达式：选择正确的匹配范围

7.1 先选接口，再写模式

模式通常写成 原始字符串 r"..."，减少 Python 字符串转义与正则转义的冲突。重复使用时可用 re.compile() 保存模式。见 re 官方文档。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

import re

# 自定义账号规则：小写字母开头，其后允许小写字母、ASCII 数字和下划线；总长 3～16。
account_pattern = re.compile(r"[a-z][a-z0-9_]{2,15}")
print(account_pattern.fullmatch("user_01") is not None)
print(account_pattern.fullmatch("user_01!") is not None)

match = re.search(r"order=(?P<number>[0-9]+)", "user=ada order=2048")
if match:
    print(match.group(0))
    print(match.group("number"))
    print(type(match.group("number")).__name__)

# 输出：
# True
# False
# order=2048
# 2048
# str

group("number") 为什么得到 2048？

1. order= 匹配这段固定文字。
2. [0-9]+ 匹配一个或多个 ASCII 数字，本例匹配到 2048。
3. (?P<number>...) 把括号内的匹配内容保存为名叫 number 的组。

group() 是匹配对象的方法。group(0) 取完整匹配，即 order=2048；group("number") 只取命名组中的数字文本，即字符串 "2048"。这里的 "number" 是组名，不是 Python 变量。

打印时字符串不会自动显示引号，返回值并不是整数。 后续要计算时再用 int() 转换。见 Match.group。

匹配失败返回 None，不能直接调用 .group()。

7.2 常用语法与边界

a|b 表示择一匹配；例如 cat|dog 可以匹配其中任意一项。

• Unicode：str 模式默认按 Unicode 处理；\d 不只匹配 0–9，\s 不只空格和 Tab，\w 包含 Unicode 字母数字与下划线。
• ASCII 数字：明确写 [0-9]；需要整体采用 ASCII 字符类规则时使用 re.ASCII。
• 整串校验：优先 fullmatch()；$ 还能匹配字符串末尾换行之前的位置，不能简单当成“绝对末尾”。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

import re

print(re.fullmatch(r"\d+", "１２３") is not None)
print(re.fullmatch(r"[0-9]+", "１２３") is not None)
print(re.sub(r"\s+", " ", "hello\t  world"))

# 输出：
# True
# False
# hello world

正则适合局部文本规则，不适合代替 HTML、JSON 等格式的解析器。 对不受信任的长文本，还要避免可能导致大量回溯的复杂模式。

8. 并发选型与 GIL

并发是多个任务在一段时间内交替推进；并行是同一时刻执行多个任务。两者不能按“任务数是否超过 CPU 核心数”来定义。

并发不是免费的加速开关。 小任务可能因调度成本变慢；选择方式前先确认瓶颈。线程池与进程池接口见 concurrent.futures。

从 3.8～3.12 到 3.13 以后，GIL 有什么变化？

• Python 3.8～3.12 的常规 CPython：已经有 GIL。同一解释器内，同一时刻通常只有一个线程执行 Python 字节码，因此纯 Python 重计算通常不能靠多线程利用多核提速。
• Python 3.13：常规构建仍默认启用 GIL；另提供实验性的 free-threaded 构建，可以在禁用 GIL 时并行运行 Python 线程。
• Python 3.14：free-threaded 正式支持，但仍是可选构建；第三方扩展兼容性仍需检查。

旧版本也能使用多线程处理 I/O 等待，某些释放 GIL 的扩展还可以并行计算。GIL 不是“Python 没有多线程”，也不是“线程永远不能使用多核”。 见 Python 3.12 的线程说明。

“升级 Python”不等于“默认关闭 GIL”；有无 GIL 都不能代替共享数据的业务同步。 见 free-threading 指南及 3.14 支持状态。

9. 线程：先启动，再等待；共享更新要加锁

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

from threading import Lock, Thread

count = 0
lock = Lock()


def increment(times):
    global count
    for _ in range(times):
        # 两个线程必须使用同一把锁，保护完整的“读取、加一、写回”。
        with lock:
            count += 1


threads = [Thread(target=increment, args=(1000,)) for _ in range(2)]
for thread in threads:
    thread.start()

for thread in threads:
    thread.join()

print(count)

# 输出：2000

• target=increment：传函数对象，不是 increment() 的执行结果。
• args=(1000,)：一个元素的元组需要逗号；本例也可用 kwargs={"times": 1000} 传入命名参数。
• start()：启动线程；直接调用 .run() 不会创建新线程。
• join()：阻塞当前调用方，等待指定线程结束。本例已先启动两个线程，不会把它们变成串行任务。

不要用 sleep() 猜测另一个线程“应该做完了”。 等待完成用 join()，传递停止信号可用 Event，在线程间交接任务可用 queue.Queue。

with lock: 能在代码块结束时释放锁，避免漏写 release()；多把锁仍需统一获取顺序，防止死锁。上述行为见 threading。

daemon 线程可能在程序退出时被直接停止，不适合承担必须完成的保存或清理任务。 长期运行的工作线程应通过停止信号自行结束，再由主线程等待；有限任务直接等待完成即可。

10. 进程池：隔离内存，返回结果

进程各有独立的 Python 对象空间，修改子进程的普通全局变量，不会自动同步到父进程。优先通过任务参数与返回值交换数据；需要通信时再考虑队列等工具。

下面仅演示进程池接口，平方运算太小，实际通常不值得启动进程。保存为 process_demo.py 后运行：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

from concurrent.futures import ProcessPoolExecutor
from multiprocessing import get_context


def square(number):
    return number ** 2


if __name__ == "__main__":
    # 显式使用 spawn：启动新解释器，让示例跨平台行为更容易理解。
    context = get_context("spawn")
    with ProcessPoolExecutor(max_workers=2, mp_context=context) as executor:
        results = list(executor.map(square, [2, 3, 4]))
    print(results)

# 输出：[4, 9, 16]

不指定 spawn，为什么输出还是 [4, 9, 16]？

• 计算逻辑没变：任务只使用传入的数字计算平方，不依赖父进程的运行时状态；启动方式不同，正常执行的结果也应相同。
• 启动机制可能也没变：macOS 从 Python 3.8 起默认就是 spawn；未另行配置时，在 Mac 上省略显式设置仍会用它。
• 显式设置的作用：get_context("spawn") 选择启动策略，mp_context=context 把策略交给这个池，而不是改变任务算法。

若要对比默认方式，应同时去掉 context 定义和构造进程池时的 mp_context=context。只删定义、却仍引用 context，会得到 NameError。

默认方式受平台、版本和程序配置影响，见下表；显式使用 spawn 是本例的跨平台选择，不是每个项目都必须这样写。macOS 的变化见 Python 3.8 更新说明。

为什么要这样组织？

• 工作函数放在模块顶层，不要用 lambda 或局部函数作为进程池任务。
• 入口保护避免新解释器导入模块时再次创建进程池；交互式环境不能照搬这个脚本示例。
• 参数、返回值及任务函数需要满足可序列化、可导入的要求；不要传打开的文件或锁来“共享一切”。

executor.map() 按输入顺序返回结果，不代表任务按此顺序完成。

需要逐个提交时，submit() 返回表示任务及其结果的 Future 对象；.result() 会等待结果，并在任务失败时重新抛出异常。见 进程池文档。

进程启动方式的版本变化

fork 从当前进程派生；spawn 启动新解释器；forkserver 通过专用服务器进程派生工作进程。不要把“创建进程就是复制主进程变量”当作通用规则。 见 3.13 启动方式和 3.14 迁移说明。

terminate() 可能跳过清理，并影响正在使用的锁或队列；不要把强制结束当成常规退出方案。本例使用 ProcessPoolExecutor 的上下文管理，退出时会等待已提交任务完成。

11. asyncio：等待时让出执行机会

asyncio 通常在一个事件循环线程中协作调度任务；适合异步 I/O，不会把普通计算自动分配到多个 CPU 核心。

• async def：定义协程函数；调用后得到协程对象，不会因此自动执行完。
• await：等待可等待对象；发生挂起时，事件循环才能推进其他任务。
• TaskGroup：Python 3.11 起提供的任务组，管理一组并发任务的生命周期。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

import asyncio


async def fetch_label(name):
    # 仅模拟一次异步等待，不访问网络，也不是性能测试。
    await asyncio.sleep(0.01)
    return f"完成 {name}"


async def main():
    async with asyncio.TaskGroup() as group:
        first = group.create_task(fetch_label("A"))
        second = group.create_task(fetch_label("B"))

    # 离开任务组后两项都已完成；按固定顺序读取结果。
    print(first.result())
    print(second.result())


if __name__ == "__main__":
    asyncio.run(main())

# 输出：
# 完成 A
# 完成 B

两个容易混淆的点：

• 连续写 await fetch_label("A")、await fetch_label("B") 是依次等待；上例先创建两个任务，让它们并发推进。
• 任务组中出现非取消异常时，会取消其余任务并等待清理，通常以异常组报告失败；不要静默吞掉取消异常。见 TaskGroup。

事件循环里不要直接执行 time.sleep()、阻塞网络请求或长时间 CPU 循环。 阻塞 I/O 可按需交给 asyncio.to_thread()；纯 Python 重计算通常另选进程池。

脚本用 asyncio.run() 启动；Jupyter 等已有事件循环的环境通常直接 await main()，不要再嵌套调用 asyncio.run()。

12. TCP：传的是字节流，不是消息包

客户端主动连接，服务端监听并接受连接。accept() 返回的连接 socket 才用于与该客户端通信；监听 socket 继续负责接受新连接。

一次 send 不对应一次 recv；TCP 本身不保留应用消息边界。 实际协议需约定长度前缀、分隔符或结束连接等规则。见 Socket HOWTO。

只在本机运行的最小示例

下面发送一条很短的消息，用“关闭发送方向”表示消息结束。特意每次最多读 2 字节，演示为什么不能收到一块就立即解码中文。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

import socket

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
    # AF_INET 使用 IPv4；127.0.0.1 仅本机可访问，端口 0 由系统分配。
    server.bind(("127.0.0.1", 0))
    server.listen(1)
    server.settimeout(2)

    with socket.create_connection(server.getsockname(), timeout=2) as client:
        connection, _ = server.accept()
        with connection:
            connection.settimeout(2)
            client.sendall("你好".encode("utf-8"))
            client.shutdown(socket.SHUT_WR)

            chunks = []
            # := 将本次读取结果绑定到 chunk，同时检查是否到达流结束。
            while chunk := connection.recv(2):
                chunks.append(chunk)

print(b"".join(chunks).decode("utf-8"))

# 输出：你好

:= 海象运算符：赋值后，把值用于判断

:= 叫赋值表达式，也叫海象运算符，不是海马运算符；它在 Python 3.8 就已经引入。

一个不涉及网络的例子：

1
2
3
4
5

text = "hello"
if (length := len(text)) > 3:
    print(length)

# 输出：5

括号里的表达式先把 5 保存到 length，再用 5 > 3 判断是否进入代码块。括号也明确了“先赋值，再比较”的分组。

回到 TCP 示例，while chunk := connection.recv(2): 每轮执行三步：

1. 调用一次 recv(2)，最多读取 2 字节。
2. 把读取结果保存到 chunk，随后判断这个字节串是否非空。
3. 非空就执行 chunks.append(chunk)，再开始下一轮；读到 b"" 就结束循环。

它相当于把“读取、赋值、判断是否结束”合在循环条件里，不是重复读取两次，也不是判断相等。对于这里的 TCP 读取，b"" 表示对方发送方向结束，暂时没有数据时会等待或超时。见 赋值表达式。

收包时还要注意：

• 先收齐，再解码：UTF-8 字符可能跨读取边界；长数据流也可使用增量解码器。
• shutdown(SHUT_WR)：关闭发送方向，不等于立即关闭整个 socket。
• with 和超时：分别负责关闭资源、限制阻塞等待；超时异常仍需由实际业务处理。

这里只发送几个字节，能在同一线程里完成演示；大数据双向通信需要双方持续收发，不能照搬这个发送完再接收的组织方式。API 细节见 socket。

本地练习绑定 127.0.0.1。 IPv4 中绑定空字符串或 0.0.0.0 会监听所有本地网络接口，不是“仅本机”。

13. HTTP：交给协议库处理

TCP 的可靠传输不等于安全传输，它本身不提供加密和身份认证。HTTPS 在 HTTP 通信中加入 TLS 保护。见 TCP 安全边界。

协议差异见 HTTP/1.1、HTTP/2和 HTTP/3。

13.1 解析 URL，不要手工切字符串

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

from urllib.parse import parse_qs, urlsplit

url = urlsplit("https://example.com/search?q=python&page=2#intro")
print(url.scheme, url.hostname)
print(url.path)
print(parse_qs(url.query))
print(url.fragment)

# 输出：
# https example.com
# /search
# {'q': ['python'], 'page': ['2']}
# intro

• 查询参数：parse_qs() 返回列表值，是因为同一个参数可以出现多次。
• 片段标识：#intro 不会作为普通 HTTP 请求的请求目标发送给服务器。
• 安全边界：解析 URL 不等于验证它安全或允许访问。见 urllib.parse。

13.2 看懂报文，不手写生产服务器

下面是 HTTP/1.1 请求结构示意，不是程序输出；实际行结束符使用 \r\n，头部后有空行：

1
2
3

GET /search?q=python HTTP/1.1
Host: example.com
Accept: application/json

• 请求：关注方法、目标路径、头部和可选正文；GET 请求正文没有通用语义，通常不应发送。
• 响应：关注状态码、头部和正文；并非所有响应都带正文，例如 204 响应。
• 消息边界：依照协议中的长度、分块等规则处理，不能靠一次 recv(1024) 或 split(" ") 解析完整请求。见 HTTP 语义和 HTTP/1.1 消息格式。

不要把请求路径直接拼到本地目录后读取文件。 旧式手写静态服务器还需要处理路径越界、协议解析、超时和并发限制；这些不适合靠几段教学代码承担。

HTTP 请求优先使用标准库 urllib.request 或项目选定的客户端库。标准库 http.server 可用于受控本地预览，但不适合作为生产服务器。