Python

基础知识

解释型 vs 编译型语言

深拷贝 vs 浅拷贝

可变类型 vs 不可变类型

• 可变与不可变

  • 在 Python 里，所谓 “可变” 和 “不可变”，并不是变量能不能被重新赋值，而是对象本身的内容能否在原地被修改

  • 变量只是一个名字，真正重要的是名字所指向的对象，而不可变类型一旦创建，其内部状态就固定下来，任何 “修改” 操作本质上都会创建一个新对象并返回新的引用；可变类型则允许在对象内部直接改变其内容，对象的身份保持不变

  • 典型的不可变类型包括 int、float、str、tuple、frozenset，而 list、dict、set 则是可变类型

    a = 1
    a += 1 # 创建了一个新的整数对象 2, a 重新绑定到新对象

  • 这一差异直接决定了它们能否作为 dict 的 key 或 set 的元素，因为 Python 要求 key 必须是 hashable 的，而 hashable 的核心前提之一就是对象在生命周期内其哈希值不变 —— 即因为 dict 依赖对象的 hash 值进行定位，而可变对象无法保证 hash 的稳定性

  • 不可变对象天然满足这一点，而可变对象一旦内容变化，其哈希值就会变化，从而破坏哈希表的正确性，因此被禁止作为 key

  • 从实现层面看，这个设计并不是语言层面的 “规矩”，而是为了保证哈希表这种核心数据结构在性能与正确性上的可行性

  • 同时，不可变对象是天然线程安全的，而可变对象容易出现共享修改 bug

  • 以上也是 Python 设计中把 tuple /frozenset 保留出来的原因

• 内建集合类型

  • list
    • list 是 Python 中最常用的序列类型，其底层是一个动态数组，支持 O (1) 的随机访问，同时通过预分配与扩容策略来摊销 append 的成本
    • 它的优势在于通用、灵活、表达力强，但代价是插入和删除中间元素需要移动大量元素，时间复杂度为 O (n)，同时由于是可变对象，使用时需要格外注意共享引用带来的副作用
  • tuple
    • tuple 在语义上是 “不可变的 list”，但在实现与使用层面，它并不只是一个被限制修改的 list
    • 由于不可变，tuple 可以被安全地作为 dict 的 key 或 set 的元素，同时在内存布局上也更紧凑（没有 resize 逻辑），创建和遍历的开销通常略小于 list
    • 在工程中，tuple 常被用来表达 “语义上不应被修改的一组值”，这本身就是一种非常重要的代码自说明能力
    • tuple 可被 hash（前提是内部元素也可 hash）
  • dict
    • dict 是 Python 生态的核心，其底层是一个哈希表，所有 key-value 操作在平均情况下都能达到 O (1)
    • dict 不只是一个映射结构，它还是 Python 对象模型、作用域、模块命名空间等机制的基础
    • dict 是无序的（3.7+ 保证插入顺序，但不是排序）
      • 旧版本的 dict 是典型哈希表，key-value 分散存放，顺序与插入无关
      • Python 3.6 起，dict 拆成两部分：稠密的 entries 数组按插入顺序存 key-value，稀疏的 index 表存 hash → entries 映射
      • 查找仍是 hash → index → entry，复杂度不变，但遍历性能更好，顺序数组的 cache locality 远胜于在哈希表里查找
    • 底层实现：哈希表、开放寻址（open addressing）、探测序列
      • 当多个 key 映射到同一个槽位时，Python 会按照一定的探测序列寻找下一个可用位置，这个探测策略经过精心设计，既能减少聚集效应，又能保持良好的缓存局部性
      • 哈希冲突本身并不可怕，可怕的是大量冲突导致探测链变长，从而使查找退化
      • 在正常情况下，dict 的查找、插入和删除都是均摊 O (1)，但在极端对抗性输入下，理论上可以退化到 O (n)
      • 现代 Python 在哈希函数中引入了随机化机制，目的正是防止恶意构造的 key 导致拒绝服务攻击
      • 一般而言，dict 会留大量空槽，且装载因子不能太高，否则会进行扩容并触发 rehash，开销很大
    • rehash
      • 在 Python 中，dict 扩容会触发 rehash，但这里的 rehash 并不等于重新计算 key 的 hash 值
      • 大多数不可变对象（如字符串）的 hash 会缓存，真正耗费的是重新分配槽位并将元素放入新位置的过程
      • dict 性能依赖装载因子，一旦表太满，冲突增加，O (1) 平均复杂度失效，因此需要扩容
      • 扩容时会分配更大的 index 表，然后根据已有 hash 值重新映射到新槽位，由于槽位数改变，hash % table_size 结果不同，每个元素必须重新 “落位”，这就是 rehash
      • 在 compact dict 中，rehash 的成本主要在 index 表重建，entries 数组可以整体复用，这是新设计的优势
      • 但扩容仍是 O (n) 操作，通过倍增策略均摊到多次插入，所以平均感觉不到
      • 总的来说，rehash 的目的不是重新算 hash，而是 “在新容量下重建 hash → slot 映射”
  • set
    • set 可以看作是 “只有 key、没有 value 的 dict”，其底层实现与 dict 基本一致，同样是哈希表
    • 它的核心语义是 “无序、去重、快速 membership 测试”，因此非常适合用于判重、集合运算以及加速查找
    • frozenset 则是 set 的不可变版本，主要意义在于可作为 dict 的 key 或 set 的元素，用于表达 “一个集合本身也是一个不可变的整体”
  • 在高性能场景下替代 dict/set
    • 如果 key 是整数且范围已知连续，用 list 或 array 就足够，O (1) 访问、内存紧凑、分支少、cache 友好，常见于模型参数、ID 映射、状态表等场景
    • 如果 key 是字符串，但集合是静态只读、频繁查找且几乎不修改，可以用排序数组 + 二分查找，或者 trie /radix tree，这在 tokenizer、词表、配置解析等延迟敏感路径中常用
    • 追求极致性能时，Python 层的 dict 可能成为瓶颈，可转向 C/C++ 扩展、Cython、NumPy 结构化数组，甚至直接在 C++ 用 robin-hood hashing、flat hash map 等，牺牲通用性换更好 cache locality 和更小常数因子
    • 还有一种优化并不是换数据结构，而是换问题表述方式：减少查找次数、提前编码 key、将字符串映射成小整数、合并多次 dict 访问，这类代码级优化在热路径上往往比任何高级数据结构更有效

• collections

  • collections 模块并不是语法层面的扩展，而是对常见数据结构使用模式的高度抽象

  • defaultdict 解决的是 “dict 在访问不存在 key 时的初始化逻辑” 这一重复性极高的问题，它通过一个默认工厂函数，将 “判断 key 是否存在” 的分支逻辑下沉到容器内部

    from collections import defaultdict
    
    d = defaultdict(list)
    d["a"].append(1)

  • Counter 本质上是一个 dict 的子类，而 value 用来做计数统计，它让 “频次统计” 从一堆样板代码，变成一个一眼就能读懂的抽象，同时还支持加减、交并等操作，非常适合文本处理、日志分析和统计类任务

    from collections import Counter
    
    cnt = Counter("aabbbc")

  • deque 是一个双端队列，其底层是由多个固定大小的块组成的链式结构，专门优化了从两端进行 append 和 pop 的场景；在 list 中，左端插入和删除是 O (n) 的，而 deque 能做到真正的 O (1)，这使得 deque 成为实现滑动窗口、BFS、生产者消费者队列等模式的首选；deque 不适合随机访问

    from collections import deque
    
    q = deque()
    q.append(1)
    q.appendleft(0)

  • namedtuple 提供了一种介于 tuple 和 class 之间的轻量结构，它本质上仍是 tuple，因此保持了不可变和高效的特性，但通过字段名提升了可读性；它非常适合用在 “结构简单、语义明确、生命周期短” 的数据载体场景中

    from collections import namedtuple
    
    # 现代工程中可以被 dataclass(frozen=True) 取代
    Point = namedtuple("Point", ["x", "y"])
    p = Point(1, 2)

*args 和**kwargs

• *args：接收多余的位置参数，打包成元组 tuple

  def func(a, b, *args):
      print(a)
      print(b)
      print(args)  # 元组
  
  func(1, 2, 3, 4, 5)
  # args = (3,4,5)

• **kwargs：接收多余的关键字参数，打包成字典 dict

  def func(**kwargs):
      print(kwargs)  # 字典
  
  func(x=1, y=2, z=3)
  # kwargs = {'x':1, 'y':2, 'z':3}

• 普通参数 → *args →**kwargs

  def func(a, b, *args,**kwargs):
      pass

• 调用函数时，是进行解包

  lst = [1,2,3]
  func(*lst)    # 等价于 func(1,2,3)
  
  d = {'x':1,'y':2}
  func(**d)     # 等价于 func(x=1,y=2)

匿名函数与闭包

• 高阶函数

  • 高阶函数指至少满足下列一个条件的函数：
    • 可以接收函数作为参数
    • 可以返回一个函数作为结果
  • 也就是说，高阶函数能对函数进行操作，使得函数本身可以像数据一样传递和处理

• 函数作为参数

  • 示例

    # 函数名不加括号，表示传递函数本身，而不是调用函数
    def apply_twice(func, x):
        return func(func(x))
    
    def square(n):
        return n * n
    
    result = apply_twice(square, 2)  # 相当于 square(square(2)) => 16
    print(result)

  • 高阶函数可以实现回调函数模式，即某个函数执行完毕后调用传入的函数

    # 定义高阶函数，接收回调函数 callback
    def process_data(data, callback):
        print("开始处理数据...")
        result = [x**2 for x in data]  # 假设处理是平方
        print("数据处理完成")
        # 调用回调函数
        callback(result)
    
    # 定义回调函数
    def print_result(res):
        print("回调函数收到结果:", res)
    
    # 调用高阶函数
    data = [1, 2, 3, 4]
    process_data(data, print_result)
    
    # 开始处理数据...
    # 数据处理完成
    # 回调函数收到结果: [1, 4, 9, 16]

  • 需要区分函数对象与函数调用

    def greet():
        print("Hello!")
    
    def call_twice(func):
        print("准备调用函数...")
        func()  # 调用函数
        func()  # 再调用一次
    
    # 传递函数对象
    call_twice(greet)  # 注意：没有括号
    # greet 是函数对象（没有括号）
    # greet() 是函数调用，会立即执行，并返回 None
    call_twice(greet()) # 会执行错误

• 函数作为返回值

  • 示例：make_multiplier 返回一个新函数，而返回的函数可以记住外部 n 的值，这就是闭包的核心

    def make_multiplier(n):
        def multiplier(x):
            return x * n
        return multiplier
    
    double = make_multiplier(2)
    triple = make_multiplier(3)
    
    print(double(5))  # 10
    print(triple(5))  # 15

• 匿名函数（Lambda）

  • 语法：lambda 参数列表: 表达式

  • 在匿名函数中，只能写单行表达式，无 return

  • 使用匿名函数，可以快速创建轻量函数，通常配合高阶函数使用

    add = lambda x, y: x + y
    print(add(3, 5))  # 8

  • lambda 与 def 在功能上可以互换，但 lambda 更偏向函数式编程风格，常用于临时函数或参数传递

• Python 内置了几个函数式编程的核心工具

  • map：将函数作用于可迭代对象的每个元素，返回 map 对象（可转 list）

    nums = [1, 2, 3, 4]
    squared = list(map(lambda x: x**2, nums))
    print(squared)  # [1, 4, 9, 16]

  • filter：根据函数返回值 True/False 过滤可迭代对象

    nums = [1, 2, 3, 4, 5, 6]
    even = list(filter(lambda x: x % 2 == 0, nums))
    print(even)  # [2, 4, 6]

  • reduce：将序列元素依次累积计算 ——“连续归约”，适合累计、归约、阶乘操作

    from functools import reduce
    nums = [1, 2, 3, 4, 5]
    
    # 累积求和
    sum_result = reduce(lambda x, y: x + y, nums)
    print(sum_result)  # 15
    
    # 阶乘
    factorial = reduce(lambda x, y: x * y, nums)
    print(factorial)  # 120 = 1*2*3*4*5

  • sorted /min/max：可以传入 key 函数，实际上也是高阶函数用法

    words = ['apple', 'banana', 'cherry']
    sorted_words = sorted(words, key=lambda x: len(x))
    print(sorted_words)  # ['apple', 'cherry', 'banana']

  • 函数式编程组合示例

    from functools import reduce
    
    nums = [1, 2, 3, 4, 5, 6]
    
    # 1. filter: 只保留偶数
    even_nums = filter(lambda x: x % 2 == 0, nums)  # [2, 4, 6]
    
    # 2. map: 对每个元素平方
    squared_nums = map(lambda x: x**2, even_nums)   # [4, 16, 36]
    
    # 3. reduce: 求和
    total = reduce(lambda x, y: x + y, squared_nums)  # 4+16+36 = 56
    
    print(total)  # 56
    
    # 链式写法
    total = reduce(
        lambda x, y: x + y,
        map(lambda x: x**2,
            filter(lambda x: x % 2 == 0, nums))
    )
    print(total)  # 56

• 闭包（Closure）

  • 闭包是一个函数对象，它记住了创建它的外部作用域变量，即使外部作用域已经结束

  • 示例

    def outer(a):
        b = 10
        def inner(x):
            return x + a + b
        return inner
    
    fn = outer(5)
    print(fn(3))  # 18

  • 在上面的例子中，inner 是闭包，a 和 b 被 “封存” 在 inner 的作用域链中

  • Python 闭包 = 嵌套函数 + 引用了外部函数的自由变量 + 外部函数返回嵌套函数

    • 嵌套函数：函数里定义函数

    • 自由变量：在内部函数使用，但既不是参数也不是局部变量的变量

    • 闭包函数：持有自由变量的内部函数

    • 外部函数执行完退出后，内部函数依然能访问自由变量

    • Python 用 cell 对象（单元格） 来保存自由变量，不让它被回收

      • 当内部函数引用外部变量时，Python 会把这个变量包装成一个 cell 对象

      • 外部函数的栈帧销毁时，cell 对象不会被销毁

      • 内部函数的 __closure__ 属性会持有这个 cell

      • 调用内部函数时，通过 cell 拿到原来的值

  • 闭包与变量保存举例

    def make_counter():
        count = 0  # 外部变量
        def counter():
            nonlocal count  # 告诉 Python 这是外层变量
            count += 1
            return count
        return counter
    
    c = make_counter()
    print(c())  # 1
    print(c())  # 2
    print(c())  # 3

  • 缓存（memoization）示例

    def memoize(func):
        cache = {}  # 外部作用域变量
        def wrapper(x):
            if x not in cache:
                cache[x] = func(x)
            return cache[x]
        return wrapper
    
    @memoize
    def fib(n):
        if n < 2:
            return n
        return fib(n-1) + fib(n-2)
    
    print(fib(10))  # 55，计算过程中使用缓存避免重复计算

  • 查看闭包保存的变量

    def outer(a):
        b = 10
        def inner(x):
            return x + a + b
        return inner
    
    fn = outer(5)
    print(fn(3))  # 18
    
    # 查看闭包保存的变量
    print(fn.__closure__)        # (<cell at 0x...: int object at 0x...>, <cell at 0x...: int object at 0x...>)
    print([c.cell_contents for c in fn.__closure__])  # [5, 10]

  • 在这个例子中，__closure__ 属性保存了闭包环境变量，而 cell_contents 可以查看保存的值

  • lambda 也可以形成闭包

    # lambda 内部访问外部变量 n
    # 与 def 定义函数形成闭包的原理相同
    def make_adder(n):
        return lambda x: x + n
    
    add5 = make_adder(5)
    print(add5(10))  # 15

装饰器

• Python 的 decorator（装饰器）是一种高级语法结构，它可以在不修改函数本身代码的情况下，给函数或类增加额外功能

• 装饰器本质上是一个高阶函数（接受函数作为参数，并返回一个新函数），它在 Python 开发中很常用，比如做日志记录、权限校验、缓存、性能统计等

• 一个最简单的装饰器示例：

  def my_decorator(func):
      def wrapper(*args,**kwargs):
          print("调用前逻辑")
          result = func(*args,**kwargs)
          print("调用后逻辑")
          return result
      return wrapper
  
  @my_decorator
  def say_hello(name):
      print(f"Hello, {name}!")
  
  say_hello("Alice")
  # 调用前逻辑
  # Hello, Alice!
  # 调用后逻辑

• 在这个例子中

  • my_decorator 是装饰器，它接受一个函数 func 作为参数

  • 内部定义了 wrapper 函数，在执行原函数前后加了额外逻辑

  • wrapper 被返回并替代了原函数 say_hello

  • @my_decorator 等价于：

    say_hello = my_decorator(say_hello)

  • 所以调用 say_hello() 时，实际上执行的是 wrapper()

• 如果装饰器本身也需要参数，可以再套一层函数：

  def repeat(times):
      def decorator(func):
          def wrapper(*args,**kwargs):
              for _ in range(times):
                  func(*args,**kwargs)
          return wrapper
      return decorator
  
  @repeat(3)
  def greet(name):
      print(f"Hi, {name}!")
  
  greet("Bob")
  # Hi, Bob!
  # Hi, Bob!
  # Hi, Bob!

• 在这个例子中

  • repeat(times) 返回 decorator
  • decorator 才是真正的装饰器，它接受 func
  • wrapper 在 func 外层加了重复执行的逻辑

• 保留原函数信息

  • 使用装饰器时，wrapper 会替代原函数，这会让函数名、文档字符串丢失

  • 可以用 functools.wraps 保留原函数信息：

    from functools import wraps
    
    def my_decorator(func):
        @wraps(func)
        def wrapper(*args,**kwargs):
            print("before")
            return func(*args,**kwargs)
        return wrapper

• 常见装饰器应用场景

  • 日志记录：记录函数调用信息
  • 性能统计：计算函数执行时间
  • 缓存：例如 functools.lru_cache
  • 权限校验：如 Web 框架中的 @login_required
  • 资源管理：自动打开 / 关闭文件、网络连接等

• 装饰器不仅能装饰函数，也能装饰类：

  def add_method(cls):
      cls.say_hi = lambda self: print("Hi!")
      return cls
  
  @add_method
  class Person:
      pass
  
  p = Person()
  p.say_hi()  # 输出 Hi!

• 总结来说，装饰器的核心思想是包装函数或类，在不修改原代码的前提下添加功能，是一种强大且灵活的元编程手段

• 在实际工程中，装饰器大量用于日志、权限、缓存、性能统计等横切关注点（cross-cutting concerns）

迭代器与生成器

• 生成器和迭代器（Iterator）是 Python 懒加载和高效内存管理的核心机制

• 迭代器是一个可以记住遍历位置的对象，可以被 next() 函数逐个访问元素，直到遍历结束

• 迭代器遵循 Python 的迭代协议

  • 必须实现 __iter__() 方法，返回自身
  • 必须实现 __next__() 方法，返回下一个元素

• 迭代器的特点：

  • 惰性计算：每次只生成一个元素
  • 状态可记忆：每次调用 next() 都从上次停止的地方继续

• 内置迭代器的例子

  # list 是可迭代对象，但不是迭代器
  lst = [1, 2, 3]
  
  it = iter(lst)  # 用 iter() 得到迭代器对象
  print(next(it))  # 1
  print(next(it))  # 2
  print(next(it))  # 3
  
  # 再 next 会报 StopIteration
  # print(next(it))

• 自定义迭代器

  class MyRange:
      def __init__(self, start, end):
          self.current = start
          self.end = end
  
      def __iter__(self):
          return self  # 迭代器返回自身
  
      def __next__(self):
          if self.current >= self.end:
              raise StopIteration
          value = self.current
          self.current += 1
          return value
  
  for i in MyRange(1, 5):
      print(i)

• 迭代器不会一次性生成所有数据，而是按需生成，适合处理大数据

  # 用 range 生成 1 亿个数
  # 列表方式（内存占用大）
  # nums = list(range(100_000_000))  # 会占用大量内存
  
  # 迭代器方式（惰性计算）
  nums_iter = iter(range(100_000_000))  # 仅生成迭代器对象
  print(next(nums_iter))  # 0
  print(next(nums_iter))  # 1

• 可迭代对象（Iterable）

  • 可迭代对象实现了 __iter__() 方法，能返回一个迭代器

  • 特点：可以用于 for 循环或 iter() 函数，但本身不一定能 next()

  • 示例

    lst = [1, 2, 3]
    
    print(hasattr(lst, "__iter__"))  # True，可迭代对象
    print(hasattr(lst, "__next__"))  # False，不是迭代器
    
    # 可以用 iter() 将可迭代对象转为迭代器
    it = iter(lst)
    print(hasattr(it, "__next__"))  # True

  • 列表、元组、字典、集合都是可迭代对象，但不是迭代器

  • 核心区别：可迭代对象可以 “获取迭代器”，迭代器可以 “被 next () 访问元素”

• 生成器是创建迭代器的一种简单方法，本质上是一个返回迭代器的函数

• 生成器的特点

  • 使用 yield 关键字生成元素
  • 每次 yield 暂停函数状态，下次继续执行
  • 惰性计算，一次只生成一个元素
  • 语法简单，比自定义迭代器更方便

• 基本生成器示例

  def my_generator(n):
      for i in range(n):
          yield i  # 暂停并返回 i
  
  gen = my_generator(5)
  print(gen)  # <generator object my_generator at 0x...>
  
  print(next(gen))  # 0
  print(next(gen))  # 1
  print(next(gen))  # 2
  
  for val in gen:
      print(val)  # 3 4

• 在上述例子中，yield 暂停了函数，保留局部变量状态；每次调用 next()，从上次 yield 停下的地方继续

• 可以看出，生成器天然就是迭代器（实现了 __iter__ 和 __next__）

• 生成器表达式

  • 类似列表推导式，但返回生成器对象，而不是列表，占用内存更少：

  • 示例

    gen_exp = (x**2 for x in range(5))
    print(next(gen_exp))  # 0
    print(next(gen_exp))  # 1

  • 与列表推导式 [x**2 for x in range(5)] 相比，生成器表达式不会一次性生成整个列表

• 生成器可以创建无限序列，迭代器做不到

  • 示例

    def infinite_sequence():
        num = 0
        while True:
            yield num
            num += 1
    
    gen = infinite_sequence()
    for _ in range(5):
        print(next(gen))  # 0 1 2 3 4

  • 流式处理

    # 假设有一个日志流，需要每次处理一条：
    def log_stream():
        n = 0
        while True:
            yield f"log-{n}"
            n += 1
    
    logs = log_stream()
    
    for _ in range(3):  # 模拟处理前三条日志
        print(next(logs))

  • 上述例子每次只生成一条日志，节省内存；可与 for 或其他处理管道组合，实现流式处理

  • 如何生成无限序列又不占用大量内存，可以使用生成器函数 + yield，每次返回一个值，同时不保存历史值，不使用列表等数据结构

    def fib_generator():
        a, b = 0, 1
        while True:
            yield a
            a, b = b, a + b
    
    fib = fib_generator()
    for _ in range(10):
        print(next(fib))

• 生成器与迭代器的区别

  特性	迭代器	生成器
  创建方式	class + __iter__/__next__	函数 + yield 或生成器表达式
  内存占用	可控，但需要手动实现	惰性生成，占用小
  无限序列	可实现，但需手动管理状态	很容易实现，自动保留状态
  状态保存	手动保存实例变量	自动保存局部变量状态
  语法简洁性	相对复杂	非常简洁

多线程、多进程、协程

• 并发和并行不是同一个概念

  • 并发的核心不是 “同时”，而是在同一时间段内，处理多个任务，通过切换来推进进度
  • 单核 CPU 也能并发，并可以通过任务切换隐藏等待时间，重点是提高资源利用率
  • 并行的核心是在同一时刻，真正同时执行多个任务
  • 并行必须依赖多核 CPU，重点是缩短总执行时间

• 三者对比

  对比项	多进程（Process）	多线程（Thread）	协程（Coroutine）
  调度主体	操作系统内核调度	操作系统内核调度	用户态自己调度，无需操作系统
  资源占用	高，独立内存空间，开销大	中，共享进程内存，开销较小	极低，栈极小，超轻量
  切换成本	高（切换虚拟地址空间）	中（保存线程上下文）	极低（仅保存寄存器）
  GIL 影响	不受 GIL 限制，真正多核并行	受 GIL 限制，计算密集无法并行	单线程内跑，天然无 GIL 竞争
  数据共享	不共享内存，通信麻烦（Queue/Pipe）	共享内存，需加锁防并发安全问题	天然单线程，无需锁，同步简单
  稳定性	一个进程崩不影响其他	一个线程崩，整个进程直接挂掉	一个协程崩，可捕获异常不影响整体
  适合场景	计算密集型、数据并行	IO 密集型（但受 GIL 限制）	高并发 IO 密集型（爬虫、接口）
  Python 模块	multiprocessing	threading	asyncio、gevent

多线程

• 使用 threading.Thread 来创建线程，线程之间共享内存，开销小，受 GIL 限制，适合 IO 密集型

• 线程模型示例

  import threading
  
  def work():
      print("working")
  
  t = threading.Thread(target=work)
  t.start()
  t.join()

• Python 的 threading.Thread 对应的是操作系统级线程，而不是用户态的 “假线程”，线程确实是 OS 管理的，调度、栈空间、上下文切换都是真实存在的，但问题出在 CPython 解释器本身引入了 GIL（Global Interpreter Lock）

• GIL 的本质不是 “线程锁”，而是 “解释器执行权的互斥锁”

• 在 CPython 中，任意时刻只允许一个线程执行 Python 字节码 —— 即使创建了 8 个线程，在 CPU-bound 的代码里，它们也会轮流抢 GIL，本质上还是单核在跑

• GIL 不是语言层面的设计，而是 CPython 实现层面的选择，它引入 GIL 的设计初衷是为了简化解释器内部对象模型和内存管理，比如引用计数的线程安全问题，但代价就是 CPU 并行能力被限制

• CPU-bound 场景示例：线程 ≈ 串行

  import threading
  
  def cpu_task():
      x = 0
      for _ in range(10**7):
          x += 1
  
  threads = [threading.Thread(target=cpu_task) for _ in range(4)]
  for t in threads:
      t.start()
  for t in threads:
      t.join()

• 在这个例子中，看起来是 4 个线程，但实际上同一时刻只有一个线程在跑 Python 代码，多线程无法利用多核 CPU

• 但这并不意味着 threading 在 Python 中 “没用”，当线程执行的是 IO 操作，比如网络请求、磁盘读写、sleep 等，线程会在进入阻塞系统调用时主动释放 GIL，这时其他线程可以拿到 GIL 执行 Python 代码

• 于是，在 IO-bound 场景下，多线程可以显著提高吞吐量，因为大量时间本来就花在等待 IO 上，而不是执行字节码

• IO-bound 场景示例：线程有效

  import threading
  import time
  
  def io_task():
      time.sleep(1)
  
  threads = [threading.Thread(target=io_task) for _ in range(5)]
  start = time.time()
  for t in threads:
      t.start()
  for t in threads:
      t.join()
  print(time.time() - start)

• 在这个例子中，sleep 会释放 GIL，线程在 IO 等待期间让出执行权，多线程可以隐藏 IO 等待时间

• 总的来说，在 CPython 中，多线程不能提升 CPU-bound 任务的计算性能，但可以显著提升 IO-bound 任务的并发能力

多进程

• 使用 multiprocessing 多进程实现真正并行，独立内存，开销大，适合计算密集型

• multiprocessing 的核心思想很简单粗暴：既然 GIL 是解释器级别的，那么就启动多个解释器进程

• 这样，每个进程都有自己独立的 Python 解释器、自己的 GIL、自己的地址空间，于是可以在多核 CPU 上真正并行执行 Python 代码

• 这也是为什么 multiprocessing 特别适合 CPU-bound 任务，比如数值计算、模型推理的前后处理、特征工程等，每个进程在一个核心上跑，CPU 利用率可以拉满

• CPU-bound 示例

  from multiprocessing import Process
  
  def cpu_task():
      x = 0
      for _ in range(10**7):
          x += 1
  
  processes = [Process(target=cpu_task) for _ in range(4)]
  for p in processes:
      p.start()
  for p in processes:
      p.join()

• 在这个例子中，可以真正使用多核 CPU，执行时间明显下降（相对于线程）

• 但是代价同样很现实，由于进程之间不共享内存（需要序列化 /pickle），通信必须通过 IPC，比如 Pipe、Queue、共享内存或序列化对象，而进程创建和上下文切换的成本也远高于线程

• 在 multiprocessing 里频繁传输大对象，性能往往会被序列化和拷贝拖垮，甚至比单进程还慢

• 所以 multiprocessing 的使用哲学是：任务要足够 “重”，计算时间要远大于进程管理和通信成本，才值得使用它，否则只是把系统复杂度和资源消耗堆高了

协程

• 异步协程（coroutine）：async/await, asyncio，单线程内切换，用户态调度，极高并发，IO 密集型首选

• 协程解决的是另一个层次的问题，它不是为了 “并行利用多核”，而是为了 “用极低的开销管理大量并发 IO 任务”

• 在 asyncio 体系中，协程本质上是用户态的可暂停函数，而 async/await 并不是语法糖那么简单，它定义了一种显式的 “挂起点”：当协程执行到 await 一个 IO 操作时，它会把控制权交还给事件循环，由事件循环去调度其他协程运行

• 关键点在于：调度权不在操作系统，而在用户态的事件循环，即协程切换不需要线程上下文切换，不需要保存寄存器、切换栈，只是 Python 对象状态的切换，成本极低，极其适合高并发 IO 场景

• 总的来说，协程不是线程，也不是进程，而是用户态调度的并发单元

• 简单示例

  import asyncio
  
  async def task_block():
      time.sleep(1)   # ❌ 阻塞
      print("done")
  
  # async def 定义协程函数
  # 调用后返回协程对象，而不是立即执行
  async def task():
      # await：挂起当前协程，把控制权交还给事件循环
      # await 会阻塞“当前协程”的后续代码，但不会阻塞整个线程，也不会阻塞其他协程
      await asyncio.sleep(1)
      print("done")
  
  async def main():
      # task() 创建两个协程对象
      # asyncio.gather 把它们包装成任务
      # 事件循环可以在两个 task 之间切换
      await asyncio.gather(task(), task())
  
  # 创建事件循环
  # 运行 main 协程
  # 所有协程完成后关闭事件循环
  asyncio.run(main())
  
  # 两个 task 并发执行，时间线是：
  # t=0s  task1 遇到 await → 挂起
  # t=0s  task2 遇到 await → 挂起
  # t=1s  task1 恢复 → print
  # t=1s  task2 恢复 → print
  # 总耗时 ≈ 1 秒，而不是 2 秒
  # 而假如运行了 task_block，那么 time.sleep 不会让出控制权，事件循环被卡死，其他协程无法运行，总耗时应该是 2 秒

• 需要特别强调的是：协程并不是自动并行，因为事件循环本身仍然运行在一个线程里，依然受 GIL 约束；只不过在 IO-bound 场景下，通过显式的 await，把 “等 IO 的时间” 让给其他任务，从而把单线程的效率榨到极限

• 可以这么理解，asyncio 是一个调度和运行协程的框架，核心是事件循环（event loop）、任务（Task）、协程（Coroutine）；即 coroutine 是 “能被挂起的函数”，asyncio 是 “决定什么时候挂起、什么时候恢复它们的调度器”

• 事件循环（Event Loop）调度原理

  • 线程和进程的调度由操作系统内核完成，属于抢占式调度，时间片、优先级、上下文切换都是 OS 决定的
  • asyncio 通过事件循环维护一个任务队列，当协程在 await IO 时主动让出控制权，事件循环调度其他协程执行
  • 因此，协程的调度由用户态事件循环完成，属于协作式调度，只有在 await 时才会让出执行权
  • 这就是为什么协程代码里必须 “自觉”，否则会饿死其他任务

• 一个事件循环内部一般有：

  • 就绪队列：马上能跑的任务
  • 等待队列：在等 IO / 时间的任务
  • IO 多路复用器：epoll 监听文件描述符
  • 定时器队列：处理 call_later、sleep
  • 回调处理：完成后执行回调

• 协程适合的场景：网络请求、文件 IO、数据库 IO、高并发爬虫、API 服务

• 但协程不适合 CPU-bound，因为 CPU 计算不会 await，会阻塞整个事件循环，例如在协程里写了一个纯 Python 的死循环或者重计算，没有 await，那整个事件循环都会被卡死

• CPU bound 的反例

  import asyncio
  import time
  
  def cpu_task():
      x = 0
      for _ in range(10_000_000):
          x += 1
      return x
  
  async def main():
      cpu_task()
  
  asyncio.run(main())

• 非要用 asyncio 跑 CPU 任务的示例

  loop = asyncio.get_running_loop()
  result = await loop.run_in_executor(
      None, cpu_task
  )

• 这个例子中，asyncio 负责调度，而 CPU 任务交给 ThreadPoolExecutor（仍受 GIL 限制）或 ProcessPoolExecutor（真正并行）

• 同步 Synchronous 与异步 Asynchronous 的区别

  • 同步与异步并不是 “有没有线程” 这么肤浅，同步的核心特征是调用方必须等待结果才能继续执行，而异步的核心特征是调用发起后立即返回，结果通过回调、future、await 等方式在未来某个时间点获取
  • 在同步模型里，控制流是线性的，思维负担低，但资源利用率容易很差；在异步模型里，控制流是分离的，代码复杂度上升，但可以把等待时间充分利用起来
  • async/await 的价值就在于，它把异步控制流 “重新写成看起来像同步” 的形式，降低了认知成本

• CPU-bound vs IO-bound

  • 如果任务瓶颈在 CPU 计算上，用多进程
  • 如果瓶颈在 IO 等待上，用多线程或协程
  • 如果是大规模、高并发、IO 密集型服务，协程通常是最优解
  • 如果是混合型任务，工程上往往是 “多进程 + 协程” 或 “多进程 + 多线程” 的组合，比如每个进程跑一个事件循环

• 多线程与协程的对比

  特性	多线程	协程 + 事件循环
  调度者	OS 内核	用户态事件循环
  切换成本	高（陷入内核）	极低（函数跳转）
  并行	多核，但 GIL 限制	单核并发
  锁	需要，易死锁	不需要
  阻塞	阻塞会卡住线程	阻塞自动挂起
  稳定性	一个线程崩可能拖垮进程	一个协程崩不影响循环

Python 内存管理与垃圾回收

• Python 的内存管理不是 JVM 那种 “纯 GC”，而是一个混合模型：

  • 第一层：引用计数（Reference Counting）
  • 第二层：垃圾回收器（Garbage Collector）处理循环引用
  • 第三层：对象池、内存池

• Python 对象 “何时被销毁”，首先由引用计数决定；只有当引用计数机制失效（循环引用）时，GC 才介入

• 引用计数

  • 每个 Python 对象内部都有一个计数器，记录当前有多少个名字 / 容器 / 结构引用了它

  • 在下列例子中，这个列表对象的引用计数是 3

    a = []
    b = a
    c = a

  • 引用计数会在以下情况下变化：

    • 赋值：b = a → +1
    • 删除引用：del b → -1
    • 函数参数传递 → +1
    • 函数返回结束 → -1
    • 放入容器（list /dict）→ +1
    • 从容器移除 → -1

  • 当引用计数变为 0 时，对象立即被销毁，内存立刻可复用

  • 这是 Python 的一个重要特性，也是很多 C 扩展依赖的行为

• del

  • del 删除的是 “名字”，不是对象，其本质是减少一次引用计数

  • 只有在全部引用消失后，对象才会被释放

    del a
    del b
    del c

  • 所以正确表述是 del 并不保证释放内存，只是减少引用

• 循环引用问题

  • 循环引用举例

    class Node:
        def __init__(self):
            self.other = None
    
    a = Node()
    b = Node()
    
    a.other = b
    b.other = a

  • 此时 a 引用 b，b 引用 a，即使 del 了两个对象，两个对象的引用计数仍然不是 0

  • 引用计数解决不了循环引用，因为引用计数是 “局部视角”，它只知道 “有没有人指向我”，不知道 “这一圈人是不是已经和外部世界断开了”

• 垃圾回收（GC）：专门清理循环引用

  • Python 的 GC 只关心容器对象，如 list、dict、set、自定义对象（有 __dict__）

  • 而不关心 int、float、str、tuple（不可变），因为只有容器才可能形成引用环

  • GC 会定期扫描对象图，找出 “彼此引用但从根对象不可达” 的那一组对象，然后统一回收，而 “根对象” 包括：

    • 全局变量
    • 当前栈帧
    • 模块对象
    • 活跃线程

  • Python 使用分代 GC：

    • 第 0 代：新对象，回收最频繁

    • 第 1 代

    • 第 2 代：老对象，回收最少

    • 核心假设是大多数对象 “活得短，死得快”

    • 这也是为什么短生命周期对象不用太担心 GC 成本

  • 如果一个对象定义了 __del__ 方法，并且它参与了循环引用，那么 Python 无法确定安全的销毁顺序，GC 会把它们放入 gc.garbage，因此会导致内存泄漏风险，所以工程经验是尽量避免在有复杂引用关系的对象中定义 __del__，而资源释放应该使用上下文管理器

    class A:
        def __del__(self):
            print("cleanup")

• weakref

  • weakref 创建的是弱引用，不增加引用计数，不影响对象生命周期

    import weakref
    
    class A:
        pass
    
    a = A()
    r = weakref.ref(a)
    
    del a
    print(r())  # None

  • 用于缓存：weakref 让 “缓存不拥有对象”，只观察对象是否还活着

    import weakref
    
    cache = weakref.WeakValueDictionary()
    
    class Data:
        pass
    
    d = Data()
    cache["x"] = d
    
    del d  # 对象消失，缓存自动清理

  • 用于 Observer / 回调系统，防止监听者被意外延长生命周期

• 如何避免内存泄漏

  • 常见泄漏源头

    • 全局变量无限增长

    • 缓存不设上限

    • 闭包捕获大对象

    • 循环引用 + __del__

    • 长生命周期对象持有短生命周期资源

  • 在 Python 中避免内存泄漏的关键不是 “手动 free”，而是控制引用关系，明确对象的拥有者

  • 具体做法包括：

    • 用 with 管理资源

    • 缓存使用 weakref 或 LRU

    • 闭包中避免捕获大对象

    • 明确对象作用域

    • 长服务中定期监控内存

数据模型与魔法方法

• Python 的数据模型（Data Model）是 Python 的核心机制之一，它定义了 Python 对象的行为规范，也就是对象如何响应操作符、内置函数和语法结构

• 理解数据模型就是理解 Python “如何工作” 的基础，而魔法方法（Magic Methods，也叫 dunder 方法，即双下划线方法 __xxx__）就是数据模型的具体体现，它们允许开发者自定义对象在各种操作下的行为

• 什么是数据模型

  • Python 数据模型是一套协议（protocols），它定义了对象在不同上下文下的行为，例如：
    • 当对象参与运算时（+, -, * 等）
    • 当对象被迭代时（for x in obj）
    • 当对象被索引或切片时（obj[0:3]）
    • 当对象被调用时（obj()）
    • 当对象被打印或转换为字符串时（str(obj)、repr(obj)）
    • 当对象被比较时（==, <, > 等）
  • 而魔法方法就是这些协议的具体接口，Python 内置类型（如 int, list, dict）都是通过实现这些魔法方法来定义行为的
  • 可以通过继承或自定义实现魔法方法，让自定义类 “像内置类型一样工作”

• 构造与表示相关的魔法方法

  • __init__(self, ...)：初始化对象，构造函数

  • __new__(cls, ...)：创建对象实例，通常用在自定义元类或不可变对象

  • __del__(self)：析构函数，Python 的垃圾回收会调用它，但通常不建议依赖

  • __repr__(self)：返回对象的官方字符串表示，调试和交互式环境使用

  • __str__(self)：返回对象的用户友好字符串表示

  • __bytes__(self)：对象转换为 bytes

  • __format__(self, format_spec)：格式化对象时调用（如 "{:0.2f}".format(obj)）

  • __hash__(self)：对象用于哈希表（如 set、dict key）的哈希值

  • __bool__(self)：对象转换为布尔值时调用

  • __len__(self)：对象长度，如 len(obj)

  • 示例

    class Point:
        def __init__(self, x, y):
            self.x = x
            self.y = y
    
        def __repr__(self):
            return f"Point({self.x}, {self.y})"
    
        def __str__(self):
            return f"({self.x}, {self.y})"
    
        def __bool__(self):
            return self.x != 0 or self.y != 0
    
    p = Point(1, 2)
    print(repr(p))  # Point(1, 2)
    print(str(p))   # (1, 2)
    print(bool(p))  # True

• 数值运算相关的魔法方法

  • Python 支持重载算术运算符，魔法方法通常分为两类：普通运算和反向运算

  • 普通运算：__add__、__sub__、__mul__、__truediv__、__floordiv__、__mod__、__pow__ 等

  • 反向运算：__radd__、__rsub__、__rmul__ 等，当左侧对象不支持该操作时调用

  • 就地运算（改变自身）：__iadd__、__isub__ 等，对应 +=、-= 等

  • 示例

    class Vector:
        def __init__(self, x, y):
            self.x = x
            self.y = y
    
        def __add__(self, other):
            return Vector(self.x + other.x, self.y + other.y)
    
        def __iadd__(self, other):
            self.x += other.x
            self.y += other.y
            return self
    
        def __repr__(self):
            return f"Vector({self.x}, {self.y})"
    
    v1 = Vector(1, 2)
    v2 = Vector(3, 4)
    print(v1 + v2)  # Vector(4, 6)
    v1 += v2
    print(v1)       # Vector(4, 6)

• 容器类型和序列相关的魔法方法

  • Python 容器类型（如 list、dict、tuple、set）都实现了一系列特殊方法，这让它们可以被索引、迭代、切片、检查成员等

  • __getitem__(self, key)：通过索引或 key 获取元素

  • __setitem__(self, key, value)：通过索引或 key 设置元素

  • __delitem__(self, key)：删除元素

  • __iter__(self)：返回可迭代对象（迭代器）

  • __next__(self)：迭代器的下一个元素

  • __contains__(self, item)：in 操作符

  • __reversed__(self)：反向迭代

  • 示例

    class MyList:
        def __init__(self, items):
            self.items = list(items)
    
        def __getitem__(self, index):
            return self.items[index]
    
        def __setitem__(self, index, value):
            self.items[index] = value
    
        def __delitem__(self, index):
            del self.items[index]
    
        def __len__(self):
            return len(self.items)
    
        def __iter__(self):
            return iter(self.items)
    
    l = MyList([1, 2, 3])
    print(l[0])    # 1
    l[1] = 20
    print(list(l)) # [1, 20, 3]
    del l[2]
    print(list(l)) # [1, 20]

• 属性访问相关的魔法方法

  • Python 对对象属性的访问提供了完整钩子：

  • __getattr__(self, name)：访问不存在的属性时调用，常用在懒加载、动态属性计算或 “代理缺失属性”

  • __getattribute__(self, name)：访问任意属性时调用（要小心避免无限递归），常用在访问日志、权限控制、动态代理、缓存机制

  • __setattr__(self, name, value)：给属性赋值时调用

  • __delattr__(self, name)：删除属性时调用

  • __dir__(self)：自定义 dir(obj) 返回列表

  • 示例

    class A:
        def __getattr__(self, name):
            return f"{name} 属性不存在，但我拦截了它"
    
    a = A()
    print(a.foo)  # foo 属性不存在，但我拦截了它

  • 注意，__getattr__ 仅在访问不存在的属性时才会被调用，而 __getattribute__ 在每次访问任意属性时都会调用

    class Demo:
        def __init__(self):
            self.existing = 42
    
        def __getattr__(self, name):
            print(f"__getattr__ called for {name}")
            return f"{name} not found!"
    
        def __getattribute__(self, name):
            print(f"__getattribute__ called for {name}")
            return super().__getattribute__(name)
    
    d = Demo()
    print(d.existing)  # __getattribute__ called for existing -> 42
    print(d.missing)   # __getattribute__ called for missing
                       # __getattr__ called for missing -> missing not found!

  • 因为 __getattribute__ 拦截所有属性访问，如果在它内部直接访问 self.xxx，会再次触发 __getattribute__，导致无限递归

    class Bad:
        def __getattribute__(self, name):
            # 直接访问 self.name -> 无限递归
            return self.name
    
    # b = Bad()
    # b.any_attr  # 会触发 RecursionError
    
    # 正确做法：使用 super().__getattribute__ 或 object.__getattribute__
    class Good:
        def __init__(self):
            self.x = 10
    
        def __getattribute__(self, name):
            print(f"Accessing {name}")
            return super().__getattribute__(name)
    
    g = Good()
    print(g.x)  # Accessing x -> 10

• 可调用对象和上下文管理

  • __call__(self, *args,**kwargs)：让对象像函数一样被调用

  • __enter__(self)、__exit__(self, exc_type, exc_val, exc_tb)：支持 with 上下文管理

  • 示例

    class Adder:
        def __init__(self, n):
            self.n = n
    
        def __call__(self, x):
            return x + self.n
    
    add5 = Adder(5)
    print(add5(10))  # 15
    
    class Managed:
        def __enter__(self):
            print("Enter")
            return self
    
        def __exit__(self, exc_type, exc_val, exc_tb):
            print("Exit")
    
    with Managed():
        print("Inside")
    # 输出：
    # Enter
    # Inside
    # Exit

• 比较运算和布尔运算

  • __eq__, __ne__, __lt__, __le__, __gt__, __ge__：比较操作

  • __bool__：用于布尔上下文

  • __len__：如果没有定义 __bool__，Python 会尝试通过 __len__ 判断真假

  • 示例：

    class Person:
        def __init__(self, age):
            self.age = age
    
        def __lt__(self, other):
            return self.age < other.age
    
    p1 = Person(20)
    p2 = Person(30)
    print(p1 < p2)  # True

Python 的动态特性

• 在 Python 中，对象的属性和方法可以在运行时动态添加，这是 Python 对象和类在内存中是字典结构的结果，每个对象都有 __dict__ 保存属性

  class A:
      def __init__(self):
          self.x = 10
  
  a = A()
  # 动态添加属性
  a.y = 20
  print(a.x, a.y)  # 10 20
  
  # 动态添加方法
  def greet(self, name):
      return f"Hello, {name}!"
  
  import types
  a.say_hello = types.MethodType(greet, a)
  print(a.say_hello("Alice"))  # Hello, Alice!

• 在上面这个例子中，对象的属性可以随时增加、修改或删除；而对象的方法本质上是函数与对象的绑定，通过 MethodType 可以动态绑定

• Python 提供了一套内置函数用于动态访问和操作对象属性

  • getattr(obj, name[, default])：获取属性值，属性不存在可以返回默认值

  • setattr(obj, name, value)：设置属性值

  • hasattr(obj, name)：检查对象是否有某个属性

  • delattr(obj, name)：删除属性

  • 示例

    class B:
        pass
    
    b = B()
    setattr(b, "name", "Bob")
    print(getattr(b, "name"))  # Bob
    print(hasattr(b, "age"))   # False
    delattr(b, "name")
    print(hasattr(b, "name"))  # False
    
    class C:
        def __getattr__(self, item):
            return f"{item} 不存在，但被拦截了"
    
    c = C()
    print(c.foo)  # foo 不存在，但被拦截了

  • 实际上，getattr 会触发对象的 __getattr__ 或 __getattribute__；setattr 会触发 __setattr__；delattr 会触发 __delattr__

• 动态代理

  • 动态代理指通过一个对象拦截另一个对象的方法调用或属性访问，然后执行自定义逻辑，这在日志、权限控制、缓存、RPC 等场景中非常常见

  • 实现方式通常是重写魔法方法：

    • __getattr__：拦截属性访问
    • __setattr__：拦截属性赋值
    • __call__：拦截对象调用

  • 示例

    class Proxy:
        def __init__(self, target):
            self._target = target
    
        def __getattr__(self, name):
            print(f"访问属性 {name}")
            return getattr(self._target, name)
    
        def __setattr__(self, name, value):
            if name.startswith("_"):
                super().__setattr__(name, value)
            else:
                print(f"设置属性 {name} = {value}")
                setattr(self._target, name, value)
    
    class Person:
        def __init__(self, name):
            self.name = name
    
    p = Person("Alice")
    proxy_p = Proxy(p)
    print(proxy_p.name)  # 访问属性 name -> Alice
    proxy_p.name = "Bob" # 设置属性 name = Bob
    print(p.name)        # Bob

  • 在这个例子中， Proxy 拦截了所有对 p 的属性访问和修改，实现了动态代理

• Mock 对象

  • 在测试中，经常需要创建一个 “假对象”，用于替代真实对象的方法或属性，而使用 Python 的动态特性和魔法方法可以轻松实现

    class Mock:
        def __init__(self):
            self._calls = []
    
        def __getattr__(self, name):
            def _mock(*args,**kwargs):
                self._calls.append((name, args, kwargs))
                print(f"调用了 {name}，参数：{args}, {kwargs}")
            return _mock
    
    mock_obj = Mock()
    mock_obj.foo(1, 2)  # 调用 foo，参数：(1, 2), {}
    mock_obj.bar(x=10)  # 调用 bar，参数：(), {'x': 10}

  • __getattr__ 返回一个函数，使得任意方法调用都被拦截

  • _calls 可以记录所有方法调用历史，方便测试断言

上下文管理器（with 语句）

• with 语句是 Python 提供的一种管理资源的优雅方式，主要作用是：

  • 确保资源正确获取与释放（如文件、锁、网络连接等）

  • 自动处理异常，保证退出上下文时资源释放

  • 基本语法：

    with <表达式> as <变量>:
        # 使用资源的代码块

• with 语句的工作流程：

  • 调用 <表达式>，返回一个上下文管理器对象（实现了 __enter__ 和 __exit__ 方法）
  • 执行 __enter__ 方法，返回值赋给 as 后的变量
  • 执行代码块
  • 无论代码块是否抛出异常，都会执行 __exit__ 方法，进行清理

• 文件操作

  # 传统写法
  f = open("test.txt", "w")
  try:
      f.write("Hello World")
  finally:
      f.close()  # 必须显式关闭，否则可能泄漏资源
  
  # with 写法
  # 避免忘记 close()，即使发生异常也会自动关闭
  with open("test.txt", "w") as f:
      f.write("Hello World")  # 自动管理资源，代码块结束自动 close

• 锁（线程同步）

  import threading
  
  # threading.Lock 实现了上下文管理器
  lock = threading.Lock()
  
  # 传统方式
  lock.acquire()
  try:
      # 临界区代码
      pass
  finally:
      lock.release()
  
  # with 语法
  # with 自动 acquire 和 release
  with lock:
      # 临界区代码
      pass

• 自定义上下文管理器，需要实现两个方法：

  class MyContext:
      def __enter__(self):
          print("进入上下文")
          return self  # 可返回资源对象
  
      def __exit__(self, exc_type, exc_val, exc_tb):
          print("退出上下文")
          if exc_type:
              print(f"处理异常: {exc_type}, {exc_val}")
          return True  # True 表示异常被处理，不会抛出
  
  with MyContext() as ctx:
      print("使用资源")
      # raise ValueError("测试异常")  # 如果取消注释，会被 __exit__ 捕获
  
  # 进入上下文
  # 使用资源
  # 退出上下文 或者 处理异常: <class 'ValueError'>, 测试异常

• 在上面的例子中，__exit__ 的四个参数的含义

  • exc_type：异常类型
  • exc_val：异常实例
  • exc_tb：traceback 对象
  • 如果返回 True，异常会被抑制；返回 False 或 None，异常会继续抛出

• Python 提供了 contextlib 模块，可以用装饰器方式创建上下文管理器，比实现类更简洁

  from contextlib import contextmanager
  
  @contextmanager
  def my_context():
      print("进入上下文")
      try:
          yield "资源对象"
      except Exception as e:
          print(f"捕获异常: {e}")
      finally:
          print("退出上下文")
  
  with my_context() as resource:
      print("使用资源:", resource)
      # raise ValueError("测试异常")
  
  # 进入上下文
  # 使用资源: 资源对象
  ## 捕获异常: 测试异常
  # 退出上下文

• 在这个例子中，yield 左边相当于 __enter__ 返回值，而 finally 块相当于 __exit__ 清理逻辑

• 数据库连接上下文管理器

  class DBConnection:
      def __enter__(self):
          print("连接数据库")
          self.conn = "DB Connection"  # 模拟数据库连接
          return self.conn
  
      def __exit__(self, exc_type, exc_val, exc_tb):
          print("关闭数据库连接")
          if exc_type:
              print(f"异常: {exc_type}, {exc_val}")
  
  with DBConnection() as conn:
      print("使用数据库:", conn)
      # raise ValueError("模拟异常")

元类（Metaclass）

• Python 中类也是对象，而创建类的 “类” 就是元类（Metaclass）；类在被创建时，会调用元类来生成类对象

  对象 -> 类 -> 元类
  对象是实例、类是对象、元类是类的类

• type 是 Python 的内置原类

  • 获取对象的类型或类

    x = 10
    print(type(x))  # <class 'int'>

  • type 也可以动态创建类

    MyClass = type("MyClass", (object,), {"x": 5, "foo": lambda self: print("Hello")})
    
    obj = MyClass()
    print(obj.x)  # 5
    obj.foo()     # Hello

  • 在上述例子中，"MyClass" 即为类名，(object,)是继承的父类元，{"x": 5, "foo": ...} 是类的属性和方法

  • 本质上，type 就是创建类对象的 “工厂函数”

• 通过继承 type，可以定制类的创建行为

  • 元类的基础结构

    class MyMeta(type):
        def __new__(cls, name, bases, dct):
            print(f"正在创建类 {name}")
            # 可以修改类属性
            dct['added_attr'] = 100
            return super().__new__(cls, name, bases, dct)

  • 上述例子中，__new__ 在类创建之前调用，负责返回类对象，其参数

    • cls：元类自身
    • name：类名
    • bases：父类元组
    • dct：类属性字典

  • 使用自定义元类

    class MyClass(metaclass=MyMeta):
        x = 10
    
    print(hasattr(MyClass, 'added_attr'))  # True
    obj = MyClass()
    print(obj.x)           # 10
    print(obj.added_attr)  # 100

  • 可以看出，元类在类创建阶段修改了类属性

  • 注意，元类影响的是类，而不是实例

• 元类的作用场景

  • 修改类属性或方法：可以在类创建时统一增加方法、属性或装饰器

  • 注册类：比如实现插件机制，把所有子类注册到一个列表

    class RegistryMeta(type):
        registry = []
        def __new__(cls, name, bases, dct):
            new_cls = super().__new__(cls, name, bases, dct)
            cls.registry.append(new_cls)
            return new_cls
    
    class PluginBase(metaclass=RegistryMeta):
        pass
    
    class PluginA(PluginBase):
        pass
    
    class PluginB(PluginBase):
        pass
    
    print(RegistryMeta.registry)
    # [<class '__main__.PluginBase'>, <class '__main__.PluginA'>, <class '__main__.PluginB'>]

  • 验证类定义：可以检查类属性或方法名是否合法

• 元类与实例化的关系

  • 元类作用于类创建，而类作用于对象实例化

  • 实例化对象时，不会再调用元类，元类只在类生成阶段参与

  • 执行顺序如下

    元类.__new__  -> 元类.__init__ -> 类定义完成 -> 类实例化 -> 实例.__new__ -> 实例.__init__

模块与包机制

• import module

  • 举例

    import math
    print(math.sqrt(16))

  • 这条语句做了三件事：

    • 查找并加载模块对象

    • 将模块对象绑定到当前命名空间的名字 math

    • 后续访问模块内容，必须通过 math.xxx

  • math 是一个 模块对象的引用，不是把模块里的名字 “拷贝” 进来

  • 其命名空间清晰，不容易污染当前作用域

• from module import name

  • 示例

    from math import sqrt
    print(sqrt(16))
    
    sqrt(16)     # OK
    math.sqrt()  # NameError

  • 这条语句的语义是：

    • 加载模块（如果尚未加载）

    • 从模块对象中取出 sqrt

    • 把 sqrt 直接绑定到当前命名空间

  • 关键区别在于：sqrt 是一个名字绑定，不是模块引用

  • 名字遮蔽

    from math import sqrt
    
    def sqrt(x):
        return x + 1
    
    print(sqrt(9))  # 10 —— math.sqrt 被遮蔽

  • 模块更新不反映：如果之后 a.x = 20，b.x 不会变，因为 x 是拷贝绑定

    # a.py
    x = 10
    
    # b.py
    from a import x

• as：别名机制

  • 示例

    import numpy as np
    from collections import defaultdict as dd

  • as 的本质是改变当前命名空间中绑定的名字，不影响模块本身

  • 可以用于缩短长模块名、解决命名冲突、提高可读性

• 模块缓存：sys.modules

  • 模块只会被加载一次

    import sys
    import math
    
    print("math" in sys.modules)  # True
    print(sys.modules["math"])

  • sys.modules 是一个字典，key 为模块名（字符串），value 为模块对象

  • Python 的 import 本质是：如果模块名已经存在于 sys.modules，直接返回该模块对象，否则才真正去加载并执行模块代码

  • 验证 “模块只执行一次”

    # a.py
    print("a is imported")
    x = 10
    import a
    import a
    
    # a is imported

  • 第一次 import：执行 a.py，放入 sys.modules；第二次 import：直接从缓存拿模块对象

  • 这也是 Python 能容忍部分循环依赖的基础

• 包（Package）与 __init__.py

  • 什么是包

    • 一个目录，只要满足以下条件之一：

      • Python 3.3+：目录即可（隐式 namespace package）

      • 传统做法：目录中有 __init__.py

    • 例如：

      mypkg/
      ├── __init__.py
      ├── a.py
      └── b.py

  • init__.py 在包第一次被导入时执行

    • 示例

      import mypkg

    • 等价于创建包对象 mypkg，并执行 mypkg/__init__.py

  • 作用一：包初始化逻辑

    # mypkg/__init__.py
    print("mypkg initialized")
    import mypkg
    
    # mypkg initialized

  • 作用二：控制 from pkg import *

    # mypkg/__init__.py
    __all__ = ["a"]
    
    from . import a
    from . import b
    from mypkg import *
    
    ## 只会导入 `a`，不会导入 `b`

  • 作用三：对外暴露统一接口

    # mypkg/__init__.py
    from .a import func_a
    from .b import func_b
    
    # 使用方式
    from mypkg import func_a, func_b

• 模块导入顺序

  • 当执行 import mypkg.a 时，Python 的顺序是：

    • 查找 mypkg 是否在 sys.modules

    • 若不在，创建包对象并执行 mypkg/__init__.py

    • 查找并加载 mypkg.a

    • 执行 a.py

    • 将 mypkg.a 放入 sys.modules

  • 包先于子模块初始化，且 __init__.py 一定先执行

• 循环依赖（Circular Import）

  • 经典循环依赖示例

    # a.py
    import b
    x = 1
    
    # b.py
    import a
    print(a.x)
    
    # main.py
    import a
    # AttributeError: partially initialized module 'a'

  • 当导入 a 时：

    • Python 创建模块对象 a，放入 sys.modules
    • 执行 a.py
    • 遇到 import b
    • Python 开始导入 b
    • b.py 中 import a
    • 从 sys.modules 中拿到尚未执行完的 a
    • 此时 a.x 还没定义 → 出错

  • 核心原因在于，模块在 “执行完成之前” 就已经被放进 sys.modules

  • 方式一：延迟导入（推荐），即把 import 放进函数体，等真正需要时再执行

    # b.py
    def func():
        import a
        print(a.x)

  • 方式二：只导入模块，不访问未初始化属性，即不要在模块顶层使用 a.x

    # b.py
    import a

  • 方式三：重构公共依赖

    # common.py
    x = 1
    
    # a.py
    import common
    
    # b.py
    import common