# Python高级进阶完全教程

如果你已经掌握了Python的基础语法，能够熟练编写脚本解决日常工作问题，那么恭喜你，你已经迈过了Python学习的第一道门槛。但真正的挑战才刚刚开始——如何写出更优雅、更高效、更可靠的Python代码？如何理解Python那些"神奇"的背后机制？如何应对高并发、大数据量的复杂场景？

这篇教程正是为你准备的。我们将深入Python的核心机制，探索那些让你成为真正"高级"程序员的奥秘。无论你是希望提升代码质量，还是准备架构复杂系统，亦或是为了面试深造，这篇教程都值得你反复研读。

这部分不需要你一个个的把代码敲一遍，重在理解，力争把每一行代码逻辑都理解透彻，你距离大神就不远啦。

让我们开始这段进阶之旅吧。

# 第一部分：基石与范式——编写"Pythonic"的高级代码

# 第1章函数式编程深度实践

Python并非纯函数式语言，但它从函数式编程范式中汲取了丰富的营养。掌握这些特性，能让你的代码更简洁、更具表现力。

# 1.1 函数作为一等公民

在Python中，函数是"一等公民"(first-class citizen)。这意味着函数可以像整数、字符串一样被赋值给变量、作为参数传递、作为返回值返回。

# 1.1.1 函数即对象

我们先看一个最基本的例子：

# 函数可以赋值给变量
def greet(name):
    return f"Hello, {name}!"

say_hello = greet  # 函数赋值给变量
print(say_hello("Alice"))  # 输出: Hello, Alice!

# 函数可以作为容器元素
funcs = [greet, str.upper, len]
for func in funcs:
    print(func("hello"))  
# 输出:
# Hello, hello!
# HELLO
# 5

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

这个看似简单的特性，却是高阶函数的基石。

# 1.1.2 高阶函数：map、filter、reduce

高阶函数是指接受函数作为参数或返回函数的函数。Python内置了几个经典的高阶函数：

# map：对序列每个元素应用函数
numbers = [1, 2, 3, 4, 5]
squared = list(map(lambda x: x**2, numbers))
print(squared)  # [1, 4, 9, 16, 25]

# 更Pythonic的方式：列表推导式
squared_py = [x**2 for x in numbers]
print(squared_py)  # [1, 4, 9, 16, 25]

# filter：筛选满足条件的元素
even = list(filter(lambda x: x % 2 == 0, numbers))
print(even)  # [2, 4]

# 列表推导式版本
even_py = [x for x in numbers if x % 2 == 0]

# reduce：累积运算
from functools import reduce
product = reduce(lambda x, y: x * y, numbers)  # 计算乘积
print(product)  # 120

# reduce的应用：找出最大值
max_num = reduce(lambda x, y: x if x > y else y, numbers)
print(max_num)  # 5

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

深入理解： map和filter返回的是迭代器(Python 3+)，这意味着它们是惰性求值的。只有当你迭代它们或转换为列表时，才会真正执行计算。这在处理大数据集时可以节省内存。

# 1.1.3 闭包：函数工厂

闭包(closure)是指引用了自由变量的函数。这个被引用的自由变量将和这个函数一同存在，即使已经离开了创造它的环境。

def make_multiplier(factor):
    """创建一个将数字乘以指定因子的函数"""
    def multiplier(x):
        return x * factor
    return multiplier

double = make_multiplier(2)
triple = make_multiplier(3)

print(double(5))  # 10
print(triple(5))  # 15

# 检查闭包中捕获的变量
print(double.__closure__[0].cell_contents)  # 2
print(triple.__closure__[0].cell_contents)  # 3

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

闭包的工作原理： 当内部函数multiplier引用了外部函数make_multiplier的变量factor时，Python会将这个变量"打包"到内部函数中，形成一个闭包。即使make_multiplier执行完毕，factor依然存在于multiplier的__closure__属性中。

闭包的经典应用：装饰器、回调函数、延迟计算。

# 闭包实现计数器
def create_counter():
    count = 0
    def counter():
        nonlocal count  # 声明count不是局部变量
        count += 1
        return count
    return counter

c1 = create_counter()
print(c1())  # 1
print(c1())  # 2
print(c1())  # 3

c2 = create_counter()
print(c2())  # 1 (独立的count)

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

注意nonlocal关键字的使用。在Python中，如果只是读取外部变量，不需要nonlocal；但若要修改外部变量(重新绑定)，则必须声明nonlocal。

# 1.2 从闭包到装饰器

装饰器是闭包最优雅的应用之一。它允许我们在不修改原函数代码的情况下，动态地增强函数的功能。

# 1.2.1 装饰器基础

# 最简单的装饰器：记录函数执行时间
import time

def timer(func):
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        end = time.time()
        print(f"{func.__name__} took {end - start:.4f} seconds")
        return result
    return wrapper

# 使用装饰器
@timer
def slow_function():
    time.sleep(1)
    return "Done"

slow_function()  # 输出: slow_function took 1.0012 seconds

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

装饰器的工作原理： @timer等价于slow_function = timer(slow_function)。所以slow_function现在指向了wrapper函数。

# 1.2.2 保留函数元数据

上面的装饰器有一个问题：原函数的元数据(如__name__、__doc__)被wrapper覆盖了。

print(slow_function.__name__)  # 输出: wrapper (应该是slow_function)

解决方案：使用functools.wraps

from functools import wraps

def timer(func):
    @wraps(func)  # 这个装饰器会复制func的元数据到wrapper
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        end = time.time()
        print(f"{func.__name__} took {end - start:.4f} seconds")
        return result
    return wrapper

@timer
def slow_function():
    """This is a slow function."""
    time.sleep(1)
    return "Done"

print(slow_function.__name__)  # 输出: slow_function
print(slow_function.__doc__)   # 输出: This is a slow function.

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

最佳实践： 在自定义装饰器时，始终使用@wraps装饰内部函数，这能保持调试信息的完整性。

# 1.2.3 参数化装饰器

有时我们需要给装饰器传递参数，这需要在装饰器外层再包一层：

def repeat(n):
    """重复执行函数n次的装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            results = []
            for i in range(n):
                results.append(func(*args, **kwargs))
            return results
        return wrapper
    return decorator

@repeat(3)
def say(message):
    return f"Saying: {message}"

print(say("Hello"))  
# 输出: ['Saying: Hello', 'Saying: Hello', 'Saying: Hello']

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

理解这个三层嵌套结构：

repeat(n)：接收装饰器参数，返回真正的装饰器decorator
decorator(func)：接收被装饰的函数，返回包装函数wrapper
wrapper(*args, **kwargs)：实现增强功能

# 1.2.4 基于类的装饰器

装饰器也可以是类，只要实现__call__方法：

class CountCalls:
    def __init__(self, func):
        self.func = func
        self.count = 0
        
    def __call__(self, *args, **kwargs):
        self.count += 1
        print(f"Call {self.count} of {self.func.__name__}")
        return self.func(*args, **kwargs)

@CountCalls
def hello():
    print("Hello!")

hello()  # Call 1 of hello / Hello!
hello()  # Call 2 of hello / Hello!

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

类装饰器的优势在于可以轻松维护状态(如这里的count)，且代码结构更清晰。

# 1.2.5 装饰器实战：实现缓存

functools.lru_cache是一个非常有用的内置装饰器，它实现了最近最少使用(LRU)缓存：

from functools import lru_cache

@lru_cache(maxsize=128)
def fibonacci(n):
    """递归计算斐波那契数列"""
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

# 第一次调用，计算所有值
print(fibonacci(30))  # 很快，但还有优化空间

# 第二次调用，瞬间返回(缓存命中)
print(fibonacci(30))  # 几乎不花时间

# 查看缓存信息
print(fibonacci.cache_info())  # CacheInfo(hits=29, misses=31, maxsize=128, currsize=31)

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

如果不使用缓存，计算fibonacci(30)需要进行数百万次递归调用；使用缓存后，每个n只计算一次。

自己实现一个简单的缓存装饰器：

def memoize(func):
    cache = {}
    
    @wraps(func)
    def wrapper(*args):
        if args not in cache:
            cache[args] = func(*args)
        return cache[args]
    return wrapper

@memoize
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

print(fibonacci(30))  # 832040

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

这个简单版本没有限制缓存大小，如果函数参数很多，可能会导致内存溢出。lru_cache则通过maxsize参数解决了这个问题。

# 1.3 从生成器到协程

生成器(generator)是Python中实现惰性求值的核心机制。理解生成器，是理解异步编程的基础。

# 1.3.1 生成器基础

def count_up_to(n):
    """生成器函数：产生从1到n的数字"""
    i = 1
    while i <= n:
        yield i
        i += 1

# 创建生成器对象
counter = count_up_to(5)

# 逐个获取值
print(next(counter))  # 1
print(next(counter))  # 2
print(next(counter))  # 3
print(next(counter))  # 4
print(next(counter))  # 5
# print(next(counter))  # StopIteration 异常

# 或者使用for循环(自动处理StopIteration)
for num in count_up_to(5):
    print(num)  # 1 2 3 4 5

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

生成器的工作原理： 当函数包含yield关键字时，它就变成了生成器函数。调用生成器函数不会立即执行函数体，而是返回一个生成器对象。每次调用next()，函数体执行到下一个yield，返回yield后的值，并暂停。

# 1.3.2 生成器表达式

就像列表推导式一样，Python也支持生成器表达式，只需将方括号换成圆括号：

import sys 
# 列表推导式：立即创建所有元素
list_squares = [x**2 for x in range(100)]
print(list_squares)  # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
print(sys.getsizeof(list_squares))  # 占用内存较大

# 生成器表达式：惰性求值
gen_squares = (x**2 for x in range(100))
print(gen_squares)  # <generator object <genexpr> at 0x...>
print(sys.getsizeof(gen_squares))  # 占用内存很小

# 逐个获取
for square in gen_squares:
    print(square, end=' ')  # 0 1 4 9 16 25 36 49 64 81

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

生成器在处理大数据集时优势明显：

# 计算1亿个数的平方和
# 列表版本(内存不足！)
# sum([x**2 for x in range(100000000)])

# 生成器版本(内存友好)
sum(x**2 for x in range(100000000))  # 可以运行

1
2
3
4
5
6

# 1.3.3 生成器的进阶用法：send()、throw()、close()

生成器不仅仅是单向产生数据，它还可以接收数据，形成双向通信。

send()方法：向生成器发送值

def echo():
    """回显生成器：接收值并打印"""
    while True:
        received = yield
        print(f"Received: {received}")

e = echo()
next(e)  # 启动生成器，执行到第一个yield
e.send("Hello")  # Received: Hello
e.send("World")  # Received: World
e.close()  # 关闭生成器

1
2
3
4
5
6
7
8
9
10
11

注意：生成器启动后，第一个yield之前是不能接收值的。所以必须先调用next(e)或e.send(None)来启动。

更实用的例子：实现一个简易的累加器

def accumulator():
    total = 0
    while True:
        value = yield total  # yield返回当前总和，并接收新值
        if value is None:
            continue
        total += value

acc = accumulator()
next(acc)  # 启动

print(acc.send(10))  # 10
print(acc.send(5))   # 15
print(acc.send(3))   # 18
acc.close()

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

throw()方法：向生成器抛出异常

def generator():
    try:
        yield 1
        yield 2
        yield 3
    except ValueError as e:
        print(f"Caught: {e}")
        yield 4

g = generator()
print(next(g))  # 1
print(g.throw(ValueError("Something wrong")))  # Caught: Something wrong / 4

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

close()方法：关闭生成器，会在生成器内部引发GeneratorExit异常。

这些特性使得生成器可以作为协程使用，为后面学习async/await打下基础。

# 1.3.4 yield from：生成器委托

yield from是Python 3.3引入的语法，用于在一个生成器中调用另一个生成器，简化了嵌套生成器的操作。

def sub_generator():
    yield 1
    yield 2
    yield 3

def main_generator():
    yield 'start'
    yield from sub_generator()  # 委托给子生成器
    yield 'end'

for value in main_generator():
    print(value, end=' ')  # start 1 2 3 end
print()

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

没有yield from的话，我们需要手动迭代子生成器：

def main_generator_manual():
    yield 'start'
    for v in sub_generator():
        yield v
    yield 'end'

1
2
3
4
5

yield from不仅简化了代码，还能正确处理send()、throw()等操作，将调用者与子生成器直接连接起来。

实战：处理嵌套的可迭代对象

def flatten(nested):
    """展开嵌套的可迭代对象"""
    try:
        for sub in nested:
            if isinstance(sub, (list, tuple)):
                yield from flatten(sub)  # 递归展开子列表
            else:
                yield sub
    except TypeError:
        # 如果不是可迭代对象，直接yield
        yield nested

nested_list = [1, [2, [3, 4], 5], 6]
for item in flatten(nested_list):
    print(item, end=' ')  # 1 2 3 4 5 6

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

# 1.3.5 生成器实战：处理大文件

这是生成器最经典的应用场景之一。假设我们有一个300GB的日志文件，无法一次性读入内存：

def read_large_file(file_path, chunk_size=4096):
    """生成器：逐块读取大文件"""
    with open(file_path, 'r', encoding='utf-8') as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            yield chunk

def process_lines(file_path):
    """生成器：逐行处理，处理跨块的行"""
    remainder = ''
    for chunk in read_large_file(file_path):
        lines = (remainder + chunk).split('\n')
        remainder = lines[-1]
        for line in lines[:-1]:
            yield line
    if remainder:
        yield remainder

# 使用示例：统计包含"ERROR"的行数
error_count = 0
for line in process_lines('huge_log.txt'):
    if 'ERROR' in line:
        error_count += 1
print(f"Found {error_count} errors")

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

这个模式保证了在任何时候，只有一小部分数据驻留在内存中。

# 1.4 本章思考与实战

思考题： 在什么场景下应该使用生成器而不是列表？内存使用和性能之间如何权衡？
实战任务： 编写一个装饰器@retry(max_attempts=3, delay=1)，当被装饰的函数抛出异常时，自动重试指定次数，每次重试间隔delay秒。
进阶挑战： 实现一个管道式的数据处理框架，使用生成器连接多个处理阶段，类似于Unix管道cat file | grep error | wc -l。

# 第2章面向对象系统定制与协议

Python的面向对象系统非常灵活，它不仅支持传统的继承和多态，还通过一系列特殊方法(魔术方法)让你可以深度定制类的行为。

# 2.1 理解Python的数据模型

Python的数据模型(data model)是理解这门语言的核心。它定义了一系列特殊方法(以双下划线开头和结尾)，这些方法让你可以赋予自定义类与内置类型相同的行为。

# 2.1.1 对象表示相关的特殊方法

class Person:
    def __init__(self, name, age):
        self.name = name
        self.age = age
    
    # __repr__：面向开发者的字符串表示
    def __repr__(self):
        return f"Person('{self.name}', {self.age})"
    
    # __str__：面向用户的字符串表示
    def __str__(self):
        return f"{self.name} ({self.age} years old)"
    
    # __bytes__：转换为字节串
    def __bytes__(self):
        return f"{self.name}:{self.age}".encode('utf-8')

p = Person("Alice", 30)
print(repr(p))   # Person('Alice', 30)
print(str(p))    # Alice (30 years old)
print(bytes(p))  # b'Alice:30'

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

关键区别： __repr__应该提供无歧义的信息，通常可以用来重新创建对象；__str__更注重可读性。如果没有定义__str__，Python会使用__repr__作为后备。

# 2.1.2 容器类型模拟

通过实现__getitem__、__setitem__等方法，你可以让自定义类表现得像列表或字典：

class SimpleDict:
    def __init__(self):
        self._data = {}
    
    # 获取元素：obj[key]
    def __getitem__(self, key):
        print(f"Getting {key}")
        return self._data[key]
    
    # 设置元素：obj[key] = value
    def __setitem__(self, key, value):
        print(f"Setting {key} = {value}")
        self._data[key] = value
    
    # 删除元素：del obj[key]
    def __delitem__(self, key):
        print(f"Deleting {key}")
        del self._data[key]
    
    # 获取长度：len(obj)
    def __len__(self):
        return len(self._data)
    
    # 支持in操作符：key in obj
    def __contains__(self, key):
        return key in self._data
    
    # 支持迭代：for key in obj
    def __iter__(self):
        return iter(self._data)

d = SimpleDict()
d['name'] = 'Alice'  # Setting name = Alice
d['age'] = 30        # Setting age = 30
print(d['name'])     # Getting name / Alice
print(len(d))        # 2
print('name' in d)   # True
for key in d:
    print(key)       # name / age

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
32
33
34
35
36
37
38
39

实现了__getitem__和__len__后，你的类就可以被视为序列类型，支持切片操作(虽然需要额外处理)：

# 添加切片支持
class SimpleList:
    def __init__(self, items):
        self._items = list(items)
    
    def __getitem__(self, index):
        if isinstance(index, slice):
            # 处理切片
            return SimpleList(self._items[index])
        return self._items[index]
    
    def __len__(self):
        return len(self._items)
    
    def __repr__(self):
        return f"SimpleList({self._items})"

sl = SimpleList([1, 2, 3, 4, 5])
print(sl[1:4])  # SimpleList([2, 3, 4])

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

# 2.1.3 可调用对象：call

通过实现__call__方法，你可以让类的实例像函数一样被调用：

class Adder:
    def __init__(self, n):
        self.n = n
    
    def __call__(self, x):
        return x + self.n

add5 = Adder(5)
print(add5(10))  # 15 (add5像函数一样被调用)
print(add5(20))  # 25

# 检查是否可调用
print(callable(add5))  # True

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

__call__的应用场景包括：带状态的函数(如上例)、装饰器类、函数式编程中的组合器等。

# 2.1.4 比较操作：eq、lt等

class Vector:
    def __init__(self, x, y):
        self.x = x
        self.y = y
    
    def __repr__(self):
        return f"Vector({self.x}, {self.y})"
    
    # 相等
    def __eq__(self, other):
        if not isinstance(other, Vector):
            return NotImplemented  # 让Python尝试反向操作
        return self.x == other.x and self.y == other.y
    
    # 小于
    def __lt__(self, other):
        if not isinstance(other, Vector):
            return NotImplemented
        # 按长度比较
        return (self.x**2 + self.y**2) < (other.x**2 + other.y**2)
    
    # 通过functools.total_ordering自动补全其他比较
from functools import total_ordering

@total_ordering
class Vector:
    def __init__(self, x, y):
        self.x = x
        self.y = y
    
    def __repr__(self):
        return f"Vector({self.x}, {self.y})"
    
    def __eq__(self, other):
        if not isinstance(other, Vector):
            return NotImplemented
        return self.x == other.x and self.y == other.y
    
    def __lt__(self, other):
        if not isinstance(other, Vector):
            return NotImplemented
        return (self.x**2 + self.y**2) < (other.x**2 + other.y**2)
    
    # __le__, __gt__, __ge__, __ne__ 自动生成

v1 = Vector(1, 2)
v2 = Vector(2, 1)
v3 = Vector(1, 2)

print(v1 == v3)  # True
print(v1 < v2)   # False (1^2+2^2=5, 2^2+1^2=5, 相等所以不小于)
print(v1 <= v2)  # True (自动生成)

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
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

关键点： 当操作数类型不兼容时，返回NotImplemented而不是抛出TypeError。这允许Python尝试另一个操作数的反向方法(如__radd__)。

# 2.1.5 算术运算：add、radd等

class Vector:
    def __init__(self, x, y):
        self.x = x
        self.y = y
    
    def __repr__(self):
        return f"Vector({self.x}, {self.y})"
    
    # 加法：self + other
    def __add__(self, other):
        if isinstance(other, Vector):
            return Vector(self.x + other.x, self.y + other.y)
        elif isinstance(other, (int, float)):
            return Vector(self.x + other, self.y + other)
        return NotImplemented
    
    # 反向加法：other + self (当other没有实现__add__时调用)
    def __radd__(self, other):
        return self.__add__(other)
    
    # 原地加法：self += other
    def __iadd__(self, other):
        result = self + other
        self.x, self.y = result.x, result.y
        return self

v1 = Vector(1, 2)
v2 = Vector(3, 4)
print(v1 + v2)      # Vector(4, 6)
print(v1 + 5)       # Vector(6, 7)
print(5 + v1)       # Vector(6, 7) (通过__radd__)

v1 += v2
print(v1)           # Vector(4, 6) (原地修改)

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
32
33
34

理解__add__和__radd__的区别至关重要，它们共同实现了算术操作的交换律。

# 2.2 属性访问控制

Python的属性访问机制非常灵活，你可以通过特殊方法拦截、定制属性的读取、赋值和删除。

# 2.2.1 getattr vs getattribute

这两个方法容易混淆，但它们有本质区别：

__getattr__：仅在访问不存在的属性时调用
__getattribute__：每次访问属性都会调用(包括存在的属性)

class AttrDemo:
    def __init__(self):
        self.name = "AttrDemo"
    
    # 当访问不存在的属性时调用
    def __getattr__(self, name):
        print(f"__getattr__: {name} not found")
        return f"Default value for {name}"
    
    # 每次访问属性都会调用(谨慎使用！)
    def __getattribute__(self, name):
        print(f"__getattribute__: accessing {name}")
        # 必须调用父类的__getattribute__来获取实际属性
        return super().__getattribute__(name)

obj = AttrDemo()
print(obj.name)       # __getattribute__: accessing name / AttrDemo
print(obj.age)        # __getattribute__: accessing age / __getattr__: age not found / Default value for age

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

重要警告： 在__getattribute__中访问属性时，必须使用super().__getattribute__(name)，否则会无限递归。同样，在__getattr__中也不要使用self.__dict__来访问不存在的属性。

# 2.2.2 setattr与delattr

这两个方法分别拦截属性赋值和删除：

class ValidatedAttr:
    def __init__(self, name):
        self.name = name
        self.age = 0  # 这会调用__setattr__
    
    def __setattr__(self, name, value):
        print(f"Setting {name} = {value}")
        if name == "age":
            if not isinstance(value, int):
                raise TypeError("age must be int")
            if value < 0 or value > 150:
                raise ValueError("age must be between 0 and 150")
        # 必须调用父类的方法来实际设置属性
        super().__setattr__(name, value)
    
    def __delattr__(self, name):
        print(f"Deleting {name}")
        if name == "name":
            raise AttributeError("Cannot delete name")
        super().__delattr__(name)

p = ValidatedAttr("Alice")
p.age = 30   # Setting age = 30
# p.age = "30"  # TypeError: age must be int
# p.age = 200   # ValueError: age must be between 0 and 150
# del p.name    # AttributeError: Cannot delete name

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

这个模式可以用来实现数据验证，确保对象始终处于有效状态。

# 2.2.3 实战：延迟加载属性

__getattr__的一个经典应用是实现延迟加载(lazy loading)，即只有在属性第一次被访问时才计算它的值：

class LazyProperty:
    def __init__(self, filename):
        self.filename = filename
        self._data = None
    
    def __getattr__(self, name):
        if name == "data":
            if self._data is None:
                print(f"Loading data from {self.filename}...")
                # 模拟从文件加载大数据
                with open(self.filename, 'r') as f:
                    self._data = f.read()
                print("Data loaded")
            return self._data
        raise AttributeError(f"'{type(self).__name__}' has no attribute '{name}'")

# 使用示例
lp = LazyProperty("large_file.txt")
# 此时data属性还不存在，文件未被加载
print("Object created")
# 第一次访问data，触发加载
data = lp.data  # Loading data from large_file.txt... / Data loaded
# 第二次访问data，直接返回缓存的值
data2 = lp.data  # 没有输出，直接返回

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

这种模式在ORM框架中很常见，用于延迟加载关联的对象。

# 2.3 描述符协议

描述符是Python中一个强大但常被忽视的特性。它是property、类方法、静态方法的底层实现机制。

# 2.3.1 描述符协议基础

描述符是实现了特定协议的类，只要实现了__get__、__set__、__delete__中的任意一个，这个类就是描述符。

# 一个简单的描述符
class VerboseAttribute:
    def __get__(self, instance, owner):
        print(f"Getting from {instance} of {owner}")
        return self._value if hasattr(self, '_value') else None
    
    def __set__(self, instance, value):
        print(f"Setting {instance} to {value}")
        self._value = value
    
    def __delete__(self, instance):
        print(f"Deleting from {instance}")
        del self._value

class MyClass:
    attr = VerboseAttribute()  # 描述符作为类属性

obj = MyClass()
obj.attr = 42   # Setting <__main__.MyClass object...> to 42
print(obj.attr) # Getting from <__main__.MyClass object...> of <class '__main__.MyClass'> / 42

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

描述符的核心在于：属性访问被委托给了描述符对象。当访问obj.attr时，Python发现attr是一个描述符，于是调用描述符的__get__方法。

# 2.3.2 数据描述符 vs 非数据描述符

描述符分为两种类型：

数据描述符：同时定义了__get__和__set__
非数据描述符：只定义了__get__

区别在于属性查找的优先级：

数据描述符的优先级高于实例的__dict__
非数据描述符的优先级低于实例的__dict__

# 非数据描述符
class NonDataDescriptor:
    def __get__(self, instance, owner):
        print("NonDataDescriptor __get__")
        return 42

# 数据描述符
class DataDescriptor:
    def __get__(self, instance, owner):
        print("DataDescriptor __get__")
        return 42
    
    def __set__(self, instance, value):
        print("DataDescriptor __set__")

class Test:
    non_data = NonDataDescriptor()
    data = DataDescriptor()
    
    def __init__(self):
        self.non_data = "instance value"
        self.data = "instance value"

t = Test()
print(t.non_data)  # "instance value" (实例属性覆盖了非数据描述符)
print(t.data)      # DataDescriptor __get__ / 42 (数据描述符优先级更高)

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

这个优先级差异解释了为什么方法(非数据描述符)可以被实例属性覆盖，而property(数据描述符)不能：

class Demo:
    def method(self):
        return "original method"
    
    @property
    def prop(self):
        return "original property"
    
    # 定义 setter
    @prop.setter
    def prop(self, value):
        # 这里可以添加赋值逻辑，比如数据校验
        self._prop = value  # 通常用一个私有属性存储实际值

d = Demo()
d.method = "override method"
d.prop = "override property"  # AttributeError: can't set attribute

print(d.method)  # "override method" (实例属性覆盖了方法)
print(d.prop)    # "original property" (property数据描述符阻止赋值)

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

# 2.3.3 实战：使用描述符实现类型验证

描述符的一个典型应用是创建类型安全的属性：

class TypedAttribute:
    def __init__(self, name, expected_type):
        self.name = name
        self.expected_type = expected_type
    
    def __get__(self, instance, owner):
        if instance is None:
            return self
        return instance.__dict__.get(self.name)
    
    def __set__(self, instance, value):
        if not isinstance(value, self.expected_type):
            raise TypeError(f"{self.name} must be {self.expected_type.__name__}")
        instance.__dict__[self.name] = value
    
    def __delete__(self, instance):
        del instance.__dict__[self.name]

class Person:
    name = TypedAttribute("name", str)
    age = TypedAttribute("age", int)
    
    def __init__(self, name, age):
        self.name = name  # 触发描述符的__set__
        self.age = age

p = Person("Alice", 30)
# p.name = 123  # TypeError: name must be str
# p.age = "30"  # TypeError: age must be int
print(p.name)  # Alice

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

这个模式可以扩展到更复杂的验证逻辑，如范围检查、正则匹配等。

# 2.3.4 property的本质

@property装饰器本质上就是一个数据描述符。我们可以自己实现一个简化版的property来理解其原理：

class MyProperty:
    def __init__(self, fget=None, fset=None, fdel=None):
        self.fget = fget
        self.fset = fset
        self.fdel = fdel
    
    def __get__(self, instance, owner):
        if instance is None:
            return self
        if self.fget is None:
            raise AttributeError("unreadable attribute")
        return self.fget(instance)
    
    def __set__(self, instance, value):
        if self.fset is None:
            raise AttributeError("can't set attribute")
        self.fset(instance, value)
    
    def __delete__(self, instance):
        if self.fdel is None:
            raise AttributeError("can't delete attribute")
        self.fdel(instance)
    
    def getter(self, fget):
        return type(self)(fget, self.fset, self.fdel)
    
    def setter(self, fset):
        return type(self)(self.fget, fset, self.fdel)
    
    def deleter(self, fdel):
        return type(self)(self.fget, self.fset, fdel)

class Circle:
    def __init__(self, radius):
        self._radius = radius
    
    @MyProperty
    def radius(self):
        return self._radius
    
    @radius.setter
    def radius(self, value):
        if value <= 0:
            raise ValueError("radius must be positive")
        self._radius = value
    
    @MyProperty
    def area(self):
        return 3.14 * self._radius ** 2

c = Circle(5)
print(c.radius)  # 5
c.radius = 10
print(c.area)    # 314.0
# c.radius = -1  # ValueError: radius must be positive

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
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

理解了这个原理，你就掌握了Python属性控制的核心。

# 2.4 抽象基类与协议

抽象基类(ABC)定义了一组接口，子类必须实现这些接口。Python的abc模块提供了对抽象基类的支持。

# 2.4.1 抽象基类基础

from abc import ABC, abstractmethod

class Shape(ABC):
    @abstractmethod
    def area(self):
        """计算面积，子类必须实现"""
        pass
    
    @abstractmethod
    def perimeter(self):
        """计算周长，子类必须实现"""
        pass
    
    def description(self):
        """具体方法，子类可以继承"""
        return f"Shape with area {self.area():.2f}"

# 尝试实例化抽象类
# s = Shape()  # TypeError: Can't instantiate abstract class Shape

class Rectangle(Shape):
    def __init__(self, width, height):
        self.width = width
        self.height = height
    
    def area(self):
        return self.width * self.height
    
    def perimeter(self):
        return 2 * (self.width + self.height)

class Circle(Shape):
    def __init__(self, radius):
        self.radius = radius
    
    def area(self):
        return 3.14 * self.radius ** 2
    
    def perimeter(self):
        return 2 * 3.14 * self.radius

# 可以实例化
rect = Rectangle(3, 4)
circle = Circle(5)

print(rect.description())   # Shape with area 12.00
print(circle.description()) # Shape with area 78.50

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
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

抽象基类的作用：

定义接口规范：明确子类必须实现哪些方法
阻止实例化：防止不完整的类被使用
提供默认实现：具体方法可以被子类继承

# 2.4.2 subclasshook与结构化类型检查

Python支持鸭子类型("如果它走起来像鸭子，叫起来像鸭子，那么它就是鸭子")。抽象基类可以通过__subclasshook__实现结构化类型检查，即根据类的方法来判断它是否应该被视为子类，而不需要显式继承。

from abc import ABC, abstractmethod

class Iterable(ABC):
    @abstractmethod
    def __iter__(self):
        pass
    
    @classmethod
    def __subclasshook__(cls, subclass):
        # 检查subclass是否有__iter__方法
        if any("__iter__" in B.__dict__ for B in subclass.__mro__):
            return True
        return NotImplemented

# 一个没有继承Iterable的类
class MyList:
    def __init__(self, items):
        self.items = items
    
    def __iter__(self):
        return iter(self.items)

# 但issubclass会返回True
print(issubclass(MyList, Iterable))  # True

# isinstance也一样
ml = MyList([1, 2, 3])
print(isinstance(ml, Iterable))  # True

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

这个机制使得抽象基类可以识别遵循特定协议(即实现了必要方法)的任何类，而不需要显式继承。

Python标准库中有许多这样的抽象基类，如collections.abc模块中的Iterable、Sequence、Mapping等：

from collections.abc import Iterable, Sized

class MySequence:
    def __init__(self, items):
        self._items = list(items)
    
    def __iter__(self): # 只要有这个，就是 Iterable
        return iter(self._items)
    
    def __len__(self):  # 只要有这个，就是 Sized
        return len(self._items)

seq = MySequence([1, 2, 3])

# 即使没有显式继承，下面也会返回 True
print(f"Is Iterable? {isinstance(seq, Iterable)}") # True
print(f"Is Sized? {isinstance(seq, Sized)}")       # True

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

# 2.4.3 注册机制

除了__subclasshook__，抽象基类还提供了注册机制，允许将某个类标记为虚拟子类：

from abc import ABC

class MyABC(ABC):
    pass

class RegularClass:
    pass

# 注册RegularClass为MyABC的虚拟子类
MyABC.register(RegularClass)

print(issubclass(RegularClass, MyABC))  # True

# 实例化
rc = RegularClass()
print(isinstance(rc, MyABC))  # True

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

注册的类不会继承任何方法或属性，但isinstance和issubclass会返回True。

# 2.4.4 协议：Python的隐式接口

协议(protocol)是Python中一种非正式的接口概念。它不是通过继承实现的，而是通过实现特定的一组特殊方法。例如：

迭代器协议：实现__iter__和__next__
上下文管理器协议：实现__enter__和__exit__
序列协议：实现__getitem__和__len__

# 实现迭代器协议
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 num in CountDown(5):
    print(num, end=' ')  # 5 4 3 2 1

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

这个类没有继承任何抽象基类，但它实现了迭代器协议，因此可以用于for循环。

# 2.5 本章思考与实战

思考题： 描述符、@property、__getattr__都可以控制属性访问，它们各自适合什么场景？优缺点是什么？
实战任务： 实现一个ValidatedString描述符，它支持验证字符串的长度、正则模式等。然后创建一个User类，包含username(长度3-20，只能包含字母数字)和password(长度至少8，包含大小写字母和数字)属性。
进阶挑战： 研究collections.abc模块，实现一个自定义的不可变序列类型，要求支持索引、切片、len()和in操作符，但不支持赋值。

# 第3章掌控对象与类的创建：元编程

元编程是指编写能够操作代码本身的程序。在Python中，这意味着你可以动态地创建类、修改类的行为、甚至改变整个类系统的工作方式。

# 3.1 类型系统的终极奥秘

要理解元编程，首先必须理解Python中"一切皆对象"的真正含义。

# 3.1.1 类也是对象

在Python中，类本身也是对象——它们是type的实例：

class MyClass:
    pass

obj = MyClass()

print(type(obj))      # <class '__main__.MyClass'>
print(type(MyClass))  # <class 'type'>
print(type(type))     # <class 'type'>

1
2
3
4
5
6
7
8

这意味着：

你可以将类赋值给变量
你可以将类作为参数传递
你可以动态地创建类

# 类可以赋值给变量
MyAlias = MyClass
obj2 = MyAlias()
print(type(obj2))  # <class '__main__.MyClass'>

# 类可以作为函数参数
def create_instance(cls):
    return cls()

obj3 = create_instance(MyClass)
print(type(obj3))  # <class '__main__.MyClass'>

1
2
3
4
5
6
7
8
9
10
11

# 3.1.2 类型系统的三层结构

Python的类型系统可以用三层结构来理解：

实例 (obj) ----> 类 (MyClass) ----> 元类 (type)

实例由类创建，实例的类型是它的类
类由元类创建，类的类型是元类
元类由type自身创建，type的类型是它自己

# 实例 (obj) ----> 类 (MyClass) ----> 元类 (type)
# 验证三层关系
class MyClass:
    pass

obj = MyClass()

print(obj.__class__)        # <class '__main__.MyClass'>
print(MyClass.__class__)    # <class 'type'>
print(type.__class__)       # <class 'type'>

# 查看继承链
print(MyClass.__bases__)    # (<class 'object'>,)
print(type.__bases__)       # (<class 'object'>,)
print(object.__class__)     # <class 'type'>

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

这个结构的核心是：元类是创建类的类。

# 3.2 元类基础

# 3.2.1 什么是元类？

元类(metaclass)是创建类的类。你可以把它想象成"类的工厂"——当你定义一个类时，Python会使用元类来创建这个类对象。

默认情况下，所有类都是由type元类创建的：

# 这两种定义类是等价的
class MyClass:
    x = 10
    def hello(self):
        return "Hello"

# 使用type动态创建类
MyClass = type('MyClass', (), {
    'x': 10,
    'hello': lambda self: "Hello"
})

1
2
3
4
5
6
7
8
9
10
11

type的三个参数：

name：类名
bases：父类元组
dict：类的属性和方法字典

# 3.2.2 自定义元类

自定义元类需要继承type，并通常覆盖__new__或__init__方法：

class MyMeta(type):
    def __new__(metacls, name, bases, namespace):
        print(f"Creating class {name}")
        # 可以在创建类之前修改namespace
        namespace['created_by'] = 'MyMeta'
        return super().__new__(metacls, name, bases, namespace)
    
    def __init__(cls, name, bases, namespace):
        print(f"Initializing class {name}")
        super().__init__(name, bases, namespace)

# 使用元类
class MyClass(metaclass=MyMeta):
    x = 10

print(MyClass.created_by)  # MyMeta

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

当Python看到class MyClass(metaclass=MyMeta):时，它会：

收集类定义中的属性和方法到命名空间(namespace)
调用MyMeta.__new__()来创建类对象
调用MyMeta.__init__()来初始化类对象

# 3.2.3 元类的应用场景

场景1：自动注册子类

# 元类实现自动注册
class RegistryMeta(type):
    registry = {}
    
    def __new__(metacls, name, bases, namespace):
        cls = super().__new__(metacls, name, bases, namespace)
        # 跳过基类自身的注册
        if name != "BasePlugin":
            metacls.registry[name] = cls
        return cls

class BasePlugin(metaclass=RegistryMeta):
    """所有插件的基类"""
    def process(self, data):
        raise NotImplementedError

class PluginA(BasePlugin):
    def process(self, data):
        return f"Plugin A processing {data}"

class PluginB(BasePlugin):
    def process(self, data):
        return f"Plugin B processing {data}"

# 自动获取所有注册的插件
print(RegistryMeta.registry)  # {'PluginA': <class '__main__.PluginA'>, 'PluginB': <class '__main__.PluginB'>}

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

这种模式在框架中很常见，如Django的模型注册、pytest的插件系统等。

场景2：类的验证

class ValidateMeta(type):
    def __new__(metacls, name, bases, namespace):
        # 确保类有description属性
        if name != "BaseService" and 'description' not in namespace:
            raise TypeError(f"{name} must have a 'description' attribute")
        
        # 确保所有方法都有文档字符串
        for attr_name, attr_value in namespace.items():
            if callable(attr_value) and not attr_value.__doc__:
                raise TypeError(f"Method {attr_name} in {name} must have docstring")
        
        return super().__new__(metacls, name, bases, namespace)

class BaseService(metaclass=ValidateMeta):
    pass

# 这会失败：缺少description属性
# class UserService(BaseService):
#     def get_user(self):
#         return "user"

# 正确的实现
class UserService(BaseService):
    description = "Manage users"
    
    def get_user(self):
        """Get user by ID"""
        return "user"

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

# 3.2.4 元类 vs 类装饰器

元类和类装饰器都可以修改类的行为，但它们有各自的适用场景：

特性	元类	类装饰器
应用时机	类创建时	类创建后
继承影响	自动应用于子类	不会自动应用于子类
复杂度	较高	较低
使用场景	需要深度定制类创建过程，需要影响继承体系	简单增强，不需要继承关系

# 类装饰器版本(用于简单场景)
def add_description(desc):
    def decorator(cls):
        cls.description = desc
        return cls
    return decorator

@add_description("Manage users")
class UserService:
    def get_user(self):
        return "user"

print(UserService.description)  # Manage users

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

# 3.3 new与init的深入理解

__new__和__init__共同控制对象的创建和初始化过程。理解它们的区别是掌握Python元编程的关键。

# 3.3.1 对象创建过程

class Demo:
    def __new__(cls, *args, **kwargs):
        print(f"1. __new__ called with {cls}")
        print(f"   args: {args}, kwargs: {kwargs}")
        # 创建实例
        instance = super().__new__(cls)
        print(f"2. Instance created: {instance}")
        return instance
    
    def __init__(self, *args, **kwargs):
        print(f"3. __init__ called with {self}")
        print(f"   args: {args}, kwargs: {kwargs}")
        self.args = args

print("Creating instance...")
d = Demo(1, 2, x=3)
print(f"4. Instance: {d}")

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

输出：

Creating instance...
1. __new__ called with <class '__main__.Demo'>
   args: (1, 2), kwargs: {'x': 3}
2. Instance created: <__main__.Demo object at 0x...>
3. __init__ called with <__main__.Demo object at 0x...>
   args: (1, 2), kwargs: {'x': 3}
4. Instance: <__main__.Demo object at 0x...>

1
2
3
4
5
6
7

关键点：

__new__是类方法，负责创建实例
__new__必须返回一个实例(通常是调用super().__new__(cls))
__init__是实例方法，负责初始化已创建的实例
如果__new__返回的不是当前类的实例，__init__不会被调用

# 3.3.2 单例模式实现

利用__new__可以轻松实现单例模式：

class Singleton:
    _instance = None
    
    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            print("Creating the instance")
            cls._instance = super().__new__(cls)
        return cls._instance
    
    def __init__(self, value=None):
        # 注意：每次Singleton()调用都会执行__init__
        # 需要额外处理避免重复初始化
        if not hasattr(self, 'initialized'):
            print(f"Initializing with {value}")
            self.value = value
            #self.initialized = True

s1 = Singleton(10)  # Creating the instance / Initializing with 10
s2 = Singleton(20)  # (只输出Initializing with 20，因为__init__又执行了)
print(s1 is s2)     # True
print(s1.value, s2.value)  # 20 20 (s1的值被覆盖了！)

# 把self.initialized = True放开
class Singleton:
    _instance = None
    
    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            print("Creating the instance")
            cls._instance = super().__new__(cls)
        return cls._instance
    
    def __init__(self, value=None):
        # 注意：每次Singleton()调用都会执行__init__
        # 需要额外处理避免重复初始化
        if not hasattr(self, 'initialized'):
            print(f"Initializing with {value}")
            self.value = value
            self.initialized = True

s1 = Singleton(10)  # Creating the instance / Initializing with 10
s2 = Singleton(20)  # 无输出
print(s1 is s2)     # True
print(s1.value, s2.value)  # 10 10 (s1的值没有被覆盖！)

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
32
33
34
35
36
37
38
39
40
41
42
43
44

更好的单例实现，避免__init__重复执行：

class Singleton:
    _instance = None
    _initialized = False
    
    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance
    
    def __init__(self, value=None):
        if not self._initialized:
            print(f"Initializing with {value}")
            self.value = value
            self._initialized = True

s1 = Singleton(10)  # Creating the instance / Initializing with 10
s2 = Singleton(20)  # (没有初始化输出)
print(s1.value, s2.value)  # 10 10

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

# 3.3.3 不可变实例

通过__new__可以创建不可变对象，类似于tuple或str：

class ImmutablePoint:
    def __new__(cls, x, y):
        instance = super().__new__(cls)
        # 使用 object.__setattr__ 绕过自定义 __setattr__
        object.__setattr__(instance, '_x', x)
        object.__setattr__(instance, '_y', y)
        return instance

    @property
    def x(self):
        return self._x

    @property
    def y(self):
        return self._y

    def __setattr__(self, name, value):
        raise AttributeError(f"ImmutablePoint object cannot modify {name}")

# 测试
p = ImmutablePoint(3, 4)
print(p.x, p.y)  # 输出: 3 4
# p.x = 5        # 抛出 AttributeError

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

# 3.3.4 元类与new的结合

在元类中覆盖__new__，可以控制类的创建过程：

class UpperAttrMeta(type):
    def __new__(metacls, name, bases, namespace):
        # 将属性名转换为大写
        upper_namespace = {
            name.upper() if not name.startswith('__') else name: value
            for name, value in namespace.items()
        }
        return super().__new__(metacls, name, bases, upper_namespace)

class MyClass(metaclass=UpperAttrMeta):
    x = 10
    y = 20

print(hasattr(MyClass, 'x'))   # False
print(hasattr(MyClass, 'y'))   # False
print(hasattr(MyClass, 'X'))   # True
print(MyClass.X)  # 10

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

# 3.4 本章思考与实战

思考题： 元类和类装饰器都可以修改类，它们各自的优缺点是什么？在哪些场景下元类是不可替代的？
实战任务： 实现一个元类SingletonMeta，使得使用该元类的类自动成为单例。注意处理好__init__的重复执行问题。
进阶挑战： 参考Django ORM，实现一个迷你ORM框架，使用元类将类定义映射到数据库表。支持字段类型、表名自动生成、基本的CRUD操作。

# 第二部分：并发与性能——构建高效的应用程序

在掌握了Python语言的核心特性之后，我们将进入并发编程的世界。现代应用程序往往需要同时处理大量任务，如何高效地利用CPU和I/O资源，是区分高级程序员的关键技能。

# 第4章并发编程全景图：选择正确的工具

并发编程并非单一概念，它包含了多线程、多进程、异步编程等多种模型。理解它们的异同，才能在实际项目中做出正确选择。

# 4.1 理论基石：同步 vs 异步 vs 并发 vs 并行

在深入代码之前，我们需要厘清几个关键概念：

同步(Synchronous)：任务按顺序执行，一个任务完成后才能开始下一个。
异步(Asynchronous)：任务可以“在后台”执行，不必等待前一个任务完成即可开始下一个。
并发(Concurrency)：多个任务在同一时间段内交替执行(单核CPU也可以实现)。
并行(Parallelism)：多个任务在同一时刻同时执行(需要多核CPU)。

图解事件循环与非阻塞I/O：

异步编程的核心是事件循环(Event Loop)和非阻塞I/O。下面是一个简化的图示：

传统同步模型：
[任务1: I/O等待] --> [任务2] --> [任务3]
        |                |
    阻塞等待          等待任务1完成

异步事件循环模型：
事件循环: 检查任务1是否完成? 否 -> 处理任务2 -> 检查任务1? 是 -> 处理任务1的结果

1
2
3
4
5
6
7

在异步模型中，当任务1遇到I/O操作(如网络请求、文件读取)时，它不会阻塞，而是将控制权交还给事件循环，事件循环可以继续执行其他任务。当I/O完成时，事件循环会回来继续处理任务1。

# 4.2 多线程编程：threading模块详解

Python的threading模块提供了基于线程的并发支持。线程是操作系统能够进行运算调度的最小单位，被包含在进程之中。

4.2.1 创建和启动线程

import threading
import time

def worker(name, delay):
    """线程工作函数"""
    print(f"线程 {name} 开始工作")
    time.sleep(delay)
    print(f"线程 {name} 完成工作，休眠了 {delay} 秒")

# 创建线程
t1 = threading.Thread(target=worker, args=("A", 2))
t2 = threading.Thread(target=worker, args=("B", 1))

# 启动线程
t1.start()
t2.start()

# 等待线程结束
t1.join()
t2.join()

print("所有线程执行完毕")

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

输出(可能顺序不定)：

线程 A 开始工作
线程 B 开始工作
线程 B 完成工作，休眠了 1 秒
线程 A 完成工作，休眠了 2 秒
所有线程执行完毕

1
2
3
4
5

4.2.2 线程同步：锁

当多个线程同时访问共享数据时，可能出现数据不一致的问题。锁(Lock)用于保护临界区。

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(1000000):
        # 获取锁
        lock.acquire()
        try:
            counter += 1
        finally:
            # 释放锁
            lock.release()

# 创建多个线程
threads = []
for i in range(10):
    # threading.Thread 是 Python 中用于创建线程的类
    # 参数 target=increment 指定了该线程启动时要执行的函数(即线程的入口点)。这里将 increment 函数对象传递给 target，而不加括号(increment 而不是 increment())，因为我们要传递函数本身，而不是调用它
    t = threading.Thread(target=increment)
    threads.append(t)
    t.start()

# 等待所有线程结束
for t in threads:
    t.join()

print(f"计数器最终值: {counter}")  # 应该输出 10000000

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

如果不使用锁，最终结果可能远小于预期值，因为counter += 1不是原子操作。

更简洁的写法：使用with语句自动管理锁。

def increment():
    global counter
    for _ in range(1000000):
        with lock:
            counter += 1

1
2
3
4
5

4.2.3 线程间通信：Queue

queue.Queue是线程安全的队列，用于在线程间传递数据。

import threading
import queue
import time
import random

# 生产者-消费者模型
def producer(q, name):
    while True:
        item = random.randint(1, 100)
        q.put(item)
        print(f"生产者 {name} 生产了 {item}")
        time.sleep(random.random())

def consumer(q, name):
    while True:
        item = q.get()
        print(f"消费者 {name} 消费了 {item}")
        # 调用 q.task_done() 通知队列当前任务已完成
        q.task_done()
        time.sleep(random.random())

q = queue.Queue(maxsize=5)

# 启动生产者
producers = []
for i in range(2):
    t = threading.Thread(target=producer, args=(q, f"P{i}"))
    t.daemon = True  # 设置为守护线程，主线程结束时自动退出
    t.start()
    producers.append(t)

# 启动消费者
consumers = []
for i in range(3):
    t = threading.Thread(target=consumer, args=(q, f"C{i}"))
    t.daemon = True
    t.start()
    consumers.append(t)

# 让主线程运行一段时间
time.sleep(5)
print("主线程结束")

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
32
33
34
35
36
37
38
39
40
41
42

4.2.4 理解GIL的影响

GIL(全局解释器锁)是Python多线程编程中无法回避的话题。GIL确保任何时候只有一个线程在执行Python字节码。这意味着，对于CPU密集型任务，多线程并不能利用多核优势，甚至可能因为锁竞争而变慢。

# CPU密集型任务：计算斐波那契数列
def fib(n):
    if n <= 1:
        return n
    return fib(n-1) + fib(n-2)

# 单线程版本
import time
start = time.time()
for i in range(10):
    fib(40)
print(f"单线程耗时: {time.time() - start:.2f}秒")

# 多线程版本(由于GIL，不会加速)
import threading
threads = []
start = time.time()
for i in range(10):
    t = threading.Thread(target=fib, args=(40,))
    threads.append(t)
    t.start()
for t in threads:
    t.join()
print(f"多线程耗时: {time.time() - start:.2f}秒")

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

你会发现多线程版本甚至比单线程更慢，因为线程创建和切换有开销。

总结：多线程适用于I/O密集型任务(如网络请求、文件读写)，因为I/O操作会释放GIL，允许其他线程运行。对于CPU密集型任务，应该使用多进程或异步编程。

# 4.3 多进程编程：multiprocessing模块详解

multiprocessing模块通过创建子进程来绕过GIL，实现真正的并行计算。

4.3.1 创建进程

import multiprocessing
import os

def worker(name):
    print(f"进程 {name} 启动，PID: {os.getpid()}")
    result = sum(i**2 for i in range(10**6))
    print(f"进程 {name} 完成，结果: {result}")

if __name__ == "__main__":
    processes = []
    for i in range(4):
        p = multiprocessing.Process(target=worker, args=(f"P{i}",))
        processes.append(p)
        p.start()
    
    for p in processes:
        p.join()
    
    print("所有进程执行完毕")

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

注意：在Windows上，必须将进程启动代码放在if __name__ == "__main__":块中，否则会无限递归创建子进程。

4.3.2 进程间通信：Queue、Pipe

多进程间的内存是独立的，不能像线程那样直接共享全局变量。需要使用专门的IPC机制。

使用Queue：

import multiprocessing

def producer(q):
    for i in range(5):
        q.put(i)
        print(f"生产者放入 {i}")

def consumer(q):
    while True:
        item = q.get()
        if item is None:
            break
        print(f"消费者取出 {item}")

if __name__ == "__main__":
    q = multiprocessing.Queue()
    
    p1 = multiprocessing.Process(target=producer, args=(q,))
    p2 = multiprocessing.Process(target=consumer, args=(q,))
    
    p1.start()
    p2.start()
    
    p1.join()
    q.put(None)  # 发送结束信号
    p2.join()

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

使用Pipe(适用于两个进程间的双向通信)：

import multiprocessing

def sender(conn):
    conn.send("Hello")
    conn.send("World")
    conn.close()

def receiver(conn):
    while True:
        try:
            msg = conn.recv()
        except EOFError:
            break
        print(f"收到: {msg}")

if __name__ == "__main__":
    parent_conn, child_conn = multiprocessing.Pipe()
    
    p1 = multiprocessing.Process(target=sender, args=(parent_conn,))
    p2 = multiprocessing.Process(target=receiver, args=(child_conn,))
    
    p1.start()
    p2.start()
    
    p1.join()
    p2.join()

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

4.3.3 进程池(Pool)

对于大量任务，手动管理多个进程较为繁琐。进程池可以自动管理进程的生命周期。

import multiprocessing
import time

def square(x):
    time.sleep(0.5)
    return x * x

if __name__ == "__main__":
    # 使用 with 语句可以确保在代码块结束后自动调用 pool.close() 和 pool.join()，释放资源。
    # 创建一个包含 4 个工作进程的进程池。如果不指定 processes，默认使用 CPU 核心数。
    with multiprocessing.Pool(processes=4) as pool:
        # 方式1：map，阻塞直到所有任务完成。主进程会等待所有任务完成，然后返回结果列表
        # map(func, iterable) 将 iterable 中的每个元素分配给进程池中的工作进程并行执行 func
        results = pool.map(square, range(10))
        print(results)
        
        # 方式2：map_async，非阻塞，返回AsyncResult对象
        # map_async 类似于 map，但它是非阻塞的：立即返回一个 AsyncResult 对象，主进程可以继续执行后续代码
        async_result = pool.map_async(square, range(10))
        # 可以做其他事情
        print("异步提交，继续执行...")
        # 获取结果async_result.get() 会阻塞直到所有任务完成，并返回结果列表。
        results = async_result.get()
        print(results)
        
        # 方式3：apply_async，提交单个任务
        # apply_async(func, args) 用于提交一个单独的任务(一个参数元组)，并立即返回一个 AsyncResult 对象
        # 这里通过列表推导式为每个 i 提交一个异步任务，将所有 AsyncResult 对象存入 results 列表
        results = [pool.apply_async(square, (i,)) for i in range(10)]
        # 通过列表推导式依次调用每个对象的 get() 方法获取结果。get() 会阻塞直到对应任务完成，但由于任务已经并行执行，总的等待时间仍接近 map 的时间。
        print([r.get() for r in results])

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

4.3.4 共享内存与同步

虽然进程间内存独立，但multiprocessing提供了一些共享内存机制，如Value、Array，以及同步原语Lock、Semaphore等。

import multiprocessing

def worker(n, a, lock):
    with lock:
        n.value += 1
        for i in range(len(a)):
            a[i] += 1

if __name__ == "__main__":
    lock = multiprocessing.Lock()
    # 在共享内存中创建一个整数(类型码 'i' 表示有符号整数)，初始值为0
    num = multiprocessing.Value('i', 0)
    # 共享数组(类型为'd'表示双精度浮点)
    arr = multiprocessing.Array('d', [1.0, 2.0, 3.0])
    
    processes = []
    for i in range(10):
        # 循环创建 10 个进程，每个进程的目标函数都是 worker，并传递相同的共享对象和锁
        p = multiprocessing.Process(target=worker, args=(num, arr, lock))
        processes.append(p)
        # p.start() 启动进程，每个进程独立运行 worker 函数
        p.start()
    
    for p in processes:
        # p.join() 阻塞主进程，直到被调用的子进程执行完毕。这确保所有 10 个进程都完成工作后，主程序才继续执行
        p.join()
    
    print(f"num: {num.value}")
    print(f"arr: {list(arr)}")

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

# 4.4 实战选择：决策流程图

面对一个并发任务，如何选择模型？可以参考以下决策树：

任务类型？
├─ CPU密集型 → 多进程(multiprocessing)
│               或 C扩展、Numba等
└─ I/O密集型 → 并发量？
   ├─ 高并发(成千上万连接) → 异步编程(asyncio)
   └─ 中等并发(几十个) → 多线程(threading)

1
2
3
4
5
6

# 第5章异步编程模型与asyncio

asyncio是Python 3.4引入的异步I/O框架，基于事件循环和协程，能够高效处理大量并发连接。

# 5.1 深入async/await

5.1.1 协程基础

协程(coroutine)是可以暂停执行并稍后恢复的函数。在Python中，使用async def定义的函数是协程函数，调用它返回一个协程对象。

# async def 定义一个协程函数。调用该函数不会立即执行函数体，而是返回一个协程对象。
async def hello():
    print("Hello")
    # await 关键字用于挂起当前协程，等待一个可等待对象(如另一个协程、asyncio.sleep、Future 等)完成后继续执行。在等待期间，事件循环可以切换到其他任务。
    await asyncio.sleep(1)  # 模拟I/O操作
    print("World")

# 协程对象需要事件循环来执行
import asyncio

# main 也是一个协程函数。它内部 await hello() 会等待 hello 协程执行完毕。main 本身也是一个可等待对象。
async def main():
    await hello()

#asyncio.run() 是 Python 3.7+ 引入的便捷函数，用来创建事件循环、运行传入的协程，并负责关闭循环和清理资源。它是运行异步程序的入口。
# 在 Jupyter 里写异步代码，记住直接用 await 就完事了。await main()
asyncio.run(main())

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

5.1.2 await的工作原理

await会挂起当前协程，将控制权交还给事件循环，直到等待的对象(awaitable)完成。awaitable对象包括：

协程(coroutine)
asyncio.Task
实现了__await__方法的对象(如asyncio.Future)

import asyncio
async def fetch_data():
    print("开始获取数据...")
    # await 会挂起当前的 fetch_data 协程，告诉事件循环：“我这里要等 2 秒，你先去干别的事，等我好了再叫我
    await asyncio.sleep(2)  # 模拟网络请求
    return {"data": 42}

async def main():
    result = await fetch_data()  # 等待协程完成
    print(result)

asyncio.run(main())

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

# 5.2 任务管理：并发执行多个协程

要并发执行多个协程，需要将它们包装成Task。

import asyncio
async def task(name, delay):
    print(f"任务 {name} 开始")
    await asyncio.sleep(delay)
    print(f"任务 {name} 完成")
    return f"结果 {name}"

async def main():
    # 创建任务
    task1 = asyncio.create_task(task("A", 2))
    task2 = asyncio.create_task(task("B", 1))
    
    # 等待任务完成
    result1 = await task1
    result2 = await task2
    print(result1, result2)

asyncio.run(main())

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

5.2.1 任务编排：gather、wait、as_completed

asyncio.gather同时运行多个协程，并收集结果：

import asyncio

async def task(name, delay):
    print(f"任务 {name} 开始")
    await asyncio.sleep(delay)
    print(f"任务 {name} 完成")
    return f"结果 {name}"

async def main():
    results = await asyncio.gather(
        task("A", 2),
        task("B", 1),
        task("C", 1.5)
    )
    print(results)  # 按输入顺序返回
    
asyncio.run(main())

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

asyncio.wait更灵活，可以设置超时、返回完成和挂起的任务集：

async def main():
    tasks = [
        asyncio.create_task(task("A", 2)),
        asyncio.create_task(task("B", 1)),
        asyncio.create_task(task("C", 1.5))
    ]
    done, pending = await asyncio.wait(tasks, timeout=2.5)
    print(f"已完成: {len(done)}, 未完成: {len(pending)}")
    # 取消未完成的任务
    for t in pending:
        t.cancel()
        
asyncio.run(main())

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

asyncio.as_completed返回一个迭代器，按完成顺序产生结果：

async def main():
    tasks = [task("A", 2), task("B", 1), task("C", 1.5)]
    for coro in asyncio.as_completed(tasks):
        result = await coro
        print(f"获得结果: {result}")
        
asyncio.run(main())

1
2
3
4
5
6
7

# 5.3 异步上下文管理器

异步上下文管理器实现了__aenter__和__aexit__方法，用于异步资源管理，如异步文件操作、数据库连接等。

class AsyncResource:
    async def __aenter__(self):
        print("获取异步资源")
        await asyncio.sleep(1)
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("释放异步资源")
        await asyncio.sleep(0.5)
    
    async def do_something(self):
        print("使用资源")
        await asyncio.sleep(1)

async def main():
    async with AsyncResource() as res:
        await res.do_something()
        
asyncio.run(main())

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

Python 3.11引入了asyncio.timeout上下文管理器，用于超时控制：

async def long_operation():
    await asyncio.sleep(10)

async def main():
    try:
        async with asyncio.timeout(2):
            await long_operation()
    except TimeoutError:
        print("操作超时")
                
asyncio.run(main())

1
2
3
4
5
6
7
8
9
10
11

# 5.4 异步迭代器与异步生成器

5.4.1 异步迭代器

实现了__aiter__和__anext__的对象，可以在async for中使用。

class AsyncCounter:
    def __init__(self, limit):
        self.limit = limit
        self.current = 0
    
    def __aiter__(self):
        return self
    
    async def __anext__(self):
        if self.current >= self.limit:
            raise StopAsyncIteration
        await asyncio.sleep(0.5)  # 模拟异步等待
        self.current += 1
        return self.current

async def main():
    async for num in AsyncCounter(5):
        print(num)
        
asyncio.run(main())

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

5.4.2 异步生成器

使用async for和yield可以更方便地定义异步迭代器。

# 使用 async def 定义的函数，内部包含 yield 语句，这样的函数被称为异步生成器函数。调用它会返回一个异步生成器对象，而不是直接执行函数体
async def async_counter(limit):
    for i in range(limit):
        await asyncio.sleep(0.5)
        yield i + 1

async def main():
    # async for 是异步迭代的专用语法，用于遍历异步可迭代对象(如异步生成器)
    async for num in async_counter(5):
        print(num)
        
asyncio.run(main())

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

# 5.5 高级异步模式

5.5.1 异步队列：生产者-消费者

asyncio.Queue是线程安全的异步队列，用于协程间通信。

import asyncio
import random

# 每个生产者循环生产 5 个随机整数
async def producer(queue, name):
    for i in range(5):
        item = random.randint(1, 100)
        # await queue.put(item)：将物品放入队列。如果队列已满(这里设置了 maxsize=3)，该操作会暂停，直到队列有空位
        await queue.put(item)
        print(f"生产者 {name} 生产了 {item}")
        # 生产后随机睡眠一段时间(0~1 秒)，模拟生产间隔
        await asyncio.sleep(random.random())
    await queue.put(None)  # 结束信号

async def consumer(queue, name):
    while True:
        # 消费者不断从队列中获取物品：await queue.get() 如果队列为空则等待
        item = await queue.get()
        # 如果获取到 None，说明是结束信号，调用 queue.task_done() 标记该信号已被处理，然后退出循环(消费者结束)
        if item is None:
            queue.task_done()
            break
        print(f"消费者 {name} 消费了 {item}")
        await asyncio.sleep(random.random())
        # 调用 queue.task_done() 通知队列该物品处理完成
        queue.task_done()

async def main():
    # 创建队列 q，最大容量为 3，意味着队列最多存放 3 个物品，超出时生产者会阻塞
    q = asyncio.Queue(maxsize=3)
    # 创建 2 个生产者任务和 3 个消费者任务，立即并发运行
    producers = [asyncio.create_task(producer(q, f"P{i}")) for i in range(2)]
    consumers = [asyncio.create_task(consumer(q, f"C{i}")) for i in range(3)]
    
    # 等待所有生产者完成
    await asyncio.gather(*producers)
    # 等待队列被消费完
    await q.join()
    # 取消消费者(已经收到None信号退出)
    for c in consumers:
        c.cancel()

asyncio.run(main())

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
32
33
34
35
36
37
38
39
40
41
42
43

5.5.2 信号量与资源限制

asyncio.Semaphore用于限制同时执行的协程数量，常用于控制并发连接数。

# pip install aiohttp
import aiohttp
import asyncio

sem = asyncio.Semaphore(10)  # 最多10个并发

async def fetch_url(session, url):
    # async with sem:：这是一个异步上下文管理器，进入时会等待获取信号量(如果当前活跃请求数已达 10，则挂起等待)，退出时会自动释放信号量。这样就保证了同时只有最多 10 个协程执行 session.get 部分
    async with sem:
        # async with session.get(url) as response:：发起异步 HTTP GET 请求，返回响应对象。上下文管理器确保响应在退出时正确关闭，释放连接
        async with session.get(url) as response:
            # return await response.text()：读取响应体文本(也是异步操作)，并作为协程的返回值
            return await response.text()

async def main():
    # 生成 URL 列表：这里用同一个 URL 重复 100 次，实际应用中通常是不同的链接
    urls = ["http://www.baidu.com"] * 100
    # async with aiohttp.ClientSession() as session:：创建一个会话对象，会话会自动管理连接池(如 keep-alive)，提高效率。退出上下文时，会话会关闭所有连接
    async with aiohttp.ClientSession() as session:
        # 对每个 URL 调用 fetch_url(session, url)，但这只是创建协程对象，不会立即执行。然后通过 asyncio.gather(*tasks) 将这些协程并发调度。gather 会等待所有协程完成，并返回它们的结果列表
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        print(f"获取了 {len(results)} 个页面")

asyncio.run(main())

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

5.5.3 锁与条件变量

asyncio.Lock和asyncio.Condition用于协程间的同步。

import asyncio
# asyncio.Lock() 创建一个异步锁对象，用于协程间的同步
lock = asyncio.Lock()
# shared_data 是一个全局整数，初始为 0，将被多个协程并发修改
shared_data = 0

async def modify():
    global shared_data
    # async with lock: 是一个异步上下文管理器，它会等待获取锁，如果锁已被其他协程持有，则当前协程挂起直到锁可用。进入代码块后，锁被当前协程持有；退出代码块时自动释放锁
    async with lock:
        # 临界区
        temp = shared_data
        # await asyncio.sleep(0.1)：模拟耗时的 I/O 操作(如网络请求、文件读写)。注意，即使协程在此处暂停，它仍然持有锁，其他协程无法进入临界区，从而保证了数据的一致性
        await asyncio.sleep(0.1)  # 模拟I/O
        shared_data = temp + 1

async def main():
    # 创建 100 个 modify 协程对象
    tasks = [modify() for _ in range(100)]
    # asyncio.gather(*tasks) 并发运行所有协程，等待它们全部完成。由于锁的存在，这 100 个协程会串行化地执行临界区代码，但整体仍是并发的(因为 I/O 等待期间事件循环可以调度其他协程，但锁阻止了同时进入临界区)
    await asyncio.gather(*tasks)
    print(shared_data)  # 100
    
asyncio.run(main())

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

# 第6章性能分析与优化

性能优化遵循“先测量，后优化”的原则。Python提供了多种性能分析工具。

# 6.1 性能分析工具

6.1.1 使用cProfile进行函数级分析

import cProfile
import pstats

def slow_function():
    total = 0
    for i in range(10**7):
        total += i
    return total

def fast_function():
    return sum(range(10**7))

def main():
    slow_function()
    fast_function()

# 运行性能分析
profiler = cProfile.Profile()
profiler.enable()
main()
profiler.disable()

# 输出统计信息
stats = pstats.Stats(profiler)
stats.sort_stats('cumulative')  # 按累计时间排序
stats.print_stats(10)  # 打印前10行

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

6.1.2 使用timeit测量小段代码

import timeit

# 测量列表推导式和生成器表达式的性能
list_time = timeit.timeit('[x**2 for x in range(1000)]', number=10000)
gen_time = timeit.timeit('list(x**2 for x in range(1000))', number=10000)
print(f"列表推导式: {list_time:.4f}s")
print(f"生成器表达式转列表: {gen_time:.4f}s")

1
2
3
4
5
6
7

6.1.3 使用line_profiler逐行分析

需要安装：pip install line_profiler

@profile
def slow_function():
    total = 0
    for i in range(10**6):
        total += i
    return total

if __name__ == "__main__":
    slow_function()

1
2
3
4
5
6
7
8
9

然后运行：kernprof -l -v script.py

(venv) root@hep-terminal:/usr/software# vim script.py
(venv) root@hep-terminal:/usr/software# kernprof -l -v script.py
Wrote profile results to 'script.py.lprof'
Timer unit: 1e-06 s

Total time: 0.918076 s
File: script.py
Function: slow_function at line 1

Line #      Hits         Time  Per Hit   % Time  Line Contents
==============================================================
     1                                           @profile
     2                                           def slow_function():
     3         1          1.0      1.0      0.0      total = 0
     4   1000001     450576.8      0.5     49.1      for i in range(10**6):
     5   1000000     467495.5      0.5     50.9          total += i
     6         1          2.7      2.7      0.0      return total

(venv) root@hep-terminal:/usr/software#

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

或者使用memory_profiler，需要安装pip install memory_profiler。

from memory_profiler import profile

@profile
def slow_function():
    total = 0
    for i in range(10**6):
        total += i
    return total

if __name__ == "__main__":
    slow_function()

1
2
3
4
5
6
7
8
9
10
11

直接 python test.py 即可。

(venv) root@hep-terminal:/usr/software# vim test.py
(venv) root@hep-terminal:/usr/software# python test.py
Filename: /usr/software/test.py
Line #    Mem usage    Increment  Occurrences   Line Contents
=============================================================
     3     49.7 MiB     49.7 MiB           1   @profile
     4                                         def slow_function():
     5     49.7 MiB      0.0 MiB           1       total = 0
     6     49.7 MiB      0.0 MiB     1000001       for i in range(10**6):
     7     49.7 MiB      0.0 MiB     1000000           total += i
     8     49.7 MiB      0.0 MiB           1       return total
(venv) root@hep-terminal:/usr/software#

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

# 6.2 Python性能优化技巧

6.2.1 选择合适的数据结构

使用set而不是list进行成员检测：O(1) vs O(n)
使用deque进行双端操作
使用heapq实现优先队列
使用array存储同类型数值，节省内存

# 成员检测
import timeit

setup = "data_list = list(range(10000)); data_set = set(data_list)"
list_time = timeit.timeit('9999 in data_list', setup=setup, number=10000)
set_time = timeit.timeit('9999 in data_set', setup=setup, number=10000)
print(f"列表成员检测: {list_time:.4f}s")
print(f"集合成员检测: {set_time:.4f}s")

1
2
3
4
5
6
7
8

6.2.2 利用内置函数和库

内置函数(如map、filter、sum)用C实现，通常比手动循环快。

# 手动循环
def sum_squares_loop(n):
    total = 0
    for i in range(n):
        total += i * i
    return total

# 内置函数
def sum_squares_builtin(n):
    return sum(i*i for i in range(n))

# 更快的：使用numpy(如果允许)
import numpy as np
def sum_squares_numpy(n):
    arr = np.arange(n)
    return np.sum(arr*arr)

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

6.2.3 避免重复计算

将循环中不变的计算移到循环外。

# 不好的写法
def process(items, factor):
    result = []
    for item in items:
        result.append(item * (factor * 0.5))  # factor * 0.5 每次都计算
    return result

# 好的写法
def process(items, factor):
    multiplier = factor * 0.5
    result = []
    for item in items:
        result.append(item * multiplier)
    return result

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

6.2.4 使用functools.lru_cache缓存

对于重复计算相同参数的函数，使用缓存可以大幅提升性能。

from functools import lru_cache

@lru_cache(maxsize=128)
def expensive_computation(n):
    # 假设这是一个耗时的计算
    return n ** n

# 第一次调用会计算
result1 = expensive_computation(10)
# 第二次直接返回缓存
result2 = expensive_computation(10)

1
2
3
4
5
6
7
8
9
10
11

6.2.5 尽量使用局部变量

在函数内部，局部变量的访问速度比全局变量快。

# 慢
total = 0
def sum_to_n(n):
    global total
    for i in range(n):
        total += i

# 快
def sum_to_n(n):
    total = 0
    for i in range(n):
        total += i
    return total

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

# 6.3 内存分析与调试

6.3.1 使用memory_profiler

安装：pip install memory_profiler

@profile
def memory_intensive():
    large_list = [i for i in range(10**6)]
    del large_list
    another_list = [i**2 for i in range(10**6)]
    return another_list

if __name__ == "__main__":
    memory_intensive()

1
2
3
4
5
6
7
8
9

运行：python -m memory_profiler script.py

6.3.2 使用tracemalloc追踪内存分配

import tracemalloc

tracemalloc.start()

# 执行一些操作
data = [i for i in range(100000)]
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')

print("[ 内存使用前10名 ]")
for stat in top_stats[:10]:
    print(stat)

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

# 6.4 多语言扩展

当Python性能无法满足需求时，可以考虑编写C/C++扩展。

6.4.1 ctypes调用C库

假设我们有一个C库libmath.so，其中有一个函数int add(int a, int b)。

import ctypes

# 加载共享库
lib = ctypes.CDLL('./libmath.so')

# 指定参数类型和返回类型
lib.add.argtypes = (ctypes.c_int, ctypes.c_int)
lib.add.restype = ctypes.c_int

result = lib.add(3, 5)
print(result)  # 8

1
2
3
4
5
6
7
8
9
10
11

6.4.2 CFFI更Pythonic

安装：pip install cffi

from cffi import FFI

ffi = FFI()
ffi.cdef("int add(int a, int b);")
lib = ffi.dlopen("./libmath.so")

result = lib.add(3, 5)
print(result)

1
2
3
4
5
6
7
8

6.4.3 pybind11 (C++扩展)

pybind11可以轻松地将C++代码绑定到Python。

// example.cpp
#include <pybind11/pybind11.h>

int add(int a, int b) {
    return a + b;
}

PYBIND11_MODULE(example, m) {
    m.def("add", &add, "A function that adds two numbers");
}

1
2
3
4
5
6
7
8
9
10

编译后，在Python中使用：

import example
print(example.add(3, 5))  # 8

1
2

# 第三部分：实战与工程化——构建可靠的软件系统

掌握了语言特性和并发编程后，我们需要将目光投向工程实践：如何编写可维护、可扩展、可靠的代码。

# 第7章类型系统与工程可靠性

Python 3.5+引入了类型注解(type hints)，配合静态类型检查工具，可以在运行前发现潜在错误。

# 7.1 基础类型注解

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

age: int = 30

1
2
3
4

# 7.2 高级类型注解

7.2.1 Union, Optional

from typing import Union, Optional

def parse_int(value: Union[str, int]) -> Optional[int]:
    """将字符串或整数转换为整数，失败返回None"""
    if isinstance(value, int):
        return value
    try:
        return int(value)
    except ValueError:
        return None

1
2
3
4
5
6
7
8
9
10

Python 3.10+可以使用|代替Union：

def parse_int(value: str | int) -> int | None:
    ...

1
2

7.2.2 Literal

限制参数只能为特定的字面量值。

from typing import Literal

def set_mode(mode: Literal["read", "write", "append"]) -> None:
    print(f"模式设置为: {mode}")

set_mode("read")   # OK
set_mode("delete") # 类型检查器会报错

1
2
3
4
5
6
7

7.2.3 TypedDict

定义字典的结构。

from typing import TypedDict

class Person(TypedDict):
    name: str
    age: int
    email: str

def process_person(p: Person) -> None:
    print(p['name'], p['age'])

p = {"name": "Alice", "age": 30, "email": "alice@example.com"}
process_person(p)

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

7.2.4 Protocol

定义鸭子类型协议。

from typing import Protocol

class Drawable(Protocol):
    def draw(self) -> None: ...

class Circle:
    def draw(self) -> None:
        print("画一个圆")

def render(obj: Drawable) -> None:
    obj.draw()

render(Circle())  # OK，因为Circle实现了draw方法

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

# 7.3 dataclasses与类型注解

dataclasses模块可以自动生成__init__、__repr__等方法。

from dataclasses import dataclass, field

@dataclass
class Person:
    name: str
    age: int
    email: str = ""  # 默认值
    tags: list[str] = field(default_factory=list)  # 可变默认值需要用field

p = Person("Alice", 30)
print(p)  # Person(name='Alice', age=30, email='', tags=[])

1
2
3
4
5
6
7
8
9
10
11

# 7.4 静态类型检查：mypy

安装：pip install mypy

运行：mypy script.py

# script.py
def greet(name: str) -> str:
    return 123  # 类型错误

greet(42)  # 类型错误

1
2
3
4
5

mypy会报告：

script.py:2: error: Incompatible return value type (got "int", expected "str")
script.py:4: error: Argument 1 to "greet" has incompatible type "int"; expected "str"

1
2

在大型项目中，集成mypy可以显著提高代码质量。

# 第8章模块化与架构设计

# 8.1 项目结构艺术

一个典型的Python项目结构：

myproject/
├── README.md
├── pyproject.toml          # 项目元数据和依赖(PEP 621)
├── .gitignore
├── src/
│   └── myproject/
│       ├── __init__.py
│       ├── core.py
│       ├── utils.py
│       └── cli.py
├── tests/
│   ├── __init__.py
│   ├── test_core.py
│   └── test_utils.py
├── docs/
├── scripts/
└── requirements/
    ├── base.txt
    ├── dev.txt
    └── prod.txt

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

8.1.1 绝对导入与相对导入

在包内部，推荐使用绝对导入或显式相对导入。

# 绝对导入
from myproject.core import some_function

# 显式相对导入
from .core import some_function

1
2
3
4
5

8.1.2 __init__.py的作用

标识目录为Python包
控制包的导入接口
执行包初始化代码

# myproject/__init__.py
from .core import some_function
__all__ = ['some_function']  # 定义from myproject import *时导入的内容

1
2
3

# 8.2 日志系统

正确配置日志是生产环境必备技能。

import logging
import sys

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('app.log'),
        logging.StreamHandler(sys.stdout)
    ]
)

logger = logging.getLogger(__name__)

def process():
    logger.info("开始处理")
    try:
        # 业务逻辑
        result = 1 / 0
    except Exception as e:
        logger.exception("处理出错")  # 自动记录异常堆栈
    logger.info("处理完成")

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

最佳实践：在模块级别使用logger = logging.getLogger(__name__)，这样日志会包含模块名，方便追踪。

# 8.3 设计模式Python版

设计模式是解决常见设计问题的经验总结。Python的动态特性使得一些模式实现更为简洁。

8.3.1 工厂模式

class Animal:
    def speak(self):
        pass

class Dog(Animal):
    def speak(self):
        return "Woof!"

class Cat(Animal):
    def speak(self):
        return "Meow!"

class AnimalFactory:
    @staticmethod
    def create_animal(animal_type):
        animals = {
            "dog": Dog,
            "cat": Cat
        }
        return animals.get(animal_type.lower(), Animal)()

factory = AnimalFactory()
dog = factory.create_animal("dog")
print(dog.speak())

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

8.3.2 策略模式

from typing import Callable

class PaymentProcessor:
    def __init__(self, payment_strategy: Callable[[float], str]):
        self.payment_strategy = payment_strategy
    
    def process(self, amount):
        return self.payment_strategy(amount)

def credit_card_payment(amount):
    return f"信用卡支付 {amount} 元"

def alipay_payment(amount):
    return f"支付宝支付 {amount} 元"

processor = PaymentProcessor(credit_card_payment)
print(processor.process(100))

# 切换策略
processor.payment_strategy = alipay_payment
print(processor.process(200))

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

8.3.3 观察者模式

class Subject:
    def __init__(self):
        self._observers = []
    
    def attach(self, observer):
        self._observers.append(observer)
    
    def detach(self, observer):
        self._observers.remove(observer)
    
    def notify(self, message):
        for observer in self._observers:
            observer.update(message)

class Observer:
    def __init__(self, name):
        self.name = name
    
    def update(self, message):
        print(f"{self.name} 收到消息: {message}")

subject = Subject()
observer1 = Observer("观察者A")
observer2 = Observer("观察者B")

subject.attach(observer1)
subject.attach(observer2)

subject.notify("事件发生了！")

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

# 第9章扩展与集成：打破语言边界

# 9.1 进程间通信(IPC)

9.1.1 命名管道

在Unix-like系统中，可以使用命名管道(FIFO)实现进程间通信。

# writer.py
import os
import time

pipe_name = '/tmp/mypipe'
if not os.path.exists(pipe_name):
    os.mkfifo(pipe_name)

with open(pipe_name, 'w') as pipe:
    for i in range(10):
        pipe.write(f"Message {i}\n")
        pipe.flush()
        time.sleep(1)

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

# reader.py
import os

pipe_name = '/tmp/mypipe'
with open(pipe_name, 'r') as pipe:
    while True:
        line = pipe.readline()
        if line:
            print(f"Received: {line.strip()}")
        else:
            break

1
2
3
4
5
6
7
8
9
10
11

9.1.2 共享内存(multiprocessing模块)

已在第4章介绍过。

9.1.3 套接字通信

# server.py
import socket

sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.bind('/tmp/mysocket')
sock.listen(1)

conn, addr = sock.accept()
data = conn.recv(1024)
print(f"收到: {data.decode()}")
conn.sendall(b"Hello from server")
conn.close()

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

# client.py
import socket

sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.connect('/tmp/mysocket')
sock.sendall(b"Hello from client")
data = sock.recv(1024)
print(f"收到: {data.decode()}")
sock.close()

1
2
3
4
5
6
7
8
9

# 9.2 与其他语言的集成

9.2.1 Jython：与Java集成

Jython是运行在JVM上的Python实现，可以直接调用Java类。

# 需要Jython环境
from java.util import ArrayList

list = ArrayList()
list.add("Hello")
list.add("World")
print(list)  # [Hello, World]

1
2
3
4
5
6
7

9.2.2 IronPython：与.NET集成

IronPython运行在.NET CLR上，可以调用C#库。

# 需要IronPython环境
import clr
clr.AddReference("System.Windows.Forms")
from System.Windows.Forms import MessageBox

MessageBox.Show("Hello from Python")

1
2
3
4
5
6

# 第10章结语与进阶之路

恭喜你，你已经完成了Python高级进阶的核心内容！但这只是开始。真正的精进在于持续实践和探索。

# 10.1 从语言到生态

根据你的兴趣选择方向，深入相关生态：

Web开发：FastAPI(异步)、Django(全能)、Flask(轻量)
数据分析：Pandas、Polars(高性能DataFrame)、NumPy
机器学习：PyTorch、TensorFlow、scikit-learn、LangChain(LLM应用)
自动化运维：Ansible、Fabric
游戏开发：Pygame、Godot(Python脚本)

# 10.2 推荐资源

书籍：《流畅的Python》(Fluent Python)第二版——深入语言细节的必读之作。
博客：Real Python、PyCoder's Weekly。
开源项目：阅读知名项目的源码，如Requests、Flask、FastAPI、CPython本身。

# 10.3 保持学习

Python语言仍在快速发展，每半年发布一个新版本。关注PEP(Python Enhancement Proposals)，了解新特性，如Python 3.12的type语句改进、3.13的JIT编译实验等。

全文完

本教程从函数式编程、面向对象深入、元编程，到并发编程(多线程、多进程、asyncio)、性能优化，再到工程实践(类型注解、架构设计、集成扩展)，全方位覆盖了Python高级开发所需的知识体系。希望它能成为你进阶路上的良师益友，反复研读，定能收获满满。

Happy Pythoning! 🐍

# Python高级进阶完全教程

# 第一部分：基石与范式——编写"Pythonic"的高级代码

# 第1章 函数式编程深度实践

# 1.1 函数作为一等公民

# 1.1.1 函数即对象

# 1.1.2 高阶函数：map、filter、reduce

# 1.1.3 闭包：函数工厂

# 1.2 从闭包到装饰器

# 1.2.1 装饰器基础

# 1.2.2 保留函数元数据

# 1.2.3 参数化装饰器

# 1.2.4 基于类的装饰器

# 1.2.5 装饰器实战：实现缓存

# 1.3 从生成器到协程

# 1.3.1 生成器基础

# 1.3.2 生成器表达式

# 1.3.3 生成器的进阶用法：send()、throw()、close()

# 1.3.4 yield from：生成器委托

# 1.3.5 生成器实战：处理大文件

# 1.4 本章思考与实战

# 第2章 面向对象系统定制与协议

# 2.1 理解Python的数据模型

# 2.1.1 对象表示相关的特殊方法

# 2.1.2 容器类型模拟

# 2.1.3 可调用对象：call

# 2.1.4 比较操作：eq、__lt__等

# 2.1.5 算术运算：add、__radd__等

# 2.2 属性访问控制

# 2.2.1 getattr vs getattribute

# 2.2.2 setattr__与__delattr

# 2.2.3 实战：延迟加载属性

# 2.3 描述符协议

# 2.3.1 描述符协议基础

# 2.3.2 数据描述符 vs 非数据描述符

# 2.3.3 实战：使用描述符实现类型验证

# 2.3.4 property的本质

# 2.4 抽象基类与协议

# 2.4.1 抽象基类基础

# 2.4.2 __subclasshook__与结构化类型检查

# 2.4.3 注册机制

# 2.4.4 协议：Python的隐式接口

# 2.5 本章思考与实战

# 第3章 掌控对象与类的创建：元编程

# 3.1 类型系统的终极奥秘

# 3.1.1 类也是对象

# 3.1.2 类型系统的三层结构

# 3.2 元类基础

# 3.2.1 什么是元类？

# 3.2.2 自定义元类

# 3.2.3 元类的应用场景

# 3.2.4 元类 vs 类装饰器

# 3.3 __new__与__init__的深入理解

# 3.3.1 对象创建过程

# 3.3.2 单例模式实现

# 3.3.3 不可变实例

# 3.3.4 元类与__new__的结合

# 3.4 本章思考与实战

# 第二部分：并发与性能——构建高效的应用程序

# 第4章 并发编程全景图：选择正确的工具

# 4.1 理论基石：同步 vs 异步 vs 并发 vs 并行

# 4.2 多线程编程：threading模块详解

# 4.3 多进程编程：multiprocessing模块详解

# 4.4 实战选择：决策流程图

# 第5章 异步编程模型与asyncio

# 5.1 深入async/await

# 5.2 任务管理：并发执行多个协程

# 5.3 异步上下文管理器

# 5.4 异步迭代器与异步生成器

# 5.5 高级异步模式

# 第6章 性能分析与优化

# 6.1 性能分析工具

# 6.2 Python性能优化技巧

# 6.3 内存分析与调试

# 6.4 多语言扩展

# 第三部分：实战与工程化——构建可靠的软件系统

# 第7章 类型系统与工程可靠性

# 7.1 基础类型注解

# 7.2 高级类型注解

# 7.3 dataclasses与类型注解

# 7.4 静态类型检查：mypy

# 第1章函数式编程深度实践

# 第2章面向对象系统定制与协议

# 2.1.4 比较操作：eq、lt等

# 2.1.5 算术运算：add、radd等

# 2.2.2 setattr与delattr

# 2.4.2 subclasshook与结构化类型检查

# 第3章掌控对象与类的创建：元编程

# 3.3 new与init的深入理解

# 3.3.4 元类与new的结合

# 第4章并发编程全景图：选择正确的工具

# 第5章异步编程模型与asyncio

# 第6章性能分析与优化

# 第7章类型系统与工程可靠性

# 第8章模块化与架构设计

# 第9章扩展与集成：打破语言边界

# 第10章结语与进阶之路