Python 深入

概述

Python 以简洁优雅著称，但其内部机制远比表面复杂。本文深入探讨 Python 的高级特性、内存模型、并发机制和性能优化方法，帮助从"会用"提升到"精通"。

---

1. 生成器与 yield

1.1 生成器函数

生成器通过 yield 关键字实现惰性求值，每次产出一个值后暂停执行：

def fibonacci():
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b

# 使用
gen = fibonacci()
for _ in range(10):
    print(next(gen))  # 0, 1, 1, 2, 3, 5, 8, 13, 21, 34

1.2 生成器表达式

# 列表推导式：立即生成所有元素，占用内存
squares_list = [x**2 for x in range(10**6)]  # ~8MB

# 生成器表达式：惰性求值，几乎不占内存
squares_gen = (x**2 for x in range(10**6))   # ~128B

1.3 yield from

yield from 用于委托给子生成器：

def chain(*iterables):
    for it in iterables:
        yield from it

list(chain([1,2], [3,4], [5,6]))  # [1, 2, 3, 4, 5, 6]

1.4 协程式生成器

生成器也可以接收值（协程的前身）：

def accumulator():
    total = 0
    while True:
        value = yield total
        total += value

acc = accumulator()
next(acc)          # 启动，返回 0
acc.send(10)       # 返回 10
acc.send(20)       # 返回 30
acc.send(5)        # 返回 35

---

2. 装饰器（Decorators）

装饰器是修改函数或类行为的高阶函数。

2.1 基本装饰器

import functools
import time

def timer(func):
    @functools.wraps(func)  # 保留原函数的元信息
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"{func.__name__} took {elapsed:.4f}s")
        return result
    return wrapper

@timer
def slow_function():
    time.sleep(1)

2.2 带参数的装饰器

def retry(max_attempts=3, delay=1):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    time.sleep(delay)
        return wrapper
    return decorator

@retry(max_attempts=5, delay=0.5)
def unreliable_api_call():
    ...

2.3 类装饰器

def singleton(cls):
    instances = {}
    @functools.wraps(cls)
    def get_instance(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]
    return get_instance

@singleton
class Database:
    def __init__(self):
        print("Connecting...")

---

3. 上下文管理器（Context Managers）

3.1 基于类的实现

class FileManager:
    def __init__(self, filename, mode):
        self.filename = filename
        self.mode = mode
        self.file = None

    def __enter__(self):
        self.file = open(self.filename, self.mode)
        return self.file

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.file.close()
        return False  # 不抑制异常

with FileManager("test.txt", "w") as f:
    f.write("Hello")

3.2 基于 contextlib 的实现

from contextlib import contextmanager

@contextmanager
def timer_context(name):
    start = time.perf_counter()
    try:
        yield  # 此处执行 with 块中的代码
    finally:
        elapsed = time.perf_counter() - start
        print(f"{name}: {elapsed:.4f}s")

with timer_context("Training"):
    model.fit(X, y)

---

4. 元类（Metaclasses）

元类是类的类——控制类本身的创建过程。

class SingletonMeta(type):
    _instances = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super().__call__(*args, **kwargs)
        return cls._instances[cls]

class Database(metaclass=SingletonMeta):
    def __init__(self):
        self.connection = "connected"

# Database() is Database()  => True

元类的实际应用：

• Django ORM：通过元类自动将类属性映射为数据库字段
• ABC：abc.ABCMeta 实现抽象基类
• dataclass：@dataclass 装饰器内部使用元类技术

---

5. GIL（全局解释器锁）

5.1 什么是 GIL

GIL（Global Interpreter Lock）是 CPython 中的一个互斥锁，确保同一时刻只有一个线程执行 Python 字节码。

线程1: ████----████----████
线程2: ----████----████----
         ↑ 同一时刻只有一个线程执行

5.2 GIL 的影响

场景	影响	建议
CPU 密集型	多线程无法利用多核	使用 multiprocessing
I/O 密集型	GIL 在 I/O 等待时释放	多线程有效
C 扩展	可手动释放 GIL	NumPy 等已优化

5.3 绕过 GIL

# 方案1：multiprocessing
from multiprocessing import Pool
with Pool(4) as p:
    results = p.map(cpu_intensive_func, data)

# 方案2：concurrent.futures
from concurrent.futures import ProcessPoolExecutor
with ProcessPoolExecutor(max_workers=4) as executor:
    futures = [executor.submit(func, arg) for arg in args]

Python 3.13+

PEP 703 提议可选的 free-threaded CPython（无 GIL 模式），Python 3.13 开始实验性支持。

---

6. asyncio 与协程

6.1 async/await 基础

import asyncio

async def fetch_data(url):
    print(f"Fetching {url}...")
    await asyncio.sleep(1)  # 模拟网络请求
    return f"Data from {url}"

async def main():
    # 并发执行多个协程
    tasks = [
        fetch_data("url1"),
        fetch_data("url2"),
        fetch_data("url3"),
    ]
    results = await asyncio.gather(*tasks)
    print(results)

asyncio.run(main())  # 总共约 1 秒（而非 3 秒）

6.2 事件循环

┌─────────────────────────────────┐
│         Event Loop              │
│  ┌────┐  ┌────┐  ┌────┐        │
│  │任务1│  │任务2│  │任务3│        │
│  └──┬─┘  └──┬─┘  └──┬─┘        │
│     │ await  │       │          │
│     ▼        ▼       ▼          │
│  [挂起]   [运行]  [等待]         │
│     │        │       │          │
│     ▼        ▼       ▼          │
│  [恢复]   [挂起]  [运行]         │
└─────────────────────────────────┘

6.3 异步迭代器和上下文管理器

async def async_range(n):
    for i in range(n):
        await asyncio.sleep(0.1)
        yield i

async def main():
    async for i in async_range(5):
        print(i)

---

7. 内存模型

7.1 引用计数 + 垃圾回收

import sys

a = [1, 2, 3]
print(sys.getrefcount(a))  # 2（a + getrefcount 参数）

b = a
print(sys.getrefcount(a))  # 3

del b
print(sys.getrefcount(a))  # 2

循环引用处理：

import gc

class Node:
    def __init__(self):
        self.ref = None

a = Node()
b = Node()
a.ref = b
b.ref = a  # 循环引用

del a, b
# 引用计数不会降到 0，需要 gc 模块的标记-清除
gc.collect()  # 回收循环引用

7.2 Python 对象内存布局

每个 Python 对象至少包含：

• 引用计数（ob_refcnt）
• 类型指针（ob_type）
• 对象数据

一个 int 对象 (28 bytes):
┌──────────────┐
│ ob_refcnt: 8B│
│ ob_type:   8B│
│ ob_size:   4B│
│ ob_digit:  4B│  ← 实际值
│ padding:   4B│
└──────────────┘

7.3 小整数缓存和字符串驻留

# 小整数缓存 [-5, 256]
a = 256
b = 256
a is b  # True（同一对象）

a = 257
b = 257
a is b  # False（不同对象，在交互模式下）

# 字符串驻留
a = "hello"
b = "hello"
a is b  # True（驻留优化）

---

8. Type Hints 类型标注

8.1 基本用法

from typing import Optional, Union, List, Dict, Tuple, Callable

def greet(name: str) -> str:
    return f"Hello, {name}"

def process(data: List[int], 
            callback: Callable[[int], bool],
            config: Optional[Dict[str, str]] = None) -> Tuple[int, ...]:
    ...

8.2 泛型和 Protocol

from typing import TypeVar, Generic, Protocol

T = TypeVar('T')

class Stack(Generic[T]):
    def __init__(self) -> None:
        self._items: List[T] = []

    def push(self, item: T) -> None:
        self._items.append(item)

    def pop(self) -> T:
        return self._items.pop()

# 结构化子类型（Duck Typing 的类型标注版本）
class Drawable(Protocol):
    def draw(self) -> None: ...

8.3 工具链

• mypy：静态类型检查器
• pyright：Microsoft 开发，VS Code 集成
• pydantic：运行时类型验证

---

9. 性能优化

9.1 性能分析

# cProfile
import cProfile
cProfile.run('my_function()')

# line_profiler
@profile
def slow_function():
    ...
# kernprof -l -v script.py

9.2 加速方案

方案	原理	加速倍数	适用场景
NumPy	C 实现的向量化运算	10-100x	数值计算
Cython	Python → C 编译	10-100x	计算密集循环
Numba	LLVM JIT 编译	10-200x	数值循环
PyPy	整体 JIT 编译	2-10x	通用 Python
multiprocessing	多进程绕过 GIL	核数倍	CPU 密集

# Numba 示例
from numba import jit

@jit(nopython=True)
def monte_carlo_pi(n):
    count = 0
    for i in range(n):
        x = random.random()
        y = random.random()
        if x**2 + y**2 <= 1.0:
            count += 1
    return 4.0 * count / n

---

10. 数据模型（Dunder Methods）

Python 的数据模型通过特殊方法（双下划线方法）让自定义对象表现得像内置类型。

10.1 核心特殊方法

class Vector:
    def __init__(self, x, y):
        self.x = x
        self.y = y

    def __repr__(self):
        return f"Vector({self.x}, {self.y})"

    def __str__(self):
        return f"({self.x}, {self.y})"

    def __add__(self, other):
        return Vector(self.x + other.x, self.y + other.y)

    def __abs__(self):
        return (self.x**2 + self.y**2)**0.5

    def __bool__(self):
        return abs(self) != 0

    def __eq__(self, other):
        return self.x == other.x and self.y == other.y

    def __hash__(self):
        return hash((self.x, self.y))

    def __len__(self):
        return 2

    def __getitem__(self, index):
        if index == 0: return self.x
        if index == 1: return self.y
        raise IndexError

10.2 描述符协议

class Validator:
    def __set_name__(self, owner, name):
        self.name = name

    def __get__(self, obj, objtype=None):
        return getattr(obj, f'_{self.name}', None)

    def __set__(self, obj, value):
        if not isinstance(value, (int, float)):
            raise TypeError(f"{self.name} must be a number")
        if value < 0:
            raise ValueError(f"{self.name} must be >= 0")
        setattr(obj, f'_{self.name}', value)

class Product:
    price = Validator()
    quantity = Validator()

---

参考资料

• "Fluent Python" - Luciano Ramalho
• "CPython Internals" - Anthony Shaw
• Python 官方文档：Data Model、asyncio

---

评论