
本文档是一份系统、深入的 Python 高级特性教程。每个主题都按"原理 → 示例 → 底层机制 → 常见陷阱 → 实践建议"展开,而非罗列代码片段。配套 9 幅图示,覆盖对象模型、函数式编程、元编程、执行模型、并发与性能。
阅读前提:熟悉 Python 基础语法(
def/class/for/try、列表/字典、模块导入)、理解"一切皆对象"的直觉。本文不重复基础语法。
__slots__ 内存优化functools 工具集itertools 与函数式工具asyncio 异步编程super() 深度解析collections 常用容器dataclassesPython 的"高级特性"看似零散,实则可以用两条主线串起来:
type 的实例),方法其实是描述符,属性访问有固定的查找链。理解了对象模型,你就理解了装饰器、描述符、元类、魔术方法为什么那样工作。yield 如何让函数挂起、事件循环如何调度协程、GIL 如何影响多线程。理解了执行模型,你就理解了生成器、asyncio、并发与性能调优。下面这张图把全部主题归入 6 个分区,建议按"对象模型 → 函数式 → 执行模型 → 元编程 → 数据结构 → 性能"的顺序阅读。

本文的定位:不是语法手册,而是"为什么这样设计"与"什么时候该用"。所有代码均可直接运行(Python 3.10+ 验证)。遇到不确定的行为,优先用
python -i或ipython亲手试。
核心概念:双下划线方法(dunder = double underscore)是 Python 对象模型的基础。当你写 a + b、len(x)、for v in obj、print(obj) 时,解释器实际调用的是 obj.__add__(b)、x.__len__()、obj.__iter__()、obj.__str__()。重载这些方法,就能让你的类"像内置类型一样自然地工作"。
__new__ / __init__ / __del__
这是最容易被误解的一组。__new__ 和 __init__ 职责完全不同:
| 方法 | 调用时机 | 职责 | 返回值 |
|---|---|---|---|
__new__(cls, ...) |
最先调用 | 分配对象内存(创建实例) | 必须返回新实例 |
__init__(self, ...) |
__new__ 之后 |
初始化实例属性 | 必须返回 None(返回其他值会报错) |
__del__(self) |
对象被 GC 回收前 | 清理资源 | 无(返回值被忽略) |
class Foo:
def __new__(cls, *args, **kwargs):
# 真正的"构造函数":负责把内存造出来
print("__new__ 分配实例")
instance = super().__new__(cls)
return instance # 关键:必须返回实例,否则 __init__ 不会被调用
def __init__(self, value):
print("__init__ 初始化")
self.value = value # 只赋值,不要 return
def __del__(self):
print("__del__ 析构") # 不可靠:调用时机不确定
f = Foo(10)
# 输出: __new__ 分配实例
# __init__ 初始化
为什么需要 __new__? 因为 __new__ 是静态方法(第一个参数是类,不是实例),它可以在返回实例前做决定,甚至返回一个不同类的实例。典型应用场景:
# 不可变类型(int/str/tuple)和需要"缓存/池化"的对象必须重写 __new__
class PositiveInt(int):
def __new__(cls, value):
# int 是不可变的,必须在创建时修正值,__init__ 无法修改已创建好的 int
return super().__new__(cls, abs(value))
PositiveInt(-5) # 5 —— 负数被转成绝对值
__del__ 为什么不可靠? 它的调用时机由垃圾回收决定,不保证"出了作用域立即执行";循环引用还可能延迟它;更危险的是,__del__ 中抛出异常会被忽略。with 语句和上下文管理器才是资源清理的正确手段(见第 9 节)。
__repr__ / __str__ / __format__
__repr__:给开发者/调试器看,目标是"尽可能无歧义、最好能 eval 还原对象"。约定:repr(x) == "ClassName(args)"。__str__:给最终用户看,目标是"可读"。print(x) 和 str(x) 优先用 __str__,没有才退回 __repr__。__format__:支持 format(x, spec) 与 f-string 的 :spec 部分。class Point:
def __init__(self, x, y):
self.x, self.y = x, y
def __repr__(self):
# 首选:能直接看到如何重建这个对象
return f"Point(x={self.x}, y={self.y})"
def __str__(self):
# 给用户:简洁可读
return f"({self.x}, {self.y})"
def __format__(self, spec):
# 支持 f"{p:.2f}" 之类
if spec == ".2f":
return f"({self.x:.2f}, {self.y:.2f})"
return str(self)
p = Point(3.14159, 2.71828)
print(repr(p)) # Point(x=3.14159, y=2.71828)
print(str(p)) # (3.14159, 2.71828)
print(f"{p:.2f}") # (3.14, 2.72)
经验法则:每个"数据类"都应实现
__repr__——调试时能立刻看清对象内容,省去大量print(obj.__dict__)。
重载运算符时,必须理解三组规则:反射运算符、原地运算符、以及 __eq__/__hash__ 的契约。
class Vector:
def __init__(self, x, y):
self.x, self.y = x, y
def __add__(self, other): # self + other
return Vector(self.x + other.x, self.y + other.y)
def __radd__(self, other): # other + self(当 other 没有 __add__ 时反射调用)
return Vector(other + self.x, other + self.y)
def __iadd__(self, other): # self += other(原地,应返回 self)
self.x += other.x
self.y += other.y
return self
def __eq__(self, other): # ==
return isinstance(other, Vector) and self.x == other.x and self.y == other.y
def __hash__(self): # 与 __eq__ 配套
return hash((self.x, self.y))
def __lt__(self, other): # <(排序用)
return (self.x, self.y) < (other.x, other.y)
三大规则:
__radd__、__rmul__ 等):当 a + b 中 a.__add__(b) 返回 NotImplemented 时,Python 会尝试 b.__radd__(a)。这是让 1 + Vector(...)(数字在左)也能工作的关键。__iadd__ 等):a += b 优先调 __iadd__,返回 self 实现真正的原地修改(可变对象省内存)。不可变对象(如自定义元组)应不实现 __iadd__,退回 __add__。__eq__ 与 __hash__ 的契约(极重要):如果两个对象 __eq__ 相等,它们的 __hash__ 必须相等。一旦你重写了 __eq__,Python 会自动把类变成不可哈希(删掉 __hash__);想让它可哈希,必须显式重写 __hash__。反之,若对象可变且参与哈希集合,应在 __hash__ 和 __eq__ 上保持一致性。陷阱:只重写
__eq__不写__hash__,对象就不能放进set/dict的键——常导致TypeError: unhashable type。列表 (list)、字典 (dict) 等可变内置类型正是因此不可哈希。
让自定义类表现得像序列/映射:
class MyList:
def __init__(self, data):
self._data = list(data)
def __len__(self): # len(obj)
return len(self._data)
def __getitem__(self, idx): # obj[idx],支持切片(idx 可能是 slice)
return self._data[idx]
def __setitem__(self, idx, val): # obj[idx] = val
self._data[idx] = val
def __delitem__(self, idx): # del obj[idx]
del self._data[idx]
def __contains__(self, val): # val in obj
return val in self._data
def __iter__(self): # for x in obj
return iter(self._data)
def __reversed__(self): # reversed(obj)
return reversed(self._data)
要点:
__getitem__ 收到的 idx 可能是整数,也可能是 slice 对象(如 obj[1:3])。若要支持切片,需自行判断 isinstance(idx, slice)。__getitem__ + __len__ 的对象就满足 collections.abc.Sequence 的鸭子类型,可当作序列使用。__setitem__/__delitem__/keys/values/items 等,或用 collections.UserDict 包装(见第 15 节)。class CallableObj:
def __call__(self, *args, **kwargs):
print("被当作函数调用:", args)
return sum(args)
f = CallableObj()
f(1, 2, 3) # 被当作函数调用: (1, 2, 3) → 6
__call__ 让实例像函数一样可调用。常见于需要"持有状态的函数":缓存器、配置化处理器、策略对象。
属性访问拦截是高级但危险的领域,务必分清三者:
| 方法 | 触发时机 | 风险 |
|---|---|---|
__getattr__(self, name) |
常规属性查找失败后(兜底) | 安全,常用于代理/惰性加载 |
__getattribute__(self, name) |
每次属性访问都触发 | 极易无限递归,必须 super().__getattribute__
|
__setattr__(self, name, val) |
每次属性赋值 | 同样易递归,必须 super().__setattr__
|
class Proxy:
def __init__(self, target):
self._target = target
def __getattr__(self, name):
# 只有找不到 name 时才走到这里(_target 自身属性不会触发)
return getattr(self._target, name)
class Dangerous:
def __getattribute__(self, name):
# 每次访问属性都触发,包括读取 self._x 也会再触发 → 无限递归
# 正确写法:
return super().__getattribute__(name)
def __setattr__(self, name, value):
# 错误写法:self._data = value 又会触发 __setattr__ → 递归
# 正确写法:
super().__setattr__(name, value)
属性访问的黄金法则:
__getattr__是"安全兜底",优先用它;__getattribute__/__setattr__必须用super()委托,否则直接栈溢出。绝大多数场景不需要它们——描述符(第 3 节)是更可控的属性拦截方案。
补充:__missing__(字典专属):dict 子类实现 __missing__(key),可在 d[key] 缺失时返回默认值而不抛 KeyError。这是 collections.defaultdict 的底层机制。
核心概念:描述符是实现了 __get__、__set__、__delete__ 中至少一个的对象。它表面上不起眼,却是 Python 属性系统的底层引擎——你每天用的 property、classmethod、staticmethod,以及 Django、SQLAlchemy、Pydantic 里的字段,全都是描述符。
描述符按"是否定义了 __set__"分为两类,这决定了它的优先级:
__set__(可能还有 __delete__)。优先级最高,连实例字典都压不过它。__get__(如普通函数/方法)。优先级低于实例字典。class Validated:
def __init__(self, validator):
self.validator = validator
def __set_name__(self, owner, name):
# Python 3.6+:描述符被赋给类属性时,自动得知自己的名字
self.name = f"_{name}"
def __get__(self, inst, owner):
# inst 为实例;通过类访问时 inst 为 None
if inst is None:
return self
return getattr(inst, self.name, None)
def __set__(self, inst, value):
if not self.validator(value):
raise ValueError(f"校验失败: {self.name}")
setattr(inst, self.name, value)
class User:
name = Validated(str.isalpha) # 仅字母
age = Validated(lambda x: isinstance(x, int) and 0 < x < 150)
u = User()
u.name = "Alice"
u.age = 30
# u.age = -1 # ValueError: 校验失败
# u.name = "Bob1" # ValueError: 校验失败
__get__(self, inst, owner):owner 是拥有该描述符的类。inst is None 表示通过类访问(如 User.name),此时通常返回描述符本身。__set_name__ 让你不必手写字符串属性名,框架里常用来自动登记字段。当你写 obj.x,Python 按以下固定顺序查找:

x 是 data descriptor(有 __set__)→ 调用它的 __get__。优先级最高。obj.__dict__['x']。type(obj).__dict__['x'],若找到的是 non-data descriptor(如函数)→ 调用 __get__(这把函数变成"绑定方法")。__getattr__(若有)。再找不到 → AttributeError。这就是关键洞察:普通方法是 non-data descriptor。因此当你写 obj.foo = 123,实例字典里有了 foo,它就遮蔽了类上的 foo 方法(因为步骤 2 在步骤 3 之前)。而 property 是 data descriptor,所以即使你给实例赋值 obj.x = ...,也会被 property 的 __set__ 拦截——实例字典里根本不会出现 x。
property本质:property(fget, fset, fdel, doc)就是一个 data descriptor 工厂。@property是语法糖,等价于创建一个带__get__/__set__的描述符实例并赋给类属性。
缓存描述符(lazy property):首次访问计算,之后缓存到实例字典。
class cached_property:
def __init__(self, func):
self.func = func
self.name = func.__name__
def __set_name__(self, owner, name):
self.name = name
def __get__(self, inst, owner):
if inst is None:
return self
# 算一次后直接写进实例字典:下次查找步骤 2 命中,不再走描述符
value = self.func(inst)
inst.__dict__[self.name] = value
return value
class Report:
def __init__(self, data):
self.data = data
@cached_property
def summary(self):
print("计算一次...")
return sum(self.data)
r = Report([1,2,3])
r.summary # 计算一次... → 6
r.summary # 直接读缓存 → 6
提示:
functools.cached_property内置了这个模式(Python 3.8+),生产代码直接用它即可。
__slots__ 内存优化原理:普通对象每个实例都带一个 __dict__(字典),用来存放任意动态属性。字典本身有开销(哈希表、指针数组),且无法高效紧凑排列。声明 __slots__ 后,Python 不再为实例创建 __dict__,而是把列出的属性变成固定大小的"槽位"(类似 C 结构体),直接存在对象内存里。
class Point:
__slots__ = ('x', 'y') # 只允许这两个属性
def __init__(self, x, y):
self.x, self.y = x, y
p = Point(1, 2)
# p.z = 3 # AttributeError: 'Point' object has no attribute 'z'
100 万个实例时,差距非常显著:

import sys, time, tracemalloc
class WithDict:
def __init__(self, x, y): self.x, self.y = x, y
class WithSlots:
__slots__ = ('x', 'y')
def __init__(self, x, y): self.x, self.y = x, y
def bench(cls, n=1_000_000):
tracemalloc.start()
objs = [cls(i, i) for i in range(n)]
cur, peak = tracemalloc.get_traced_memory()
tracemalloc.stop()
return peak / 1024 / 1024
print(f"普通类: {bench(WithDict):.1f} MB") # ~150 MB
print(f"slots: {bench(WithSlots):.1f} MB") # ~55 MB(约省 60%)
__slots__ 不是免费的,务必注意:
| 收益 | 代价 |
|---|---|
| 省内存(大量实例时显著) | 不能动态添加未在 __slots__ 中声明的属性 |
| 属性访问略快(少一层字典查找) | 多继承时,只有所有父类都定义 __slots__ 才有效 |
| 可阻止意外属性名拼写错误 | 默认没有 __dict__,需要弱引用支持时还要加 '__weakref__'
|
| 继承链复杂时易踩坑 |
class Node:
__slots__ = ('value', '__weakref__') # 若要用 weakref.ref,需显式加入 __weakref__
def __init__(self, v): self.value = v
决策准则:仅当实例数量达到十万级以上且属性集合固定时才值得用(如 ORM 行对象、海量数据点)。日常业务对象、配置类、DTO 用
@dataclass更合适——可读性与灵活性远比那点内存重要。__slots__ = ()还可让类本身不可实例化出有__dict__的对象。
概念:内层函数捕获了外层作用域的变量(自由变量),即使外层函数已经返回,这些变量依然存活——因为 Python 用"cell 对象"持有它们,形成闭包。
def make_multiplier(factor):
def multiply(x):
return x * factor # factor 是自由变量,被闭包捕获
return multiply
double = make_multiplier(2)
triple = make_multiplier(3)
print(double(5), triple(5)) # 10 15
# make_multiplier 已返回,但 factor=2 / 3 仍被 double/triple 持有
验证闭包捕获的变量:
print(double.__closure__[0].cell_contents) # 2
关键点:
multiply没有自己的factor,它引用的是外层作用域的factor。每次调用make_multiplier都会创建一个新的 cell。注意循环 + 闭包的经典陷阱(见下)。
装饰器本质就是语法糖:@deco 等价于 func = deco(func)。它把闭包用于"在不修改原函数代码的前提下,增强其行为"。
import functools, time
def timer(func):
@functools.wraps(func) # 关键:保留原函数的 __name__/__doc__/__module__
def wrapper(*args, **kwargs):
t = time.perf_counter()
result = func(*args, **kwargs)
print(f"{func.__name__} 耗时 {time.perf_counter()-t:.4f}s")
return result # 必须原样返回原函数的返回值
return wrapper
@timer
def add(a, b):
"""两数相加"""
return a + b
# 等价于: add = timer(add)
add(3, 4) # 打印耗时并返回 7
print(add.__name__) # "add"(因为 @wraps);否则会是 "wrapper"
为什么 @functools.wraps 不可省略? 没有它,被装饰函数的元信息(__name__、__doc__、__module__、参数签名)会变成 wrapper 的,导致:日志显示错误函数名、inspect.signature 失效、pickle 失败、基于 __name__ 的路由/注册出错。
当装饰器本身要接收参数时,就变成三层结构:最外层接收参数,返回"真正的装饰器",再由它包裹函数。
import functools, time
def retry(max_attempts=3, delay=1):
def decorator(func): # 第二层:真正的装饰器
@functools.wraps(func)
def wrapper(*args, **kwargs):
for i in range(max_attempts):
try:
return func(*args, **kwargs)
except Exception:
if i == max_attempts - 1:
raise
time.sleep(delay)
return wrapper
return decorator # 第一层返回装饰器
@retry(max_attempts=5, delay=0.5)
def fetch_data():
...
难点:@retry 与 @retry(...) 写法不同。@retry 时 Python 把被装饰函数直接传给 retry;@retry(5) 时先调用 retry(5) 得到装饰器再包裹。要同时支持两种写法,可在第一层判断第一个参数是否是可调用的函数(略复杂,生产建议分开)。
当装饰器需要持有状态(如计数、缓存字典),用类更自然——类实例本身就是状态容器。
import functools
class CountCalls:
def __init__(self, func):
functools.update_wrapper(self, func) # 等价于 wraps
self.func = func
self.count = 0
def __call__(self, *args, **kwargs):
self.count += 1
return self.func(*args, **kwargs)
@CountCalls
def greet(name):
return f"hi {name}"
greet("A"); greet("B")
print(greet.count) # 2
多个装饰器从下往上包裹(离函数最近的先应用):
@a
@b
@c
def f(): ...
# 等价于: f = a(b(c(f))) # c 先包,a 最后包;调用时 a→b→c→原函数的层层进入

实战清单:计时、日志、重试、缓存 (
lru_cache)、权限校验、注册路由、单例、弃用警告 (warnings) 都是装饰器的典型用途。记住铁律——装饰器必须返回可被调用且能透传返回值/异常的对象,并保持@wraps。
functools 工具集functools 是函数式编程的工具箱。以下几项是高频必备。
partial — 偏函数(冻结部分参数)from functools import partial
def power(base, exp):
return base ** exp
square = partial(power, exp=2) # 固定 exp=2
cube = partial(power, exp=3)
print(square(5), cube(4)) # 25 64
# 实战:给回调预置参数(GUI/异步里非常常见)
on_click = partial(handle_click, button_id=7)
partial 返回的对象会保留 __name__/__doc__,但 inspect.signature 会反映剩余参数。注意 partial 是浅冻结:若冻结的是可变对象,调用方仍可能修改它。
lru_cache — 记忆化缓存把函数结果按参数缓存,相同输入直接返回,避免重复计算。
from functools import lru_cache
@lru_cache(maxsize=128)
def fib(n):
if n < 2:
return n
return fib(n-1) + fib(n-2)
print(fib(100)) # 毫秒级(否则指数爆炸)
print(fib.cache_info())
# CacheInfo(hits=98, misses=101, maxsize=128, currsize=101)
关键细节:
__hash__)。传列表会报错——先转元组。maxsize=None 表示无限缓存;maxsize=2**n 性能最佳(用环形链表)。typed=True 时,f(3) 与 f(3.0) 视为不同键。fib.cache_clear() 清空缓存。singledispatch — 单分派泛型函数按第一个参数类型自动选择实现,是"重载"的优雅替代(Java/C++ 才有真正重载,Python 用这个)。
from functools import singledispatch
@singledispatch
def process(data):
raise NotImplementedError(f"不支持类型: {type(data)}")
@process.register
def _(data: int):
return data * 2
@process.register
def _(data: str):
return data.upper()
@process.register(list)
def _(data): # 也可写成 register(list)
return [process(x) for x in data]
print(process(42)) # 84
print(process("hello")) # HELLO
print(process([1, "a", 3])) # [2, 'A', 6]
Python 3.11+ 还有
singledispatchmethod:用于类方法的分派。
reduce — 累积归约from functools import reduce
product = reduce(lambda a, b: a * b, [1, 2, 3, 4]) # 24
max_val = reduce(lambda a, b: a if a > b else b, [3, 7, 2, 9]) # 9
# 可给初始值(避免空序列报错):reduce(fn, seq, init)
多数情况下
functools.reduce不如显式for或sum/max可读。仅当"把序列折叠成一个值"且逻辑紧凑时使用。
wraps / update_wrapper
已见 5.2。补充:update_wrapper 是底层 API,wraps 是它的装饰器封装。它复制 __module__、__name__、__qualname__、__doc__、__dict__ 等,并设 __wrapped__ 指向原函数(供 inspect 穿透)。
核心概念:Python 的 for 循环、解包、* 展开、in 判断,底层都依赖迭代器协议。理解它,就理解了"惰性计算"和"流式处理"的基础。
__iter__() 返回迭代器的对象(如 list、str、dict、set、生成器)。__iter__()(返回自身)和 __next__()(返回下一个元素;耗尽时抛 StopIteration)的对象。class Countdown:
def __init__(self, start):
self.start = start
def __iter__(self): # 返回迭代器(这里就是自己)
return self
def __next__(self): # 每次调用推进一次
if self.start <= 0:
raise StopIteration
self.start -= 1
return self.start + 1
for n in Countdown(3):
print(n) # 3 2 1
# 底层完全等价:
it = iter(Countdown(3)) # 调用 __iter__
while True:
try:
print(next(it)) # 调用 __next__
except StopIteration:
break

关键区别:迭代器是"一次性"的——走完(StopIteration)后再 next() 仍抛异常,且不会自动重置。列表是可迭代但不是迭代器(每次 iter(lst) 都返回全新的迭代器);而生成器本身就是迭代器。
nums = [1, 2, 3]
it1 = iter(nums); it2 = iter(nums)
next(it1); next(it1) # 1, 2
next(it2) # 1 —— 独立的迭代器,互不影响
StopIteration的现代陷阱(PEP 479):在生成器里,若正常return会以StopIteration结束;但如果在生成器内部意外抛出StopIteration,Python 3.7+ 会把它转换为RuntimeError,防止它错误地终止外层for。这是历史包袱的修复,理解即可。
实战价值:自定义迭代器让你能处理无限流或超大数据集而不占内存——每次只产生一个元素。这也是 itertools、生成器 的基石。
itertools 与函数式工具itertools 提供一组惰性、内存友好的迭代器,是高效数据处理利器。下面逐一给出用法与输出。
import itertools
# --- 无限迭代器(务必配合 islice / 条件终止,否则死循环)---
itertools.count(10, 2) # 10, 12, 14, 16, ...(从 10 起步长 2)
itertools.cycle("AB") # A, B, A, B, A, B, ...(无限循环)
itertools.repeat("x", 3) # x, x, x(有限次,可省略次数成无限)
# --- 组合迭代器 ---
itertools.product("AB", "12") # 笛卡儿积
# ('A','1'), ('A','2'), ('B','1'), ('B','2')
itertools.permutations("AB", 2) # 排列(有序):('A','B'), ('B','A')
itertools.combinations("ABC", 2) # 组合(无序):('A','B'), ('A','C'), ('B','C')
itertools.combinations_with_replacement("AB", 2) # 允许重复:('A','A'),('A','B'),('B','B')
# --- 终止迭代器 ---
list(itertools.accumulate([1, 2, 3, 4])) # 前缀和: [1, 3, 6, 10]
list(itertools.chain([1, 2], "ab")) # 拼接: [1, 2, 'a', 'b']
list(itertools.chain.from_iterable([[1,2],[3,4]])) # 展平: [1, 2, 3, 4]
list(itertools.groupby("aaabbc")) # 按键分组(注意:键必须已排序!)
# [('a', <g>), ('b', <g>), ('c', <g>)] 每个值是迭代器
list(itertools.islice(range(100), 5, 15)) # 切片(惰性): [5..14]
list(itertools.zip_longest("AB", "123")) # 不等长补 None: [('A','1'),('B','2'),(None,'3')]
list(itertools.pairwise([1, 2, 3, 4])) # 3.10+ 滑动窗口: [(1,2),(2,3),(3,4)]
# 1) 分组前必须先排序(groupby 只连续归组)
data = sorted([("a",1),("b",2),("a",3)])
for key, grp in itertools.groupby(data, key=lambda x: x[0]):
print(key, list(grp))
# a [('a',1),('a',3)]
# b [('b',2)]
# 2) 滑窗统计(如滚动平均)
def windowed(seq, n):
iters = itertools.tee(seq, n) # 复制 n 份迭代器
for i, it in enumerate(iters):
next(itertools.islice(it, i, i), None) # 各自偏移 i
return zip(*iters)
list(windowed([1,2,3,4,5], 3)) # [(1,2,3),(2,3,4),(3,4,5)]
# 3) 生成器表达式(最 Pythonic 的惰性映射+过滤)
squares = (x*x for x in range(10) if x % 2 == 0) # 不立即计算,省内存
tee会把迭代器"快照"成多份,但会缓冲已消费的元素——若只需顺序访问,优先考虑islice/手动游标,避免tee占用过多内存。
核心概念:上下文管理器让资源"自动清理"——无论代码块正常结束还是抛异常,__exit__(或 yield 之后)都会运行。这是 try/finally 的优雅封装,用于文件、锁、连接、计时、临时状态等。
__enter__ / __exit__
import time
class Timer:
def __enter__(self):
self.t = time.perf_counter()
return self # with ... as <val> 接收的就是它的返回值
def __exit__(self, exc_type, exc_val, exc_tb):
print(f"耗时 {time.perf_counter()-self.t:.4f}s")
return False # False=不吞异常(向外传播);True=吞掉异常
with Timer() as t:
time.sleep(0.5)
# 输出: 耗时 0.50xx s
# sleep 中若抛异常,__exit__ 仍会被调用,且异常继续向外抛出(因返回 False)
__exit__ 的三个参数在无异常时全为 None;有异常时分别为异常类型、实例、回溯对象。
contextmanager — 生成器方式(更简洁)@contextmanager 把"含 yield 的生成器"变成上下文管理器:yield 之前是进入逻辑,yield 之后是退出逻辑。
from contextlib import contextmanager
import time, os
@contextmanager
def timer():
t = time.perf_counter()
yield # 这里的 yield 值 = as 接收的值(此处为 None)
print(f"耗时 {time.perf_counter()-t:.4f}s")
@contextmanager
def temp_env(key, value): # 临时修改环境变量,退出还原
old = os.environ.get(key)
os.environ[key] = value
try:
yield
finally: # 即使块内异常,也确保还原
if old is None:
os.environ.pop(key, None)
else:
os.environ[key] = old
with temp_env("MODE", "prod"):
print(os.environ["MODE"]) # prod
print(os.environ.get("MODE")) # 恢复为 None
异常处理:
@contextmanager包裹的函数里,若yield之后的代码抛异常(如finally中的清理也出错),该异常会传播;若块内抛异常,它会在yield处重新抛出,由try/except捕获或传播。务必用try/finally保证清理。
contextlib 工具from contextlib import suppress, closing, ExitStack, nullcontext, redirect_stdout
import os, io, sys
with suppress(FileNotFoundError): # 优雅忽略指定异常(不必写 try/except/pass)
os.remove("nonexistent.txt")
with closing(open("f.txt")) as f: # 确保 .close() 被调用(对象本身不支持 with 时)
data = f.read()
with ExitStack() as stack: # 动态管理数量不定的多个上下文
files = [stack.enter_context(open(f"file{i}.txt")) for i in range(3)]
# 退出时全部自动关闭,按逆序
with redirect_stdout(io.StringIO()) as buf: # 临时重定向 stdout(测试很有用)
print("hello")
captured = buf.getvalue() # "hello\n"
with nullcontext() as c: # 占位上下文(根据条件决定是否套真实上下文)
pass

额外:
contextlib.asynccontextmanager提供异步版本的@contextmanager,配合async with使用(见第 11 节)。
yield 让函数在产出值后挂起并保留局部状态,下次被唤醒时从暂停点继续。与一次性算出全部结果的列表相比,生成器按需产生、几乎零内存。
# 生成器表达式(惰性)
g = (x*x for x in range(10) if x % 2 == 0)
print(next(g), next(g)) # 0 4
# yield from:委托给子生成器(自动转发 yield/值,并收集子生成器 return 的值)
def chain(*iters):
for it in iters:
yield from it # 等价于 for x in it: yield x
print(list(chain([1, 2], (3, 4)))) # [1, 2, 3, 4]
生成器不仅能产出(yield 给调用者),还能通过 send() 接收值——这就成了协程(双向数据流)。throw() 注入异常,close() 终止。
def accumulator():
total = 0
while True:
x = yield total # 暂停,把 total 交出去,等待 send 注入到 x
if x is None:
break
total += x
acc = accumulator()
next(acc) # 预激活:必须先 next() 或 send(None) 跑到第一个 yield
print(acc.send(10)) # 10 (total=10)
print(acc.send(5)) # 15 (total=15)
acc.send(None) # 结束(break → StopIteration)
# acc.close() # 也可主动关闭,触发 GeneratorExit
生命周期状态机:

| 状态 | 说明 |
|---|---|
| CREATED | 调用生成器函数,返回生成器对象;函数体尚未执行 |
| RUNNING | 调用 next()/send() 后进入,执行函数体直到遇到 yield
|
| SUSPENDED | 在 yield 处挂起,把值交给调用者;等待下一次唤醒 |
| CLOSED | 函数执行完毕(return/StopIteration)、被 close() 或抛异常,生命周期结束 |
yield from 异常传播yield from 不仅是语法糖,它还会:把子生成器的 yield 直接透传给最外层、把 send 直接转发给子生成器、把子生成器的 return value 作为 yield from 表达式的值(PEP 380 的核心贡献)。
def sub():
yield 1
return "done"
def outer():
result = yield from sub() # result == "done"
yield result
g = outer()
print(next(g)) # 1
print(next(g)) # "done"
send/throw 构建生产者 - 消费者。itertools + 生成器表达式替代多轮列表拷贝。现代写法:需要双向协程时,优先用
async/await(见第 11 节)——它语义更清晰、配套设施更全。手写yield协程如今主要用于数据管道和惰性求值。
asyncio 异步编程核心概念:asyncio 提供单线程事件循环 + 协作式调度。协程在 await 处主动交还控制权,事件循环转去运行其他就绪任务;IO 完成后再唤醒它。没有抢占、没有线程切换开销——因此能以极低代价支撑海量并发 IO。
import asyncio, time
async def fetch(i):
await asyncio.sleep(1) # 模拟 IO:此处让出控制权,不阻塞线程
return i * 10
async def main():
t = time.perf_counter()
results = await asyncio.gather(fetch(1), fetch(2), fetch(3))
print(results, f"用时 {time.perf_counter()-t:.2f}s")
asyncio.run(main()) # [10, 20, 30] 用时约 1.00s(而非 3s)
关键术语:
| 概念 | 说明 |
|---|---|
coroutine |
async def 返回的对象本身不执行,必须被 await 或包成 Task |
Task |
asyncio.create_task(coro) 把协程登记到事件循环,立即开始调度(并发的真正来源) |
await |
只对 awaitable 生效;阻塞当前协程直到完成,但不阻塞事件循环 |
Event Loop |
调度中枢:维护就绪队列,IO 完成回调唤醒对应 Task |
async def fetch(i):
await asyncio.sleep(1)
return i
# ❌ 错误:顺序 await → 退化成串行(总耗时 3s)
async def wrong():
r1 = await fetch(1)
r2 = await fetch(2)
r3 = await fetch(3)
# ✅ 正确:先把协程变成 Task 并发调度
async def right():
t1 = asyncio.create_task(fetch(1))
t2 = asyncio.create_task(fetch(2))
t3 = asyncio.create_task(fetch(3))
results = await asyncio.gather(t1, t2, t3) # 总耗时约 1s
# gather 也支持直接传协程(内部会自动 create_task)
results = await asyncio.gather(fetch(1), fetch(2), fetch(3))
import asyncio
async def worker(q, wid):
while True:
item = await q.get()
if item is None:
q.task_done()
break
await asyncio.sleep(0.1) # 模拟处理
print(f"worker {wid} 处理 {item}")
q.task_done()
async def producer(q):
for i in range(5):
await asyncio.sleep(0.05)
await q.put(i)
for _ in range(3): # 发 3 个 None 让 3 个 worker 退出
await q.put(None)
async def main():
q = asyncio.Queue()
workers = [asyncio.create_task(worker(q, i)) for i in range(3)]
await asyncio.gather(producer(q), *workers)
asyncio.run(main())
在协程里调用同步阻塞函数(如 requests.get、time.sleep、open 读大文件)会卡死整个事件循环——因为阻塞期间控制权无法让出,所有其他任务都被冻住。解决方案:
import asyncio, concurrent.futures, requests
async def main():
loop = asyncio.get_running_loop()
# 把阻塞函数丢到线程池执行,await 等待其完成后不阻塞循环
resp = await loop.run_in_executor(
None, requests.get, "https://example.com"
)
# 或者用 asyncio.to_thread(Python 3.9+,语义相同,更简洁)
data = await asyncio.to_thread(blocking_io_func, arg)
aiohttp、httpx.AsyncClient)或 run_in_executor/to_thread。loop.run_in_executor 丢到 ProcessPoolExecutor,或用 asyncio 之外的 multiprocessing。
更多要点:
asyncio.wait可设FIRST_COMPLETED/FIRST_EXCEPTION精细控制;asyncio.TimeoutError与asyncio.wait_for(coro, timeout)做超时;asyncio.shield防止取消传播;asyncio.CancelledError用于优雅取消;@asynccontextmanager写异步上下文管理器;async for遍历异步迭代器。取消时用task.cancel()并在协程内try/finally清理。
super() 深度解析最大的误解:super() 不是"父类的引用"。它返回的是一个代理对象,沿当前实例的 MRO(方法解析顺序)链,查找调用者的下一个类。这个区分在多继承时至关重要。
class A:
def foo(self):
print("A.foo")
class B(A):
def foo(self):
print("B.foo")
super().foo() # 沿 MRO 找 B 之后的下一个 foo(即 A.foo)
class C(A):
def foo(self):
print("C.foo")
super().foo()
class D(B, C):
def foo(self):
print("D.foo")
super().foo()
D().foo()
# 输出:
# D.foo
# B.foo
# C.foo
# A.foo
D 的 MRO 是 [D, B, C, A, object]。D.foo 里的 super() 找的是 MRO 中 D 之后第一个实现了 foo 的类 = B;B.foo 里的 super() 找 B 之后 = C;C.foo 里的 super() 找 C 之后 = A。注意 B 的 super() 跳到了"兄弟类" C——这正是协作式多继承(cooperative MRO)的精髓。
super() 的两种形式super() # 零参数:Python 3 自动填入当前类和 self
super(B, self) # 显式:沿 B 在 self 的 MRO 之后的类查找
super(__class__, self) # 函数内部用(避免硬编码类名)
零参数 super() 实际上是语法糖:编译器把当前 __class__ 和第一个参数(self/cls)注入进去。
要让 super() 链不断裂,所有协作者必须:
super() 调用链上的下一个方法;**kwargs 透传);object)能处理剩余参数。class Base:
def __init__(self, x, **kwargs):
super().__init__(**kwargs) # 把剩余 kwargs 传给下一个
self.x = x
class Mixin:
def __init__(self, y, **kwargs):
super().__init__(**kwargs)
self.y = y
class Derived(Base, Mixin):
def __init__(self, z, **kwargs):
super().__init__(**kwargs) # 沿 MRO: Derived→Base→Mixin→object
self.z = z
d = Derived(x=1, y=2, z=3) # 三个 __init__ 都被正确调用
经验:不要滥用手写多继承 +
super()协作链——复杂度高且易错。能用组合(has-a)就用组合。super()的常规正确用法是单继承里调用父类方法(最常见、最安全)。
核心概念:元类是"创建类的类"。在 Python 中,class 语句并非"声明",而是执行:解释器收集类体命名空间,再调用 type(name, bases, ns)(或指定的元类)构造出类对象。所以"类"也是对象——type 的实例。元类就是 type 的子类,拦截"类被创建"的瞬间。
type 的三参数形式# 动态创建类的等价写法:
def say_hi(self):
return f"hi from {self.name}"
Attrs = type("Attrs", (object,), {"name": "x", "say_hi": say_hi})
obj = Attrs()
print(obj.say_hi()) # hi from x
这等价于 class Attrs: name="x"; def say_hi(self): ...。
class LogMeta(type):
def __new__(mcs, name, bases, ns):
# __new__ 负责"造出类对象"
print(f"正在创建类: {name}, 基类: {bases}")
for k, v in ns.items():
if callable(v) and not k.startswith("_"):
ns[k] = debug_wrapper(v) # 在类创建期改写方法
return super().__new__(mcs, name, bases, ns)
def __init__(cls, name, bases, ns):
# __init__ 负责"初始化已造好的类对象"
super().__init__(name, bases, ns)
cls.__author__ = "meta"
class Base(metaclass=LogMeta):
def foo(self): ...
def bar(self): ...
# 打印: 正在创建类: Base, 基类: (<class 'object'>,)
# Base.foo / Base.bar 被自动包上日志;其所有子类也会经 LogMeta 创建
__new__ vs __init__(元类):__new__ 返回新类对象;__init__ 在类对象造好后做初始化。多数"改造类"的逻辑放 __new__。__prepare__(可选)还能定制类体命名空间的字典类型(如自动排序)。
class PluginMeta(type):
registry = {}
def __new__(mcs, name, bases, ns):
cls = super().__new__(mcs, name, bases, ns)
if bases: # 跳过基类自身
PluginMeta.registry[name.lower()] = cls
return cls
class Plugin(metaclass=PluginMeta):
pass
class JsonExporter(Plugin): ...
class CsvExporter(Plugin): ...
print(PluginMeta.registry)
# {'jsonexporter': <class ...>, 'csvexporter': <class ...>}
90% 的场景不需要元类。现代替代方案:
# 1) __init_subclass__(Python 3.6+):子类创建时回调,无需元类
class Plugin:
registry = {}
def __init_subclass__(cls, **kwargs):
super().__init_subclass__(**kwargs)
Plugin.registry[cls.__name__.lower()] = cls
# 2) __set_name__:描述符自动获知属性名(见第 3 节)

元类 vs 类装饰器:元类影响该类及其所有子类(在继承链创建时生效);类装饰器只作用在当前类上。决策顺序:能用函数/组合解决就别用装饰器;能用
__init_subclass__就别用元类;能用元类就别用更复杂的反射。元类适合框架级"批量改造类"(ORM、序列化器、API 路由自动注册),业务代码几乎用不到。
核心概念:抽象基类定义接口契约——声明"必须实现哪些方法",阻止不完整的类被实例化,并在缺失实现时尽早报错(而非运行到调用才崩溃)。
from abc import ABC, abstractmethod
class Animal(ABC):
@abstractmethod
def speak(self):
pass # 子类必须实现
@property
@abstractmethod
def legs(self): # 抽象属性
pass
class Dog(Animal):
legs = 4
def speak(self):
return "woof"
# Animal() # TypeError: Can't instantiate abstract class Animal with abstract method speak
dog = Dog() # OK
register() 把一个不继承 ABC 的类登记为"虚拟子类",使其通过 isinstance 检查——实现"结构化兼容"而不强耦合。
from collections.abc import Sized
class MyCollection:
def __len__(self):
return 42
Sized.register(MyCollection) # 注册为虚拟子类
print(isinstance(MyCollection(), Sized)) # True(虽未继承 Sized)
__subclasshook__:在 ABC 里重写它可实现自动鸭子类型识别(如Iterable只要对象有__iter__就认为是它的子类)。collections.abc大量使用此技巧。
collections.abc 抽象基类Iterable、Iterator、Sized、Container、Sequence、MutableSequence、Mapping、MutableMapping、Set、MutableSet、Callable、Hashable。用 isinstance(x, Sequence) 做接口检查比 hasattr(x,'__getitem__') 更可靠。
ABC vs Protocol(typing):ABC 是运行时强制(实例化时报错、可被
isinstance检测);Protocol是静态类型层面的结构化类型(mypy检查,运行时无约束,零开销)。需要运行时强制用 ABC;只做静态检查用 Protocol。
collections 常用容器标准 dict/list/tuple/set 之外,collections 提供一组高频专用容器。
| 容器 | 用途 | 关键方法/特性 |
|---|---|---|
namedtuple |
轻量不可变记录 |
_asdict, _replace, _make, _fields
|
deque |
双端队列,两端 O(1) |
appendleft, popleft, maxlen, 线程安全(append/pop) |
Counter |
计数/频次 |
most_common, elements, subtract, +/- 运算 |
defaultdict |
带默认值工厂的 dict | 访问缺失键自动调用工厂 |
ChainMap |
多 dict 逻辑合并 | 不拷贝、查找按序、写只命中第一个 |
OrderedDict |
保序 dict(3.7+ 内置 dict 已保序) |
move_to_end, popitem(last=)
|
from collections import namedtuple, deque, Counter, defaultdict, ChainMap, OrderedDict
# namedtuple:不可变数据记录(比 class 省内存,比 tuple 可读)
Point = namedtuple("Point", ["x", "y"])
p = Point(10, 20)
p.x, p[0] # 10, 10
p._asdict() # {'x': 10, 'y': 20}
p._replace(x=99) # Point(x=99, y=20) 返回新对象(不可变)
Point._fields # ('x', 'y')
Point._make([1, 2]) # Point(x=1, y=2)
# deque:高效队列/栈(list 在头部插入是 O(n),deque 两端都 O(1))
dq = deque(maxlen=3) # 定长:满了自动丢最旧
dq.extend([1, 2, 3, 4]) # deque([2, 3, 4])
dq.appendleft(0)
dq.popleft()
# Counter:统计神器
cnt = Counter("abracadabra")
cnt.most_common(2) # [('a', 5), ('b', 2)]
cnt.elements() # 展开成迭代器(按计数重复)
cnt.subtract("abra") # 减法
Counter("aabb") + Counter("bbcc") # Counter({'b':4,'a':2,'c':2})
# defaultdict:省去 "if key not in dd"
dd = defaultdict(list)
dd["users"].append("Alice") # 不抛 KeyError
dd["score"] # [](默认工厂返回空列表)
# ChainMap:配置层层覆盖(不合并拷贝,查找按顺序)
defaults = {"host": "localhost", "port": 8080}
overrides = {"host": "prod"}
cm = ChainMap(overrides, defaults)
cm["host"] # 'prod'(先查 overrides)
cm["port"] # 8080(回退 defaults)
cm["host"] = "test" # 只写入 overrides(第一个映射)
# OrderedDict:需依赖顺序语义时使用(如 LRU 实现)
od = OrderedDict()
od["a"] = 1; od["b"] = 2
od.move_to_end("a") # 移到末尾
UserDict/UserList/UserString:需要继承并定制 dict/list/str 行为时,继承这些包装类比直接继承内置类型更安全(避免内置类型绕过重写的特殊方法)。
dataclasses
核心概念:@dataclass 是一个类装饰器,自动为你生成 __init__、__repr__、__eq__ 等样板方法,让你用声明式语法写数据类,少敲大量重复代码。
from dataclasses import dataclass, field
@dataclass
class User:
name: str
age: int = 0
tags: list = field(default_factory=list) # 可变默认值必须用 default_factory
active: bool = True
def greet(self):
return f"Hi, {self.name}"
u = User("Alice", 30)
u2 = User("Bob")
print(u) # User(name='Alice', age=30, tags=[], active=True)
print(u == u2) # False(自动 __eq__)
u.tags.append("vip") # tags 是独立列表
致命陷阱:默认参数若直接写
tags: list = [],所有实例会共享同一个列表(和函数默认参数陷阱一致)。必须用field(default_factory=list)。
from dataclasses import dataclass, field, asdict, astuple
@dataclass(frozen=True) # 不可变:禁止赋值,自动生成 __hash__
class FrozenPoint:
x: float
y: float
@dataclass(order=True) # 自动生成 __lt__/__le__/__gt__/__ge__(按字段顺序比较)
class Rank:
score: int = field(compare=True)
name: str = field(compare=False) # 比较时忽略 name
@dataclass
class Config:
items: list = field(default_factory=list, repr=False) # repr 中隐藏
_cache: dict = field(default_factory=dict, init=False) # 不在 __init__ 中
def __post_init__(self):
# __init__ 之后自动调用:做额外校验/推导
if not self.items:
raise ValueError("items 不能为空")
c = Config([1, 2])
asdict(c) # {'items': [1, 2], '_cache': {}} 转成普通 dict
astuple(c) # ([1, 2], {})
field 常用参数:default(不可变默认值)、default_factory(工厂函数,处理可变默认值)、init(是否进 __init__)、repr(是否进 __repr__)、compare(是否参与比较)、kw_only(Python 3.10+,强制关键字参数)。
slots 结合 & 与 namedtuple 对比@dataclass(slots=True) # Python 3.10+:自动加 __slots__,省内存
class Efficient:
x: int
y: int
| 维度 | namedtuple |
dataclass |
|---|---|---|
| 可变性 | 不可变、可哈希 | 默认可变(frozen=True 才可哈希) |
| 继承 | 不支持字段继承 | 支持字段继承 |
| 默认值 | 仅尾部可选 | 任意字段默认值、灵活 |
| 内存 | 最小(C 级元组) | 普通对象;slots=True 后可接近 |
| 方法 | 需 _replace/_asdict
|
可自由定义方法 |
选型:仅做简单不可变记录 →
namedtuple;需要可变性/默认值/继承/校验/序列化 →dataclass;需要运行时校验与序列化 → 考虑pydantic。
核心概念:Python 是动态类型语言,但类型注解(type hints)让你标注变量/参数/返回值的预期类型。注解运行时被忽略(不强制、不改行为),主要服务于静态检查器(mypy/pyright)、IDE 提示、和文档。
from typing import Optional, Union, List, Dict, Tuple, Callable, Any
def greet(name: str) -> str:
return f"Hello, {name}"
def get_config(key: str) -> Optional[int]: # int | None
return config.get(key)
def process(data: Union[int, str]) -> str: # int | str
return str(data)
users: List[str] = ["Alice", "Bob"] # 注意:仅注解,运行时不检查
scores: Dict[str, int] = {}
handler: Callable[[int, int], str] = lambda a, b: f"{a}+{b}"
Python 3.9+ 推荐:直接用内置泛型
list[str]、dict[str, int]、tuple[float, float]、X | None、X | Y,无需从typing导入容器类型。上面写法是为兼容旧版本。
from typing import TypeVar, Generic
T = TypeVar("T")
U = TypeVar("U", int, str) # 约束:只能是 int 或 str
B = TypeVar("B", bound="Sequence") # 上界:必须是 Sequence 的子类
class Stack(Generic[T]):
def __init__(self):
self._items: list[T] = []
def push(self, item: T) -> None:
self._items.append(item)
def pop(self) -> T:
return self._items.pop()
s = Stack[int]()
s.push(1)
Protocol / TypedDict / Literal
from typing import Protocol, TypedDict, Literal, NewType, cast, overload
# Protocol:鸭子类型的静态表达(运行时零开销,不需继承)
class Flyable(Protocol):
def fly(self) -> None: ...
def takeoff(obj: Flyable) -> None:
obj.fly() # mypy 只要对象有 fly() 就通过
# TypedDict:字典的结构化注解(字段名+类型)
class UserInfo(TypedDict):
name: str
age: int
u: UserInfo = {"name": "Alice", "age": 30}
# Literal:精确字面值集合
def set_mode(mode: Literal["read", "write", "append"]) -> None: ...
# NewType:创建"名义上不同"的类型(运行时是同一类型,仅静态区分)
UserId = NewType("UserId", int)
uid = UserId(42)
# @overload:为同一函数声明多种签名(实现里仍是单个函数)
@overload
def to_str(x: int) -> str: ...
@overload
def to_str(x: bytes) -> str: ...
def to_str(x):
return x.decode() if isinstance(x, bytes) else str(x)
import typing
def f(a: "list[int]", b: int = 0): ...
typing.get_type_hints(f) # 运行时读取注解(解析前向/字符串引用)
# 输出如 {'a': list[int], 'b': int} —— 注解被保留在 __annotations__
要点:注解存在
obj.__annotations__,但不做强制。要真正检查,需mypy your_file.py或编辑器实时检查。PEP 563(3.10+ 默认)让注解延迟求值(from __future__ import annotations在所有版本可用),避免循环引用与启动开销。cast(T, val)只告诉检查器"当成 T",运行时不改变val。
核心概念:Python 有三种并发武器,选择取决于任务是 CPU 密集还是 IO 密集,根因是 GIL(下节详述)。
import threading, time
# threading —— IO 密集型(sleep/网络/磁盘等待会释放 GIL,其他线程能跑)
def io_work():
time.sleep(1) # 释放 GIL
threads = [threading.Thread(target=io_work) for _ in range(10)]
[t.start() for t in threads]; [t.join() for t in threads] # 约 1s
# multiprocessing —— CPU 密集型(多进程绕过 GIL,真并行)
import multiprocessing as mp
def compute():
return sum(i*i for i in range(10**7))
with mp.Pool() as pool:
pool.map(compute, range(4)) # 4 进程并行,约 1/4 时间
# asyncio —— 高并发 IO(单线程,协程协作式,开销最低)
import asyncio, aiohttp
async def fetch(url):
async with aiohttp.ClientSession() as s:
async with s.get(url) as r:
return await r.text()
| 场景 | 推荐 | 原因 |
|---|---|---|
| CPU 密集(计算) | multiprocessing |
多进程真并行,绕过 GIL |
| IO 密集(少量) | threading |
等待 IO 时释放 GIL,共享内存简单 |
| IO 密集(高并发 1000+) | asyncio |
协程切换开销极低,单线程撑海量连接 |
| 混合 | 进程池 + 异步 | CPU 与 IO 分离,进程跑计算、异步跑 IO |
| 需要共享状态 + 简单 |
threading + Lock
|
但要注意竞态 |
import threading
counter = 0
lock = threading.Lock()
def inc():
global counter
for _ in range(100000):
with lock: # 不加锁会出现竞态,结果 < 100000*N
counter += 1
ts = [threading.Thread(target=inc) for _ in range(4)]
[t.start() for t in ts]; [t.join() for t in ts]
# 有锁时 counter == 400000
asynciovsthreading心智模型:threading是抢占式(OS 决定切换点,需锁保护共享数据);asyncio是协作式(只在await处切换,单线程无竞态,但不写await就占着循环)。asyncio的优势在"海量连接"而非"单连接更快"。
Python(CPython)用引用计数为主、分代 GC 兜底:
import sys, gc
a = [1, 2, 3]
print(sys.getrefcount(a)) # 查看当前引用计数(调用本身会 +1)
print(gc.get_threshold()) # (700, 10, 10) 各代阈值
print(gc.get_stats()) # 各代回收统计
gc.collect() # 手动触发全量回收
weakref — 打破循环引用弱引用不增加引用计数,常用于缓存、观察者,避免对象被"本应回收"的引用钉住。
import weakref
class Node:
def __init__(self, val):
self.val = val
self.parent = None
n = Node(1)
r = weakref.ref(n) # 弱引用
print(r() is n) # True(对象还活着)
del n
print(r() is None) # True(对象被回收,弱引用失效)
定义:CPython 同一进程同一时刻只有一个线程能执行 Python 字节码。哪怕多核,多线程 Python 代码在 CPU 密集任务上无法真正并行。
multiprocessing(多进程,真并行)、C 扩展(NumPy 等释放 GIL 做计算)、nogil 构建(PEP 703,Python 3.13+ 实验性"free-threaded"构建可禁用 GIL)。id() / is 与对象身份a = [1, 2]; b = [1, 2]
print(a is b) # False(不同对象)
print(a == b) # True(值相等)
print(id(a)) # 对象内存地址(进程内唯一标识)
is比较身份(是否同一对象),==比较值。常见误用:x is None(正确,单例)、x == None(不推荐)。注意小整数(-5~256)和短字符串会被 ** intern(缓存)**,所以a = 256; b = 256; a is b为True——这是实现细节,不要依赖。
int/str/tuple/frozenset 不可变:看起来"修改"其实是创建新对象。memoryview / bytearray:零拷贝地切片/操作缓冲,处理大二进制数据时省内存。核心原则:先测量,再优化("过早优化是万恶之源")。用工具定位真正的瓶颈,而非凭直觉。
import timeit, cProfile, pstats
# timeit:微基准(多次运行取均值,自动屏蔽噪声)
timeit.timeit("sum(range(1000))", number=10000)
# cProfile:函数级剖析(谁最耗时、调用几次)
cProfile.run("sum(range(10**7))", sort="tottime")
# CLI: python -m cProfile -s tottime your_script.py
# 更细:line_profiler(逐行)、memory_profiler(逐行内存)、py-spy(采样,生产无侵入)
cProfile 字段解读:tottime=函数自身耗时(不含子调用)、cumtime=含子调用总耗时、ncalls=调用次数、percall=单次平均。优先优化 tottime 高且 ncalls 大的函数。
import math
# 1) 局部变量缓存:循环内反复访问的全局/属性函数,先绑定到局部变量
# 避免每次都走 LEGB 查找链
_abs = abs
for x in range(10**6):
_abs(x)
# 2) 列表推导 vs 循环 append:列表推导在 C 层循环,无每次 append 的属性查找
data = [x*2 for x in range(1000)] # 快
# 3) " ".join() vs += 拼接:join 预计算总长一次分配;+= 字符串不可变,反复拷贝
parts = ["a", "b", "c"]
result = "".join(parts) # 快一个数量级
# 4) set/dict 查找 O(1) :成员检查用 set 而非 list
haystack = set(big_list)
if item in haystack: ... # 比 "item in big_list" 快 N 倍
# 5) @lru_cache:缓存重复/递归计算
from functools import lru_cache
@lru_cache(maxsize=256)
def expensive(n): ...
# 6) __slots__:大量实例时省内存(见第 4 节)
# 7) 用内置与 C 扩展:sum/min/max/sorted、NumPy 向量化远快于手写 Python 循环
total = sum(values) # 而非手写循环累加
| 手段 | 原理 | 适用 |
|---|---|---|
| 列表推导 | C 层循环,无 append 属性查找 | 数据转换 |
str.join |
预计算总长,单次分配 | 字符串拼接 |
set/dict 查找 |
哈希表 O(1) | 成员检查 |
| 局部变量缓存 | 避免 LEGB 查找链 | 循环内函数调用 |
__slots__ |
省 __dict__,属性访问略快 |
大量实例 |
lru_cache |
记忆化,避免重复计算 | 递归/频繁调用 |
| 内置/NumPy | C 实现,向量化 | 数值计算 |
警告:大多数"优化"代价是可读性。
+拼接一行字符串、普通for累加小幅数据,在绝大多数业务代码里完全够用。只针对cProfile指出的热点动手。必要时把热路径用numba/Cython/C 扩展重写,而非在 Python 层抠细节。
按由内到外、由基础到框架的顺序,建议分 9 步:
functools(第 5、6 节)——最常用、最立竿见影。闭包是装饰器的基础,@wraps、带参装饰器、缓存/重试是日常必备。itertools(第 7、8 节)——理解 for 的底层,掌握惰性计算与组合工具,写高效数据流。asyncio(第 11 节)——从手写协程过渡到原生异步;吃透 Task/Queue/gather/run_in_executor,避免阻塞陷阱。super() + 元类 + ABC(第 12、13、14 节)——框架/ORM 的底层。以"理解其机制"为主,业务代码慎用元类,super() 单继承用法要熟练。collections + dataclasses(第 15、16 节)——选对数据结构、用声明式减少样板代码,少写字多做事。typing(第 17 节)——提升可读性与可维护性,配合 mypy/Pylance 做静态检查,大型项目收益显著。cProfile/timeit 量化收益。一句话心智模型:Python 里"类是
type造出的对象(元类),方法/字段是描述符,协程是能挂起/恢复的调度单元(asyncio),__slots__是内存优化开关"。把这四点串起来,全部高级特性就不再是散点,而是一棵自洽的树。
Python 高级知识点完全指南(Markdown 版)| 2026-07 | 21 节 · 40+ 代码示例 · 9 幅图示 · 覆盖对象模型 / 函数式 / 元编程 / 执行模型 / 并发 / 数据结构 / 性能 | 建议在 Python 3.10+ 环境运行全部示例