Python 高级面试题

核心答案 GIL(Global Interpreter Lock)是 CPython 解释器中的一把互斥锁,确保同一时刻只有一个线程执行 Python 字节码。 深入分析 存在原因: 【注解】 GIL 是 CPython 实现细节,不是 Python 语言规范。Jython(JVM 实现)和 IronPython(.NET 实现)没有 GIL。PyPy 有自己的 STM(Software Transactional Memory)方案。 追问方向 核心答案 深入分析 GIL 释放时机: 解决方案: 代码示例 【注解】 误区:认为 Python 多线程完全

分享

官方文档:https://docs.python.org/3/
最后更新:2026-05-08

一、GIL(全局解释器锁)

Q1: 什么是 GIL?为什么存在?

核心答案

GIL(Global Interpreter Lock)是 CPython 解释器中的一把互斥锁,确保同一时刻只有一个线程执行 Python 字节码。

深入分析

存在原因:

  • CPython 使用引用计数进行内存管理,ob_refcnt 字段需要在多线程环境中保持一致
  • GIL 使引用计数的增减成为原子操作,避免竞态条件导致内存错误
  • 简化了 CPython 实现,方便与 C 扩展集成(很多 C 扩展库假定 GIL 存在)

【注解】

GIL 是 CPython 实现细节,不是 Python 语言规范。Jython(JVM 实现)和 IronPython(.NET 实现)没有 GIL。PyPy 有自己的 STM(Software Transactional Memory)方案。

追问方向

  • GIL 对 I/O 密集型和 CPU 密集型有何不同影响?
  • 如何绕过 GIL 实现真正的并行?

Q2: GIL 对多线程的影响?何时释放?

核心答案

  • CPU 密集型:多线程无效甚至更慢(线程切换开销 + GIL 竞争)
  • I/O 密集型:线程在等待 I/O 系统调用时主动释放 GIL,其他线程可以运行

深入分析

GIL 释放时机:

  • 每 5ms 执行间隔(Python 3.2+ 后用时间而非指令数,通过 sys.setswitchinterval 设置)
  • 调用阻塞的 I/O 系统调用(read、write、socket 操作等)
  • C 扩展显式使用 Py_BEGIN_ALLOW_THREADS / Py_END_ALLOW_THREADS(如 numpy 的计算操作)

解决方案:

  • multiprocessing:每个进程有独立的 GIL
  • ProcessPoolExecutor:进程池,适合 CPU 密集型
  • numpy、scipy 等 C 扩展:在 C 层释放 GIL 执行计算
  • Cython:可用 with nogil:

代码示例

import threading
import time

def cpu_bound():
    # CPU 密集型:GIL 导致多线程几乎无加速
    count = 0
    for _ in range(10_000_000):
        count += 1
    return count

def io_bound():
    # I/O 密集型:等待时释放 GIL,多线程有效
    time.sleep(1)  # 模拟 I/O 等待

# CPU 密集型:多线程不如单线程
start = time.perf_counter()
t1 = threading.Thread(target=cpu_bound)
t2 = threading.Thread(target=cpu_bound)
t1.start(); t2.start()
t1.join(); t2.join()
print(f'多线程 CPU 密集: {time.perf_counter() - start:.2f}s')
# 通常比单线程串行还慢,因为 GIL 竞争

【注解】

误区:认为 Python 多线程完全无用。对于 I/O 密集型(爬虫、数据库查询、文件操作)多线程是有效的并发手段。

追问方向

  • sys.setswitchinterval 的作用?
  • Python 3.13 free-threaded mode 是什么?

Q3: Python 3.13 free-threaded mode(PEP 703)

核心答案

Python 3.13 引入实验性的 free-threaded 模式,编译时使用 --disable-gil 选项,允许多线程真正并行执行 Python 字节码。

深入分析

主要挑战:

  • 引用计数安全:需要原子操作(使用 Biased Reference Counting 方案)
  • 性能权衡:单线程性能会有一定损耗(约 5-10%),换取多线程并行能力
  • 生态兼容性:依赖 GIL 的 C 扩展需要更新

启用方式(CPython 需要特殊编译版本):

import sys
# 检查是否在 free-threaded 模式
print(sys._is_gil_enabled())  # Python 3.13+

【注解】

PEP 703 是"使 GIL 可选"而非"移除 GIL"。默认仍启用 GIL,确保向后兼容。生产环境大规模使用还需等待生态成熟。

追问方向

  • Biased Reference Counting 的原理?
  • 无 GIL 后如何保证线程安全?

二、内存管理与垃圾回收

Q4: Python 内存管理机制全貌

核心答案

Python 内存管理分为三层:

层次 管理者 负责内容
第一层 OS 物理内存/虚拟内存分配
第二层 CPython pymalloc 小对象(< 512 字节)内存池
第三层 对象分配器 Python 对象的创建与管理

深入分析

pymalloc 内存池层次:

  • Arena(256KB)→ Pool(4KB)→ Block(8~512 字节,8 字节对齐)
  • 小对象(< 512 字节)从内存池分配,避免频繁调用 malloc
  • 大对象直接调用系统 malloc

私有堆(Private Heap):Python 维护自己的堆,与系统堆隔离,所有 Python 对象都存在私有堆上。

【注解】

Python 的内存池设计减少了内存碎片和系统调用频率,提升了小对象分配的性能。但也意味着即使 del 对象后,内存不一定立即归还给 OS(保留在内存池中供复用)。

追问方向

  • 为什么 Python 进程内存只增不减?
  • 引用计数如何与内存池协作?

Q5: 引用计数(Reference Counting)

核心答案

CPython 中每个对象有 ob_refcnt 字段记录引用数量。引用数降为 0 时立即释放内存。

深入分析

引用计数增加的操作:

  • 赋值给变量:x = obj
  • 添加到容器:lst.append(obj)
  • 传入函数:func(obj)
  • 对象别名:y = x

引用计数减少的操作:

  • 变量离开作用域
  • del x
  • 从容器移除

优点:立即回收,内存使用可预测

缺点:

  • 循环引用无法通过引用计数回收
  • ob_refcnt 修改需要 GIL 保护(影响多线程性能)
  • 每个对象额外占用一个字(8 字节)存储引用计数

代码示例

import sys

x = []
print(sys.getrefcount(x))  # 2,而不是 1

# 原因:传入 getrefcount 本身增加了一次引用
# 1 次来自变量 x,1 次来自 getrefcount 的参数

y = x
print(sys.getrefcount(x))  # 3(x、y、getrefcount 参数)

del y
print(sys.getrefcount(x))  # 2(x、getrefcount 参数)

【注解】

sys.getrefcount() 的结果总比预期多 1,这是新手常见困惑点。传参本身是一次引用,函数返回后才释放。

追问方向

  • 循环引用为什么引用计数无法处理?
  • 如何查看一个对象的引用计数?

Q6: 分代垃圾回收(GC)

核心答案

Python 的分代 GC 专门处理引用计数无法解决的循环引用问题,采用标记-清除算法,将对象分为三代。

深入分析

三代设计(基于"弱代假说":年轻对象死得快):

  • Generation 0:新创建的对象,GC 最频繁
  • Generation 1:从 Gen 0 存活下来的对象
  • Generation 2:长期存活的对象,GC 最少

默认阈值 (700, 10, 10)

  • Gen 0 中对象数超过 700 时触发 Gen 0 GC
  • Gen 0 GC 触发 10 次后触发 Gen 1 GC
  • Gen 1 GC 触发 10 次后触发 Gen 2 GC

标记-清除过程:

  1. 找出所有容器对象(list、dict、set、class 实例等)
  2. 遍历引用关系,对每个被引用对象的引用计数临时减一
  3. 引用计数仍 > 0 的对象有外部引用,标记为可达
  4. 引用计数为 0 的对象是孤立循环,标记为垃圾,清除

代码示例

import gc
import sys

# 演示循环引用
a = []
b = [a]
a.append(b)

ref_a = sys.getrefcount(a)  # a 被 b[0] 引用,b 被 a[0] 引用
print(f'a 的引用计数: {ref_a}')  # 3(a 变量 + b[0] + getrefcount 参数)

del a, b
# 此时两个列表仍互相引用,引用计数不为 0
# 分代 GC 会识别出这是孤立循环

collected = gc.collect()  # 手动触发 GC
print(f'回收了 {collected} 个对象')  # 2

# 查看各代对象计数
print(gc.get_count())        # (gen0_count, gen1_count, gen2_count)
print(gc.get_threshold())    # (700, 10, 10)

# 禁用 GC(谨慎!)
gc.disable()
# ... 某些高性能场景
gc.enable()

【注解】

gc.disable() 在某些特定场景下可以提升性能(如 Instagram 的实践),因为 GC 扫描本身有开销。但需要确保代码没有循环引用,或在关键阶段完成后手动 gc.collect()

追问方向

  • 什么类型的对象不会被分代 GC 扫描?
  • 如何手动触发特定代的 GC?

Q7: weakref(弱引用)

核心答案

弱引用不增加对象的引用计数,不影响对象的生命周期,适用于缓存和观察者模式。

深入分析

弱引用主要工具:

  • weakref.ref(obj):创建弱引用,调用时返回对象或 None(对象已被回收)
  • weakref.WeakValueDictionary:值是弱引用的字典
  • weakref.WeakKeyDictionary:键是弱引用的字典
  • weakref.WeakSet:弱引用集合

代码示例

import weakref
import gc

class ExpensiveResource:
    def __init__(self, name):
        self.name = name

    def __repr__(self):
        return f'ExpensiveResource({self.name!r})'

# 带弱引用的对象缓存
class ResourceCache:
    def __init__(self):
        self._cache = weakref.WeakValueDictionary()

    def get(self, name):
        if name in self._cache:
            return self._cache[name]
        resource = ExpensiveResource(name)
        self._cache[name] = resource
        return resource

cache = ResourceCache()
r1 = cache.get('database')
r2 = cache.get('database')
print(r1 is r2)  # True,返回同一个对象

del r1, r2       # 没有强引用了
gc.collect()
r3 = cache.get('database')  # 重新创建,因为之前的被回收了
print(r3)        # ExpensiveResource('database')

# 基本弱引用用法
obj = ExpensiveResource('test')
weak = weakref.ref(obj)
print(weak())    # ExpensiveResource('test')
del obj
print(weak())    # None

【注解】

并非所有对象都支持弱引用。intstrtuple 等内置类型默认不支持。自定义类通过添加 __weakref__ 属性支持(默认已有)。使用 __slots__ 时需要显式包含 '__weakref__'

追问方向

  • weakref 如何实现"对象被回收时通知我"?(finalizer callback)
  • WeakValueDictionary 适合哪些场景?

Q8: 内存泄漏排查

核心答案

Python 中的内存泄漏常见原因:全局容器积累引用、未取消的回调/信号、循环引用(GC 配置问题)。

深入分析

常见内存泄漏场景:

  • 全局缓存(列表/字典)不断增长但不清理
  • 事件系统中注册了监听器但忘记取消注册
  • 线程局部存储(threading.local)积累
  • C 扩展内存管理 bug

排查工具:

工具 用途
tracemalloc 跟踪内存分配,定位到源码行
objgraph 显示对象引用关系图
memory_profiler 逐行内存使用分析
gc.get_objects() 列出所有 GC 跟踪的对象

代码示例

import tracemalloc
import gc

tracemalloc.start()

# 模拟泄漏
leaky_cache = []
for i in range(10000):
    leaky_cache.append({'data': 'x' * 1000})

snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')

print('内存占用 Top 5:')
for stat in top_stats[:5]:
    print(stat)

# objgraph 示例(需安装)
# import objgraph
# objgraph.show_most_common_types(limit=10)
# objgraph.show_growth()  # 显示增长最多的类型

【注解】

Python 进程内存只增不减是常见误解:即使对象被回收,pymalloc 内存池不一定归还给 OS。tracemalloc 追踪的是 Python 层面的分配,不能替代进程级内存监控(如 /proc/self/status 中的 RSS)。

追问方向

  • 如何在生产环境中低开销地监控内存?
  • gc.get_objects()objgraph 如何配合使用?

三、元类(Metaclass)

Q9: 什么是元类?type 的角色

核心答案

在 Python 中,类本身也是对象。元类是类的类——创建类对象的类。type 是所有类的默认元类。

深入分析

type 的两种用法:

  • type(obj):返回对象的类型
  • type(name, bases, dict):动态创建新类

类定义的完整执行顺序:

  1. __prepare__:创建类的命名空间(通常是有序字典)
  2. 执行类体,填充命名空间
  3. __new__:用命名空间创建类对象
  4. __init__:初始化类对象

代码示例

# 以下两种写法等价
class Foo:
    x = 1

Foo = type('Foo', (object,), {'x': 1})

# 验证
print(type(Foo))          # <class 'type'>
print(type(type))         # <class 'type'>  type 的元类是它自己
print(isinstance(Foo, type))  # True

# 动态创建类的实际应用
def make_class(name, fields):
    def __init__(self, **kwargs):
        for field in fields:
            setattr(self, field, kwargs.get(field))
    return type(name, (object,), {'__init__': __init__, 'fields': fields})

Point = make_class('Point', ['x', 'y'])
p = Point(x=1, y=2)
print(p.x, p.y)  # 1 2

【注解】

type 既是 object 的子类,object 又是 type 的实例——这是 Python 类型系统的自举(bootstrap)设计,初学者容易陷入这个鸡生蛋的问题。实际使用中不必过度深究,理解"类是对象、元类控制类的创建"即可。

追问方向

  • __prepare__ 返回的命名空间有什么特殊用途?
  • metaclass= 关键字如何指定元类?

Q10: 自定义元类

核心答案

通过继承 type 并重写 __new____init__,可以在类创建时插入自定义逻辑。

深入分析

  • __new__:控制类对象的创建,可以修改类的属性、方法
  • __init__:类对象创建后的初始化
  • __call__:控制类被实例化时的行为(即 MyClass()

代码示例

# 注册模式元类:自动注册所有子类
class PluginMeta(type):
    registry = {}

    def __new__(mcs, name, bases, namespace):
        cls = super().__new__(mcs, name, bases, namespace)
        if bases:  # 不注册基类本身
            mcs.registry[name] = cls
        return cls

class BasePlugin(metaclass=PluginMeta):
    pass

class CsvPlugin(BasePlugin):
    def process(self, data): pass

class JsonPlugin(BasePlugin):
    def process(self, data): pass

print(PluginMeta.registry)
# {'CsvPlugin': <class 'CsvPlugin'>, 'JsonPlugin': <class 'JsonPlugin'>}

# 单例元类
class SingletonMeta(type):
    _instances = {}

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

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

db1 = Database()
db2 = Database()
print(db1 is db2)  # True

【注解】

元类 __new__ 的参数中,mcs 代表元类本身(对应普通类的 cls),namespace 是类体执行后的命名空间字典。命名约定:元类的第一个参数常用 mcsmcls

追问方向

  • 元类的 __new__ 和普通类的 __new__ 有何区别?
  • 如何在元类中强制子类必须实现某些方法?

Q11: 元类 vs __init_subclass__ vs 类装饰器对比

核心答案

三种方式都可以在类创建时插入逻辑,但适用场景和复杂度不同。

深入分析

特性 元类 __init_subclass__ 类装饰器
Python 版本 全版本 3.6+ 全版本
复杂度
自动继承 否(需手动应用)
控制粒度 最全(创建过程) 类创建后 类创建后
与其他元类兼容 可能冲突 无冲突 无冲突

代码示例

# __init_subclass__:更简洁的注册模式(Python 3.6+)
class BasePlugin:
    _registry = {}

    def __init_subclass__(cls, plugin_name=None, **kwargs):
        super().__init_subclass__(**kwargs)
        name = plugin_name or cls.__name__
        BasePlugin._registry[name] = cls

class CsvPlugin(BasePlugin, plugin_name='csv'):
    pass

class JsonPlugin(BasePlugin):  # 使用类名
    pass

print(BasePlugin._registry)
# {'csv': <class 'CsvPlugin'>, 'JsonPlugin': <class 'JsonPlugin'>}

# 类装饰器:一次性增强,不自动传递给子类
def add_repr(cls):
    def __repr__(self):
        attrs = ', '.join(f'{k}={v!r}' for k, v in self.__dict__.items())
        return f'{cls.__name__}({attrs})'
    cls.__repr__ = __repr__
    return cls

@add_repr
class Point:
    def __init__(self, x, y):
        self.x = x
        self.y = y

print(Point(1, 2))  # Point(x=1, y=2)

【注解】

优先考虑 __init_subclass__(Python 3.6+),它足以处理大多数"监控子类创建"的需求。只有在需要控制类创建的早期阶段(如修改 __prepare__ 返回的命名空间、控制 MRO)时才使用元类。

追问方向

  • __init_subclass__ 如何传递参数?
  • dataclass 的实现用了什么机制?

四、描述符(Descriptor)

Q12: 描述符协议

核心答案

描述符是定义了 __get____set____delete__ 中至少一个方法的对象,当作为类属性时可以控制属性访问行为。

深入分析

描述符分类:

类型 定义的方法 优先级
数据描述符 __get__ + __set__(或 __delete__ 高于实例 __dict__
非数据描述符 只有 __get__ 低于实例 __dict__

方法签名:

  • __get__(self, obj, objtype=None)obj 为 None 时表示通过类访问
  • __set__(self, obj, value)
  • __delete__(self, obj)

@property 本质是数据描述符(同时实现了 __get____set____delete__)。

函数对象是非数据描述符(只有 __get__),这是方法绑定的底层机制。

追问方向

  • 为什么实例方法是非数据描述符?
  • staticmethodclassmethod 是如何用描述符实现的?

Q13: 描述符查找顺序

核心答案

属性查找顺序:数据描述符 > 实例 __dict__ > 非数据描述符(及其他类属性)。

深入分析

完整查找流程(obj.attr 触发 type(obj).__getattribute__(obj, 'attr')):

  1. type(obj).__mro__ 中查找 attr
  2. 如果找到且是数据描述符:调用描述符的 __get__,返回结果
  3. 如果 obj.__dict__ 中有 attr:返回实例字典中的值
  4. 如果找到且是非数据描述符:调用描述符的 __get__,返回结果
  5. 如果找到普通类属性:返回该属性
  6. 抛出 AttributeError

代码示例

class DataDescriptor:
    def __get__(self, obj, objtype=None):
        print('数据描述符 __get__')
        return 'descriptor value'
    def __set__(self, obj, value):
        print(f'数据描述符 __set__: {value}')

class NonDataDescriptor:
    def __get__(self, obj, objtype=None):
        print('非数据描述符 __get__')
        return 'non-data descriptor value'

class MyClass:
    data = DataDescriptor()
    non_data = NonDataDescriptor()

obj = MyClass()

# 数据描述符优先于实例字典
obj.__dict__['data'] = 'instance value'
print(obj.data)       # 打印"数据描述符 __get__",返回 descriptor value
                      # 实例字典被忽略!

# 非数据描述符低于实例字典
obj.__dict__['non_data'] = 'instance value'
print(obj.non_data)   # 返回 instance value(实例字典优先)

【注解】

@property 是数据描述符,所以实例字典中的同名属性无法覆盖它。这是一个常见的混淆点:为什么直接修改 obj.__dict__ 绕过了 property?答案是 property 是数据描述符,查找顺序在实例字典之前。

追问方向

  • 如何让 @property 的值存储在实例中而不是类中?

Q14: 自定义描述符实现类型验证

核心答案

通过描述符可以实现声明式的属性类型验证,比在每个 setter 中写 isinstance 更优雅。

代码示例

class TypedField:
    def __set_name__(self, owner, name):
        # Python 3.6+:类创建时自动调用,获取属性名
        self.name = name
        self.private_name = f'_{name}'

    def __init__(self, expected_type):
        self.expected_type = expected_type

    def __get__(self, obj, objtype=None):
        if obj is None:
            return self  # 通过类访问时返回描述符本身
        return getattr(obj, self.private_name, None)

    def __set__(self, obj, value):
        if not isinstance(value, self.expected_type):
            raise TypeError(
                f'{self.name} 必须是 {self.expected_type.__name__},'
                f'收到 {type(value).__name__}'
            )
        setattr(obj, self.private_name, value)

class Person:
    name = TypedField(str)
    age = TypedField(int)

    def __init__(self, name, age):
        self.name = name
        self.age = age

p = Person('Alice', 30)
print(p.name, p.age)  # Alice 30

p.age = 'old'
# TypeError: age 必须是 int,收到 str

【注解】

__set_name__ 是 Python 3.6 引入的,在此之前描述符需要手动传入属性名:name = TypedField('name', str)__set_name__ 大大简化了描述符的使用。注意私有属性名加下划线前缀是为了避免与描述符本身的属性名冲突。

追问方向

  • 描述符如何与 __slots__ 交互?
  • dataclassfield() 是否用了描述符?

五、生成器与协程

Q15: 生成器工作原理

核心答案

生成器函数(含 yield 的函数)被调用时返回生成器对象,而不立即执行函数体。每次调用 next() 执行到下一个 yield,保存完整的栈帧状态。

深入分析

生成器的三个控制方法:

方法 说明
next(gen) / gen.__next__() 执行到下一个 yield,返回 yield 的值
gen.send(value) 将值发送给生成器(作为 yield 表达式的结果),执行到下一个 yield
gen.throw(ExcType) 在暂停点抛出异常
gen.close() 在暂停点抛出 GeneratorExit

生成器状态属性:

  • gi_frame:当前栈帧(挂起时非 None,结束后为 None)
  • gi_code:代码对象
  • gi_running:是否正在执行

代码示例

def countdown(n):
    print('开始倒计时')
    while n > 0:
        received = yield n  # yield 表达式可以接收 send 的值
        print(f'收到: {received}')
        n -= 1
    print('结束')
    return 'done'  # StopIteration 的 value

gen = countdown(3)
print(next(gen))      # 开始倒计时 / 3
print(gen.send('ok')) # 收到: ok / 2
print(next(gen))      # 收到: None / 1

try:
    next(gen)         # 收到: None,然后函数结束
except StopIteration as e:
    print(f'返回值: {e.value}')  # 返回值: done

【注解】

第一次必须用 next(gen)gen.send(None) 启动生成器(因为此时还没有 yield 表达式在等待值)。直接 gen.send('value') 会抛出 TypeError

追问方向

  • 生成器的内存优势体现在哪里?
  • yield from 和直接 yield 有什么区别?

Q16: yield from

核心答案

yield from iterable 将子迭代器的 send/throw/close 操作透明地转发,并将子生成器的 return 值作为 yield from 表达式的结果。

深入分析

yield from 的作用:

  • 等价于但比手动 for item in sub_gen: yield item 功能更强
  • 透明转发 send():外部发送的值直接传给子生成器
  • 透明转发 throw():外部抛入的异常直接抛给子生成器
  • 透明转发 close():关闭信号传给子生成器
  • 子生成器 return 的值成为 yield from 表达式的值(通过 StopIteration.value

代码示例

def sub_gen():
    received = yield 'sub: first'
    print(f'子生成器收到: {received}')
    yield 'sub: second'
    return 'sub done'  # 这个值会传给 yield from 表达式

def delegating_gen():
    result = yield from sub_gen()  # 透明转发
    print(f'子生成器返回: {result}')  # sub done
    yield 'after delegation'

gen = delegating_gen()
print(next(gen))       # sub: first
print(gen.send('hello'))  # 子生成器收到: hello / sub: second
print(next(gen))       # 子生成器返回: sub done / after delegation

【注解】

yield from 是 Python 3.3 引入的,是 asyncio 协程(基于 @asyncio.coroutine + yield from)的基础。async/awaityield from 的语法糖版本,在 Python 3.5+ 中引入。

追问方向

  • yield from 如何处理子生成器中的异常?
  • 协程链(coroutine chaining)的工作原理?

Q17: async/await 与事件循环

核心答案

async def 定义原生协程,await 暂停协程等待另一个可等待对象完成。asyncio 事件循环在单线程中调度协程,通过 I/O 多路复用(epoll/kqueue)实现高并发。

深入分析

核心概念关系:

概念 说明
Coroutine async def 函数的调用结果,可以被等待
Task 对 Coroutine 的封装,交由事件循环调度
Future 表示异步操作的最终结果
Event Loop 调度器,管理 Task 队列,监听 I/O 事件

asyncio.gather vs asyncio.wait vs asyncio.create_task 对比:

函数 返回值 异常处理 取消行为
gather() 有序结果列表 默认传播第一个异常 取消 gather 取消所有子任务
wait() (done_set, pending_set) 不传播,需手动检查 取消 wait 不取消子任务
create_task() Task 对象 不自动传播 需手动 task.cancel()

代码示例

import asyncio
import time

async def fetch(name, delay):
    print(f'{name} 开始')
    await asyncio.sleep(delay)
    print(f'{name} 完成')
    return name

async def main():
    # gather: 并发执行,全部完成才返回,结果按输入顺序
    start = time.perf_counter()
    results = await asyncio.gather(
        fetch('A', 2),
        fetch('B', 1),
        fetch('C', 3),
    )
    print(f'总耗时: {time.perf_counter() - start:.1f}s')  # ~3s(不是 6s)
    print(results)  # ['A', 'B', 'C'],即使 B 先完成

    # wait: 可以部分等待
    tasks = [asyncio.create_task(fetch(f'task{i}', i)) for i in range(3)]
    done, pending = await asyncio.wait(tasks, timeout=1.5)
    print(f'完成: {len(done)}, 未完成: {len(pending)}')
    for task in pending:
        task.cancel()

asyncio.run(main())

【注解】

asyncio 是"协作式"并发,不是抢占式。协程必须主动 await 才能让出控制权。如果某个协程执行了 CPU 密集型操作而不 await,会阻塞整个事件循环。解决方案:loop.run_in_executor() 将 CPU 密集型任务放到线程池或进程池中执行。

追问方向

  • 如何在 asyncio 中调用同步阻塞函数?
  • asyncio.TaskGroup(Python 3.11+)相比 gather 有何优势?

Q18: 异步上下文管理器和异步迭代器

核心答案

异步上下文管理器实现 __aenter__/__aexit__,异步迭代器实现 __aiter__/__anext__,分别用于 async withasync for

代码示例

import asyncio

# 异步上下文管理器
class AsyncTimer:
    async def __aenter__(self):
        import time
        self.start = time.perf_counter()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        elapsed = time.perf_counter() - self.start
        print(f'耗时: {elapsed:.4f}s')
        return False  # 不抑制异常

async def main():
    async with AsyncTimer():
        await asyncio.sleep(0.1)
    # 耗时: 0.1002s

# 异步迭代器
class AsyncRange:
    def __init__(self, stop):
        self.stop = stop
        self.current = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.current >= self.stop:
            raise StopAsyncIteration
        await asyncio.sleep(0)  # 模拟异步操作
        value = self.current
        self.current += 1
        return value

async def main():
    async for i in AsyncRange(5):
        print(i)  # 0 1 2 3 4

asyncio.run(main())

【注解】

asynccontextmanager(来自 contextlib)可以用 yield 语法创建异步上下文管理器:

from contextlib import asynccontextmanager

@asynccontextmanager
async def managed_connection():
    conn = await open_connection()
    try:
        yield conn
    finally:
        await conn.close()

追问方向

  • async for 和普通 for 的底层区别?
  • 异步生成器(async def + yield)的使用场景?

六、__slots__ 与内存

Q19: __slots__ 的作用与原理

核心答案

__slots__ 用固定内存布局(类似 C 结构体)替代实例 __dict__,显著减少内存占用,适合创建大量小对象的场景。

深入分析

默认情况下,每个实例都有 __dict__(一个字典),字典本身就有约 200+ 字节开销。__slots__ 通过在类层面定义描述符数组来存储属性,避免了字典开销。

内存节省原理:

  • __dict__ 是一个哈希表,初始分配较大内存(Python 3.6+ 约 232 字节)
  • __slots__ 中每个属性是一个固定偏移的描述符,按需分配

代码示例

import sys

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

class WithSlots:
    __slots__ = ('x', 'y')

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

a = WithDict(1, 2)
b = WithSlots(1, 2)

print(sys.getsizeof(a))           # 48(对象本身)
print(sys.getsizeof(a.__dict__))  # 232(字典开销)
print(sys.getsizeof(b))           # 56(无 __dict__,包含两个 slot)

# 尝试添加未声明的属性
b.z = 3  # AttributeError: 'WithSlots' object has no attribute 'z'

# 批量创建对象时的内存差异
N = 1_000_000
with_dict = [WithDict(i, i) for i in range(N)]    # ~约 400MB
with_slots = [WithSlots(i, i) for i in range(N)]  # ~约 60MB

【注解】

sys.getsizeof() 只返回对象本身的大小,不包括其引用的对象。对于 WithDict,还需要加上 __dict__ 的大小。实际节省的内存 = 每个实例节省约 200+ 字节。

追问方向

  • __slots__ 会影响 dir() 的输出吗?
  • 如何在 __slots__ 类上实现动态属性?

Q20: __slots__ 与继承的坑

核心答案

__slots__ 在继承中有多个容易出错的地方:子类不声明 __slots__ 会重新获得 __dict__,多继承时所有父类必须都声明 __slots__

深入分析

主要陷阱:

场景 结果
子类不声明 __slots__ 子类重新有 __dict__,slots 节省失效
多个父类都有 __slots__ 需要每个都声明,否则有 __dict__
weakref 一起用 需要在 __slots__ 中包含 '__weakref__'
默认 pickle 失败,需要自定义 __getstate__/__setstate__

代码示例

class Base:
    __slots__ = ('x',)

class Child(Base):
    # 忘记声明 __slots__,重新获得 __dict__
    pass

c = Child()
c.z = 'dynamic'  # 成功!slots 节省失效
print(hasattr(c, '__dict__'))  # True

# 正确做法
class ChildCorrect(Base):
    __slots__ = ('y',)  # 只需声明新增的属性,继承 x

cc = ChildCorrect()
cc.z = 'dynamic'  # AttributeError

# 支持弱引用的 slots
class WeakRefSlots:
    __slots__ = ('x', '__weakref__')

# 支持 pickle 的 slots
class PicklableSlots:
    __slots__ = ('x', 'y')

    def __getstate__(self):
        return {slot: getattr(self, slot) for slot in self.__slots__}

    def __setstate__(self, state):
        for slot, value in state.items():
            setattr(self, slot, value)

【注解】

__slots__ 不会阻止通过 object.__setattr__(obj, name, value) 设置任意属性——这只是描述符层面的保护。真正阻止的是 __slots__ 没有对应描述符的属性查找路径,因为没有 __dict__ 回退。

追问方向

  • 如何让 slots 类支持 copy.copy()
  • __slots__dataclass 可以结合使用吗?

七、MRO 与多重继承

Q21: C3 线性化算法

核心答案

Python 使用 C3 线性化算法计算 MRO(Method Resolution Order),解决多重继承中的方法查找顺序问题,保证单调性和本地优先顺序。

深入分析

C3 算法公式:
L(C) = C + merge(L(B1), L(B2), ..., [B1, B2, ...])

merge 规则:

  • 取第一个列表的头部元素
  • 如果该元素不在任何列表的尾部,则取出放入结果,并从所有列表中删除
  • 否则跳到下一个列表尝试
  • 重复直到所有列表为空(否则抛出 TypeError)

代码示例

class A: pass
class B(A): pass
class C(A): pass
class D(B, C): pass

print(D.__mro__)
# (<class 'D'>, <class 'B'>, <class 'C'>, <class 'A'>, <class 'object'>)

# 手动计算菱形继承 MRO:
# L(D) = D + merge(L(B), L(C), [B, C])
# L(B) = [B, A, object]
# L(C) = [C, A, object]
# merge([B, A, object], [C, A, object], [B, C])
# 步骤1: 取 B,B 不在任何尾部 → 结果=[D, B]
# 步骤2: 取 A,A 在[C, A, object]的尾部 → 跳过,取 C
# 步骤3: 取 C,C 不在任何尾部 → 结果=[D, B, C]
# 步骤4: 取 A,A 不在任何尾部 → 结果=[D, B, C, A]
# 步骤5: 取 object → 结果=[D, B, C, A, object]

# 不合法的 MRO 会报错
try:
    class X(A): pass
    class Y(A): pass
    # 违反单调性的继承
    class Z(X, A): pass  # Z 的 MRO 要求 X 在 A 前,但同时 A 必须在 X 前
except TypeError as e:
    print(e)  # Cannot create a consistent method resolution order

【注解】

深度优先搜索(DFS)是 Python 2.1 之前的 MRO 算法,在菱形继承中会产生不直观的结果。C3 算法(Python 2.3+)解决了这个问题。可以用 ClassName.__mro__ClassName.mro() 查看线性化结果。

追问方向

  • 什么情况下 C3 算法会失败?
  • object 在 MRO 中为什么总在最后?

Q22: super() 的工作原理

核心答案

super() 不是简单地调用"父类"方法,而是在 MRO 链中查找当前类的下一个类。这使合作式多继承成为可能。

深入分析

super() 的两个参数形式(Python 3 中无参数形式等价):

  • super()super(__class__, self) (Python 3 魔法,通过 __class__ cell 变量)
  • super(type, obj):在 type(obj).__mro__ 中,找到 type 的下一个类
  • super(type, type2):用于类方法

合作式多继承要求:

  • 所有类都调用 super()
  • 所有 __init__ 接受 **kwargs(容纳不认识的参数)

代码示例

class Base:
    def hello(self):
        print('Base.hello')

class Left(Base):
    def hello(self):
        print('Left.hello')
        super().hello()  # 调用 MRO 中 Left 之后的类

class Right(Base):
    def hello(self):
        print('Right.hello')
        super().hello()

class Child(Left, Right):
    def hello(self):
        print('Child.hello')
        super().hello()

# MRO: Child → Left → Right → Base → object
Child().hello()
# Child.hello
# Left.hello
# Right.hello
# Base.hello

# 合作式 __init__
class A:
    def __init__(self, **kwargs):
        print(f'A.__init__, 剩余: {kwargs}')
        super().__init__(**kwargs)

class B(A):
    def __init__(self, b_param=None, **kwargs):
        print(f'B.__init__, b_param={b_param}')
        super().__init__(**kwargs)

class C(A):
    def __init__(self, c_param=None, **kwargs):
        print(f'C.__init__, c_param={c_param}')
        super().__init__(**kwargs)

class D(B, C):
    def __init__(self, **kwargs):
        print(f'D.__init__')
        super().__init__(**kwargs)

D(b_param='b', c_param='c')
# D.__init__
# B.__init__, b_param=b
# C.__init__, c_param=c
# A.__init__, 剩余: {}

【注解】

super() 的无参数形式依赖 __class__ 单元变量(编译器隐式创建),所以必须在方法体内使用。如果将 super() 赋值给变量后在外部调用,__class__ 可能已失效。不要试图"优化"成 ParentClass.method(self) 的形式——这会破坏合作式继承链。

追问方向

  • super() 无参数形式的 __class__ 是如何工作的?
  • 为什么合作式多继承需要 **kwargs

八、上下文管理器

Q23: 上下文管理器协议

核心答案

上下文管理器实现 __enter____exit__ 方法,with 语句保证在代码块结束时(无论是否异常)执行清理逻辑。

深入分析

__exit__ 方法签名:__exit__(self, exc_type, exc_val, exc_tb)

  • 正常退出:三个参数均为 None
  • 异常退出:包含异常信息
  • 返回 True:抑制异常(不重新抛出)
  • 返回 False/None:异常继续传播

contextlib 工具:

工具 用途
@contextmanager 用生成器函数创建上下文管理器
ExitStack 动态组合多个上下文管理器
suppress(*exceptions) 抑制指定异常
nullcontext 空上下文管理器(占位符)

代码示例

from contextlib import contextmanager, ExitStack
import time

# 基于类的实现
class Timer:
    def __enter__(self):
        self.start = time.perf_counter()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.elapsed = time.perf_counter() - self.start
        print(f'耗时: {self.elapsed:.4f}s')
        return False  # 不抑制异常

# 基于生成器的实现
@contextmanager
def timer(name):
    start = time.perf_counter()
    try:
        yield  # with 块在这里执行
    except Exception as e:
        print(f'{name} 发生异常: {e}')
        raise  # 重新抛出
    finally:
        elapsed = time.perf_counter() - start
        print(f'{name} 耗时: {elapsed:.4f}s')

with timer('数据库查询'):
    time.sleep(0.1)

# ExitStack:动态管理多个上下文
files = ['a.txt', 'b.txt', 'c.txt']
with ExitStack() as stack:
    handles = [stack.enter_context(open(f, 'w')) for f in files]
    # 所有文件都会被正确关闭,即使某个打开失败

【注解】

@contextmanager 的生成器中,yield 之前的代码对应 __enter__yield 之后(finally 块)对应 __exit__yield 的值成为 as 子句绑定的对象(with timer('x') as t)。

追问方向

  • ExitStack 在什么场景下特别有用?
  • 如何实现可重入的上下文管理器?

九、数据模型与魔术方法

Q24: __new__ vs __init__

核心答案

__new__ 创建并返回新对象(工厂),__init__ 初始化已创建的对象(配置)。__new__ 的返回值是 __init__self

深入分析

__new__ 的使用场景:

  • 子类化不可变类型(intstrtuplefrozenset):不可变类型的值在创建时确定,__init__ 阶段已经太晚
  • 单例模式
  • 元类中控制类的创建

执行顺序:

  1. MyClass(args) 调用 type.__call__(MyClass, args)
  2. type.__call__ 调用 MyClass.__new__(MyClass, args) 得到 obj
  3. 如果 objMyClass 的实例,调用 obj.__init__(args)
  4. 返回 obj

代码示例

# 单例模式
class Singleton:
    _instance = None

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self, value=None):
        # 注意:__init__ 每次都会被调用!
        self.value = value

s1 = Singleton(1)
s2 = Singleton(2)
print(s1 is s2)    # True
print(s1.value)    # 2(被第二次 __init__ 覆盖)

# 不可变类型子类化
class PositiveInt(int):
    def __new__(cls, value):
        if value <= 0:
            raise ValueError(f'必须为正数,收到 {value}')
        return super().__new__(cls, value)

    def __init__(self, value):
        # int 是不可变的,值已经在 __new__ 中确定
        pass  # 不需要做什么

n = PositiveInt(5)
print(n + 1)    # 6(继承了 int 的所有行为)
print(type(n))  # <class 'PositiveInt'>
PositiveInt(-1) # ValueError: 必须为正数

【注解】

单例的 __init__ 问题是常见陷阱:__new__ 返回现有实例时,__init__ 仍然会被调用。解决方案是检查是否已初始化:if not hasattr(self, '_initialized'),或使用元类实现单例。

追问方向

  • __new__ 不返回 cls 的实例时会发生什么?
  • 如何用元类实现更健壮的单例?

Q25: __getattr__ vs __getattribute__

核心答案

__getattribute__ 在每次属性访问时调用(包括成功的),__getattr__ 只在正常查找失败后调用。自定义 __getattribute__ 必须小心避免无限递归。

深入分析

方法 触发时机 递归风险 常用场景
__getattribute__ 每次属性访问 高(访问 self.xxx 再次触发) 属性访问日志、代理对象
__getattr__ 正常查找失败后 默认值、动态属性、懒加载

代码示例

class Safe:
    def __getattr__(self, name):
        # 只在找不到属性时触发,相对安全
        return f'属性 {name!r} 不存在'

s = Safe()
s.x = 1
print(s.x)        # 1(正常访问,不触发 __getattr__)
print(s.missing)  # 属性 'missing' 不存在

class Logging:
    def __init__(self):
        self._data = {}  # 注意:这里会触发 __getattribute__

    def __getattribute__(self, name):
        print(f'访问属性: {name}')
        # 必须用 object.__getattribute__ 避免无限递归
        return object.__getattribute__(self, name)

    def __setattr__(self, name, value):
        print(f'设置属性: {name} = {value}')
        # 必须用 object.__setattr__ 避免无限递归
        object.__setattr__(self, name, value)

obj = Logging()  # 设置属性: _data = {}
obj.x = 1        # 设置属性: x = 1
_ = obj.x        # 访问属性: x

【注解】

__getattribute__ 中访问 self.anything 会再次触发 __getattribute__,导致无限递归。唯一安全的访问方式是 object.__getattribute__(self, 'attr')。大多数场景下 __getattr__ 足够,避免使用 __getattribute__

追问方向

  • __setattr____delattr__ 有类似问题吗?
  • 如何用 __getattr__ 实现懒加载属性?

Q26: __repr__ vs __str__

核心答案

__repr__ 用于调试,应返回无歧义、理想情况下可重现对象的表示;__str__ 用于用户展示,可以更友好。

深入分析

调用场景:

  • repr(obj):调用 __repr__
  • str(obj):调用 __str__,若未定义则回退到 __repr__
  • print(obj):调用 str(obj)
  • 交互式解释器显示:调用 repr(obj)
  • f-string f'{obj}':调用 __format__(默认调用 str
  • f-string f'{obj!r}':调用 __repr__
  • f-string f'{obj!s}':调用 __str__

代码示例

from datetime import datetime

class Event:
    def __init__(self, name, time):
        self.name = name
        self.time = time

    def __repr__(self):
        # 可重现的表示,包含足够信息重建对象
        return f"Event(name={self.name!r}, time={self.time!r})"

    def __str__(self):
        # 用户友好的展示
        return f"{self.name}({self.time.strftime('%Y-%m-%d %H:%M')})"

e = Event('会议', datetime(2026, 3, 9, 10, 0))
print(repr(e))   # Event(name='会议', time=datetime.datetime(2026, 3, 9, 10, 0))
print(str(e))    # 会议(2026-03-09 10:00)
print(f'{e}')    # 会议(2026-03-09 10:00)
print(f'{e!r}')  # Event(name='会议', time=datetime.datetime(2026, 3, 9, 10, 0))

【注解】

__repr__ 的标准建议:"如果可能,返回一个有效的 Python 表达式,可以用 eval() 重建对象"。即 eval(repr(obj)) == obj(在适当环境下)。如果无法做到,至少用 <ClassName ...> 格式提供有用信息。

追问方向

  • 容器打印其元素时调用哪个方法?
  • __format__ 方法的用途?

十、函数与闭包高级

Q27: 闭包与 LEGB 规则

核心答案

闭包是引用了外层作用域变量的函数。自由变量存储在 __closure__(cell 对象数组)中,nonlocal 关键字允许内层函数修改外层变量。

深入分析

LEGB 查找顺序:

  • Local(函数内局部变量)
  • Enclosing(外层函数的变量,闭包)
  • Global(模块全局变量)
  • Built-in(内置名称空间)

__closure__ 属性:

  • 每个 cell 对象包装一个自由变量
  • cell.cell_contents 获取当前值

常见陷阱——循环中的闭包:所有闭包共享同一个变量引用。

代码示例

# 基本闭包
def make_counter():
    count = 0
    def counter():
        nonlocal count  # 声明修改外层变量
        count += 1
        return count
    return counter

c1 = make_counter()
c2 = make_counter()  # 独立的闭包
print(c1(), c1(), c1())  # 1 2 3
print(c2())              # 1(独立)
print(c1.__closure__[0].cell_contents)  # 3

# 循环闭包陷阱
funcs_wrong = [lambda: i for i in range(5)]
print([f() for f in funcs_wrong])  # [4, 4, 4, 4, 4](全是最终值)

# 解决方案 1:默认参数固定值
funcs_fixed = [lambda i=i: i for i in range(5)]
print([f() for f in funcs_fixed])  # [0, 1, 2, 3, 4]

# 解决方案 2:工厂函数
def make_func(i):
    return lambda: i

funcs_factory = [make_func(i) for i in range(5)]
print([f() for f in funcs_factory])  # [0, 1, 2, 3, 4]

【注解】

nonlocal 只能引用最近一层外层函数的变量,不能跳级引用,也不能引用全局变量(全局变量用 global)。闭包中的 cell 对象是共享的:如果外层函数修改了自由变量,闭包看到的是修改后的值。

追问方向

  • 如何实现一个记忆化(memoize)装饰器?
  • 闭包和对象有什么共同之处?

Q28: 默认参数陷阱

核心答案

函数默认参数在函数定义时求值一次,所有调用共享同一个默认值对象。可变对象(列表、字典、集合)作为默认参数会导致意外的状态共享。

代码示例

# 错误写法:默认参数只求值一次
def append_to(item, lst=[]):
    lst.append(item)
    return lst

print(append_to(1))  # [1]
print(append_to(2))  # [1, 2]   不是 [2]!
print(append_to(3))  # [1, 2, 3]

# 检查:默认值对象是同一个
print(append_to.__defaults__)  # ([1, 2, 3],)

# 正确写法:使用 None 作为哨兵
def append_to_correct(item, lst=None):
    if lst is None:
        lst = []
    lst.append(item)
    return lst

print(append_to_correct(1))  # [1]
print(append_to_correct(2))  # [2]  正确!

# 利用这个特性实现简单缓存(通常不推荐,但值得了解)
def memoize_demo(n, _cache={}):
    if n not in _cache:
        _cache[n] = n * n  # 模拟计算
    return _cache[n]

【注解】

这是 Python 最著名的"陷阱"之一。它源于 Python 的一致性:默认参数是函数对象的属性(存储在 func.__defaults__ 元组中),在 def 语句执行时求值。这个行为本身没有 bug,理解原因后可以利用它实现默认可变缓存。

追问方向

  • datetime.now() 作为默认参数会有什么问题?
  • 如何检查一个函数的默认参数值?

Q29: 装饰器高级——带参数的装饰器

核心答案

带参数的装饰器是"返回装饰器的函数",需要三层嵌套。functools.wraps 保留原函数的元数据。

代码示例

import functools
import time

def retry(max_times=3, delay=1.0, exceptions=(Exception,)):
    """带参数的装饰器:自动重试"""
    def decorator(func):
        @functools.wraps(func)  # 保留 __name__、__doc__、__annotations__ 等
        def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(max_times):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_error = e
                    print(f'{func.__name__} 第 {attempt+1}/{max_times} 次失败: {e}')
                    if attempt < max_times - 1:
                        time.sleep(delay)
            raise last_error
        return wrapper
    return decorator

@retry(max_times=3, delay=0.1, exceptions=(ConnectionError, TimeoutError))
def fetch_data(url):
    """获取数据"""
    import random
    if random.random() < 0.7:
        raise ConnectionError(f'连接 {url} 失败')
    return f'来自 {url} 的数据'

# 验证 wraps 的效果
print(fetch_data.__name__)  # fetch_data(不是 wrapper)
print(fetch_data.__doc__)   # 获取数据

# 类装饰器(更灵活)
class Retry:
    def __init__(self, max_times=3):
        self.max_times = max_times

    def __call__(self, func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for i in range(self.max_times):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if i == self.max_times - 1:
                        raise
        return wrapper

@Retry(max_times=5)
def risky():
    pass

【注解】

不使用 functools.wraps 会导致装饰后函数的 __name____doc____annotations__ 等属性丢失,这会影响文档生成、调试和内省。functools.wraps 本身也是一个装饰器,它复制这些属性并设置 __wrapped__ 指向原始函数。

追问方向

  • 如何让装饰器既可以带参数又可以不带参数使用?
  • __wrapped__ 属性有什么用途?

十一、并发与并行

Q30: threading vs multiprocessing vs asyncio 选型

核心答案

三种并发方案各有适用场景,选型取决于任务类型(I/O 还是 CPU 密集)和并发规模。

深入分析

维度 threading multiprocessing asyncio
GIL 影响 受限(CPU 密集无效) 无(独立进程) 无(单线程)
内存模型 共享内存 独立内存空间 共享内存
进程间通信 Queue / Lock / 共享对象 Pipe / Queue / Manager await / asyncio.Queue
适用场景 I/O 密集型 CPU 密集型 高并发 I/O
创建开销 中(MB 级) 高(进程启动) 极低(协程 < 1KB)
并发规模 数十到数百 受 CPU 核数限制 数万级别
调试难度 较高

选型指南:

  • 爬虫/API 调用:asyncio 或 threading
  • 图像处理/数学计算:multiprocessing 或 numpy(C 扩展)
  • 混合场景:asyncio + run_in_executor(线程/进程池)

追问方向

  • 什么时候用 ThreadPoolExecutor 而不是直接用 Thread
  • 进程间如何共享大量数据?

Q31: 线程安全工具

核心答案

Python threading 模块提供多种同步原语,queue.Queue 是线程间通信的推荐方式。

深入分析

同步原语对比:

工具 用途 特点
Lock 互斥访问 不可重入
RLock 可重入互斥锁 同一线程可多次获取
Semaphore 控制并发数量 计数器
Event 线程间信号通知 set/wait/clear
Condition 条件等待 notify/wait
Barrier 等待所有线程到达 同步点

代码示例

import threading
import queue

# 生产者-消费者模式(推荐用 Queue)
def producer(q, n):
    for i in range(n):
        q.put(i)
        print(f'生产: {i}')
    q.put(None)  # 结束信号

def consumer(q):
    while True:
        item = q.get()
        if item is None:
            break
        print(f'消费: {item}')
        q.task_done()

q = queue.Queue(maxsize=5)
p = threading.Thread(target=producer, args=(q, 10))
c = threading.Thread(target=consumer, args=(q,))
p.start(); c.start()
p.join(); c.join()

# Lock 保护共享状态
counter = 0
lock = threading.Lock()

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

threads = [threading.Thread(target=increment) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()
print(counter)  # 400000(有锁保护,结果正确)

【注解】

CPython 中,简单的赋值操作(x = value)在 GIL 保护下通常是"原子的"(不会被中断到一半),但这是实现细节,不是语言规范保证。对于需要读-改-写的操作(counter += 1),即使有 GIL,也不是原子的(三条字节码指令),必须用 Lock。

追问方向

  • RLockLock 在什么场景下必须用 RLock
  • 如何检测和避免死锁?

Q32: concurrent.futures 模块

核心答案

concurrent.futures 提供统一的高层接口管理线程池和进程池,Future 对象表示异步执行的结果。

代码示例

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed
import time

def fetch_url(url):
    # 模拟网络请求
    time.sleep(0.1)
    return f'数据来自 {url}'

def cpu_task(n):
    return sum(i * i for i in range(n))

urls = [f'https://example.com/{i}' for i in range(20)]

# I/O 密集型:线程池
with ThreadPoolExecutor(max_workers=10) as executor:
    futures = {executor.submit(fetch_url, url): url for url in urls}

    for future in as_completed(futures):
        url = futures[future]
        try:
            result = future.result(timeout=5)
            print(f'{url}: {result}')
        except Exception as e:
            print(f'{url} 失败: {e}')

# CPU 密集型:进程池
data = [1_000_000] * 8
with ProcessPoolExecutor() as executor:
    # map 保持顺序,as_completed 按完成顺序
    results = list(executor.map(cpu_task, data))
    print(f'结果: {results[:3]}...')

# 结合 asyncio(在协程中使用线程池执行阻塞调用)
import asyncio

async def main():
    loop = asyncio.get_event_loop()
    with ThreadPoolExecutor(max_workers=5) as pool:
        result = await loop.run_in_executor(pool, fetch_url, 'https://example.com')
        print(result)

asyncio.run(main())

【注解】

executor.map() 懒惰地提交任务并阻塞等待结果(按输入顺序),as_completed() 按完成顺序返回 Future。大规模任务时,as_completed 可以更快处理先完成的结果。进程池在 Windows 上必须有 if __name__ == '__main__': 保护(避免子进程重复 fork)。

追问方向

  • Future.cancel() 什么时候有效?
  • 如何限制进程池任务的超时时间?

十二、性能优化

Q33: 性能分析工具

核心答案

性能优化的第一步是定位瓶颈,不要过早优化。cProfile 找函数级热点,timeit 精确测量小代码段。

代码示例

import cProfile
import pstats
import io

# cProfile 分析
def slow_function():
    result = []
    for i in range(10000):
        result.append(str(i))
    return '-'.join(result)

# 方式 1:运行并保存
cProfile.run('slow_function()', 'profile_output')
stats = pstats.Stats('profile_output')
stats.sort_stats('cumulative').print_stats(10)

# 方式 2:上下文管理器
with cProfile.Profile() as pr:
    slow_function()
stream = io.StringIO()
stats = pstats.Stats(pr, stream=stream)
stats.sort_stats('cumulative').print_stats(10)
print(stream.getvalue())

# timeit:精确测量
import timeit

# 比较字符串拼接方式
t1 = timeit.timeit(
    '"-".join(str(n) for n in range(100))',
    number=10000
)
t2 = timeit.timeit(
    '"-".join([str(n) for n in range(100)])',
    number=10000
)
print(f'生成器: {t1:.3f}s, 列表推导: {t2:.3f}s')
# 列表推导通常比生成器快(避免了 __next__ 调用开销)

# line_profiler(需安装):逐行分析
# @profile  # kernprof -l -v script.py
# def my_function():
#     pass

【注解】

cProfile 的开销相对较大(每次函数调用都有统计),生产环境可以用 py-spy(基于采样,无侵入)。timeit 默认会禁用垃圾收集器(gc.disable())以减少测量噪声,这可能不代表真实运行环境。

追问方向

  • 如何分析内存使用而不是时间?
  • yappicProfile 的区别?

Q34: 常见性能优化技巧

核心答案

Python 层面的优化技巧,无需引入 C 扩展,通过利用字节码特性和数据结构选择提升性能。

深入分析

主要优化技巧:

技巧 原因 提升幅度
局部变量缓存属性 LOAD_FAST < LOAD_ATTR
列表推导代替 for+append 字节码更优化 小-中
''.join() 代替字符串 + 避免 O(n²) 内存复制 大(长字符串)
字典/集合查找代替列表搜索 O(1) vs O(n) 大(大数据量)
__slots__ 减少实例内存 中(大量对象)
避免全局变量 LOAD_GLOBALLOAD_FAST

代码示例

import timeit

# 技巧1:缓存方法引用
class DataProcessor:
    def __init__(self):
        self.data = []

    def slow_append(self, n):
        for i in range(n):
            self.data.append(i)  # 每次循环查找 self.data 和 .append

    def fast_append(self, n):
        data_append = self.data.append  # 缓存方法引用
        for i in range(n):
            data_append(i)

# 技巧2:集合代替列表做成员检查
big_list = list(range(100000))
big_set = set(range(100000))

t1 = timeit.timeit('99999 in big_list', globals=globals(), number=1000)
t2 = timeit.timeit('99999 in big_set', globals=globals(), number=1000)
print(f'列表查找: {t1:.4f}s, 集合查找: {t2:.6f}s')
# 集合查找快约 1000 倍

# 技巧3:字符串拼接
def slow_join(words):
    result = ''
    for word in words:
        result += word + ', '  # 每次创建新字符串对象
    return result

def fast_join(words):
    return ', '.join(words)  # 一次分配,线性时间

words = ['word'] * 10000
t1 = timeit.timeit(lambda: slow_join(words), number=100)
t2 = timeit.timeit(lambda: fast_join(words), number=100)
print(f'字符串+: {t1:.3f}s, join: {t2:.3f}s')

追问方向

  • numpy 的向量化操作为什么比 Python 循环快?
  • Cython 和 ctypes 的适用场景?

Q35: dis 模块查看字节码

核心答案

dis 模块反汇编 Python 字节码,帮助理解代码执行细节、优化性能、深入理解语言机制。

代码示例

import dis

def f(x):
    return x * 2 + 1

dis.dis(f)
# 输出(Python 3.11+):
#   2           RESUME          0
#   3           LOAD_FAST       0 (x)    ← 局部变量(快)
#               LOAD_CONST      1 (2)
#               BINARY_OP       5 (*)
#               LOAD_CONST      2 (1)
#               BINARY_OP       0 (+)
#               RETURN_VALUE

# 对比:全局变量访问
y = 10
def g(x):
    return x * y  # y 是全局变量

dis.dis(g)
# LOAD_FAST  0 (x)
# LOAD_GLOBAL 0 (y)   ← 全局变量查找(慢)
# BINARY_OP  5 (*)
# RETURN_VALUE

# 列表推导 vs for + append 的字节码差异
def list_comp():
    return [x * 2 for x in range(10)]

def for_loop():
    result = []
    for x in range(10):
        result.append(x * 2)
    return result

print('--- 列表推导 ---')
dis.dis(list_comp)
print('--- for 循环 ---')
dis.dis(for_loop)
# 列表推导在内部使用 LIST_APPEND 指令,比显式 append 更高效

【注解】

dis 是学习 Python 内部机制的绝佳工具。Python 3.11+ 引入了 specializing adaptive interpreter(专化自适应解释器),字节码会在运行时根据实际类型进行专化优化,dis 输出的是未专化版本。

追问方向

  • 如何查看生成器函数的字节码?
  • Python 3.11 的"零开销异常处理"如何体现在字节码中?

十三、类型系统

Q36: Python 类型标注体系

核心答案

Python 的类型标注是可选的静态标注,运行时不强制检查,由 mypy/pyright 等工具进行静态分析。

深入分析

类型系统演进:

版本 新增特性
3.5 PEP 484:typing 模块,基础类型标注
3.6 变量注解(x: int = 0
3.9 内置集合泛型(list[int] 代替 List[int]
3.10 X | Y 代替 Union[X, Y]match 语句
3.11 SelfNeverLiteralString 类型
3.12 TypeVar 语法糖(type Alias = ...

重要类型工具:

工具 说明
TypeVar 泛型类型变量
Generic[T] 泛型类
Protocol 结构子类型(鸭子类型的正式化)
overload 函数重载标注
Literal 字面量类型
TypedDict 有类型标注的字典

代码示例

from typing import TypeVar, Generic, Protocol, overload

T = TypeVar('T')

# 泛型类
class Stack(Generic[T]):
    def __init__(self) -> None:
        self._items: list[T] = []

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

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

    def __len__(self) -> int:
        return len(self._items)

stack: Stack[int] = Stack()
stack.push(1)
stack.push(2)
print(stack.pop())  # 2

# Protocol:结构子类型(不需要显式继承)
class Drawable(Protocol):
    def draw(self) -> None: ...

class Circle:
    def draw(self) -> None:
        print('画圆')

class Square:
    def draw(self) -> None:
        print('画方形')

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

render(Circle())  # 满足 Protocol,无需继承 Drawable
render(Square())

# Python 3.10+ 的新语法
def process(value: int | str | None) -> str:
    match value:
        case None:
            return '空值'
        case int(n):
            return f'整数: {n}'
        case str(s):
            return f'字符串: {s}'

【注解】

Protocol 实现了"结构子类型"(Structural Subtyping),是 Python 鸭子类型的类型系统版本。与 ABC(名义子类型)不同,只要具备协议要求的方法就满足,无需显式继承。这让第三方库中的类也可以满足自定义 Protocol。

追问方向

  • @runtime_checkable 装饰的 Protocol 可以用于 isinstance 检查吗?
  • TypeVarbound 参数有什么用?

Q37: isinstance vs type()

核心答案

isinstance() 支持继承链和 ABC 虚拟子类,type() 严格比较类型身份。生产代码中几乎总是用 isinstance()

深入分析

比较 isinstance type() is
继承链
ABC 虚拟子类 是(register())
多类型检查 isinstance(x, (A, B)) type(x) in (A, B)
用途 鸭子类型友好 需要精确类型时

代码示例

from abc import ABC, abstractmethod

class Animal(ABC):
    @abstractmethod
    def speak(self): pass

class Dog(Animal):
    def speak(self): return '汪'

class Cat:  # 没有继承 Animal
    def speak(self): return '喵'

# 注册虚拟子类
Animal.register(Cat)

d = Dog()
c = Cat()

print(isinstance(d, Animal))   # True(继承)
print(isinstance(c, Animal))   # True(虚拟子类注册)
print(type(d) is Animal)       # False
print(type(c) is Animal)       # False

# 精确类型检查(罕见场景)
class SpecialList(list): pass

lst = SpecialList([1, 2, 3])
print(isinstance(lst, list))   # True
print(type(lst) is list)       # False(SpecialList 不完全是 list)
# 如果需要区分 list 和其子类,用 type() is

【注解】

isinstance() 是多态友好的写法,符合里氏替换原则。type() is 只在极少数情况下合理:比如需要区分基类和子类行为时。避免用 type(x).__name__ == 'list' 这种字符串比较,这是反模式。

追问方向

  • ABC 的 __subclasshook__ 如何自定义 isinstance 行为?
  • isinstance 和 Protocol 如何配合(@runtime_checkable)?

十四、模块与导入

Q38: Python 导入机制

核心答案

import 语句触发一个多步查找过程:检查 sys.modules 缓存 → finder 定位模块 → loader 加载执行 → 缓存结果。

深入分析

导入流程:

  1. 检查 sys.modules:已导入则直接返回缓存
  2. 遍历 sys.meta_path:找到能处理该模块的 finder
  3. finder 返回 loader(或 ModuleSpec)
  4. loader 加载并执行模块代码
  5. 将模块对象存入 sys.modules

sys.path 查找顺序:

  • 当前目录
  • PYTHONPATH 环境变量
  • 标准库目录
  • site-packages(第三方库)

代码示例

import sys
import importlib

# 查看已加载的模块
print('json' in sys.modules)  # True(如果已导入)

# 动态导入
module_name = 'json'
json = importlib.import_module(module_name)
print(json.dumps({'key': 'value'}))

# 解决循环导入
# 方式1:延迟导入(在函数内导入)
def get_something():
    import heavy_module  # 只在需要时导入
    return heavy_module.compute()

# 方式2:导入模块而非对象
# 而不是 from module_a import ClassA(可能触发循环)
# 改为 import module_a,然后用 module_a.ClassA

# 重新导入(测试/开发时用)
importlib.reload(json)  # 重新执行模块代码

# 检查导入路径
import os
print(os.__file__)  # 模块的文件路径

【注解】

sys.modules 缓存是导入系统的核心。模块只执行一次(除非 reload)。这意味着模块级变量是单例的——整个进程中所有导入者共享同一个模块对象。循环导入问题的根源是:模块 A 导入 B,B 再导入 A 时,A 还没完全初始化(在 sys.modules 中是不完整的对象)。

追问方向

  • __import__importlib.import_module 的区别?
  • 如何实现自定义 finder/loader?

Q39: __all__ 的作用

核心答案

__all__ 是模块级列表,控制 from module import * 时导出的名称集合。

深入分析

__all__ 的行为:

  • 定义 __all__from module import * 只导入 __all__ 中的名称
  • 不定义 __all__from module import * 导入所有不以 _ 开头的名称
  • 不影响 import modulefrom module import specific_name

代码示例

# mymodule.py
__all__ = ['PublicClass', 'public_function']
# _private 和 _INTERNAL 不会被 * 导出

class PublicClass:
    pass

def public_function():
    pass

def _private_helper():  # 约定私有
    pass

_INTERNAL = 'internal'

ALSO_PUBLIC = 'but not in __all__'  # 存在但不在 __all__ 中

# 使用方:
# from mymodule import *
# → 导入 PublicClass, public_function
# → 不导入 _private_helper, _INTERNAL, ALSO_PUBLIC

# from mymodule import ALSO_PUBLIC  → 仍然可以显式导入
# import mymodule; mymodule.ALSO_PUBLIC  → 也可以直接访问

【注解】

__all__ 也是文档工具:它明确声明了模块的公共 API。好的库设计应该定义 __all__。注意 __all__ 中的名称必须存在,否则 import * 时会报 AttributeError

追问方向

  • __all___ 前缀的约定有什么区别?
  • 包的 __init__.py 中如何用 __all__ 控制导出?

十五、其他高频题

Q40: 可哈希性

核心答案

对象要作为字典键或集合元素,必须是可哈希的(实现 __hash__)。定义 __eq____hash__ 自动被设为 None,需要同时定义。

深入分析

哈希一致性规则:

  • a == b,则 hash(a) == hash(b)(必须保证)
  • hash(a) == hash(b),不一定 a == b(哈希碰撞)

Python 内置类型的哈希性:

类型 可哈希 原因
int, float, str, tuple, frozenset 不可变
list, dict, set 可变
自定义类(默认) 基于 id()

代码示例

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

    def __eq__(self, other):
        if not isinstance(other, Point):
            return NotImplemented
        return self.x == other.x and self.y == other.y

    def __hash__(self):
        # 基于 __eq__ 用到的字段
        return hash((self.x, self.y))

p1 = Point(1, 2)
p2 = Point(1, 2)
p3 = Point(3, 4)

print(p1 == p2)          # True
print(hash(p1) == hash(p2))  # True(一致性)

# 可以用作字典键和集合元素
d = {p1: 'origin'}
print(d[p2])             # 'origin'(p1 == p2,所以查到了)
s = {p1, p2, p3}
print(len(s))            # 2(p1 和 p2 相等,集合去重)

【注解】

对于可变对象,通常不应该定义 __hash__(对象内容变化后哈希值不应变化,否则在字典中会"丢失")。如果类需要是可变的又需要哈希,可以基于对象 id:__hash__ = object.__hash__(但此时 __eq__ 也应该基于 id)。

追问方向

  • frozenset 为什么可哈希而 set 不可以?
  • 如何为包含列表的类实现哈希?

Q41: namedtuple vs dataclass vs TypedDict

核心答案

三种数据容器各有适用场景,选择取决于可变性需求、继承需求和使用场景。

深入分析

特性 namedtuple dataclass TypedDict
可变性 不可变 可变(frozen=True 则不可变) 可变(字典)
继承 有限(元组限制) 完整 完整
默认值 有限(仅末尾字段) 完整(field()
内存 小(元组) 中(__dict__ 或 slots) 字典开销
运行时类型检查 否(仅静态)
解包 是(元组兼容)
JSON 序列化 需要 ._asdict() 需要自定义或库 直接(字典)
适用场景 轻量不可变记录 有逻辑的数据对象 API 响应/配置

代码示例

from collections import namedtuple
from dataclasses import dataclass, field
from typing import TypedDict

# namedtuple:轻量、不可变、可解包
Point = namedtuple('Point', ['x', 'y'])
p = Point(1, 2)
x, y = p  # 解包
print(p._asdict())  # {'x': 1, 'y': 2}

# dataclass:功能完整
@dataclass
class Person:
    name: str
    age: int
    hobbies: list[str] = field(default_factory=list)

    def greet(self):
        return f'你好,我是 {self.name}'

alice = Person('Alice', 30)
alice.hobbies.append('编程')

# frozen dataclass:不可变 + 可哈希
@dataclass(frozen=True)
class ImmutablePoint:
    x: float
    y: float

ip = ImmutablePoint(1.0, 2.0)
{ip: 'point'}  # 可作为字典键

# TypedDict:类型化的字典(适合 API 响应)
class UserResponse(TypedDict):
    id: int
    name: str
    email: str

def get_user() -> UserResponse:
    return {'id': 1, 'name': 'Alice', 'email': '[email protected]'}

user = get_user()
print(user['name'])  # 普通字典访问

【注解】

dataclass(slots=True)(Python 3.10+)可以让 dataclass 使用 __slots__,兼顾便利性和内存效率。TypedDict 的类型标注只在静态分析时有效,运行时它就是普通字典,没有验证。

追问方向

  • dataclass__post_init__ 有什么用?
  • NamedTuple(typing 模块)和 collections.namedtuple 有什么区别?

Q42: 字符串驻留(String Interning)

核心答案

Python 自动驻留(intern)小整数(-5 到 256)和符合标识符规则的字符串,使相同值的对象共享内存。is 比较内存地址,不能用于比较字符串值。

深入分析

自动驻留的对象:

  • 小整数:-5 到 256(作为常量预先创建)
  • 编译时确定的字符串字面量(如果符合标识符规则)
  • 模块、类、函数的名称属性

代码示例

import sys

# 小整数驻留
a = 256
b = 256
print(a is b)  # True(驻留)

a = 257
b = 257
print(a is b)  # False(不驻留,可能为 True,是实现细节)

# 字符串驻留
s1 = 'hello'
s2 = 'hello'
print(s1 is s2)  # True(编译时优化)

s1 = 'hello world'  # 含空格,可能不驻留
s2 = 'hello world'
print(s1 is s2)  # 可能 True 或 False(实现细节!)

# 永远用 == 比较字符串值
print(s1 == s2)  # True(正确做法)

# 手动驻留:用于大量重复字符串的内存优化
s = sys.intern('frequently_used_string')
t = sys.intern('frequently_used_string')
print(s is t)  # True(手动驻留后保证)

# 实际应用:解析大量重复键的数据
def parse_records(records):
    # 驻留键名减少内存占用
    return [{sys.intern(k): v for k, v in r.items()} for r in records]

【注解】

is 比较字符串是危险的:CPython 的驻留行为是实现细节,不同版本、不同执行方式(交互式 vs 文件)可能有不同结果。面试中如果问"两个字符串 is 比较结果",正确答案是"不确定,取决于实现",并强调应该用 ==

追问方向

  • 整数 257 在什么情况下 is 会返回 True?
  • sys.intern 适合在什么场景使用?

Q43: 异常链

核心答案

Python 支持隐式异常链(__context__)和显式异常链(from),可以保留原始异常信息,也可以用 from None 隐藏。

代码示例

# 显式异常链:from 关键字
try:
    open('not_exist.txt')
except FileNotFoundError as e:
    raise RuntimeError('配置文件缺失') from e
# Traceback 会显示两个异常,并说明"The above exception was the direct cause"

# 隐藏原始异常
try:
    open('not_exist.txt')
except FileNotFoundError:
    raise RuntimeError('配置文件缺失') from None
# 只显示 RuntimeError

# 隐式异常链(不使用 from)
def risky():
    try:
        1 / 0
    except ZeroDivisionError:
        raise ValueError('计算错误')
        # Python 自动设置 __context__ 为 ZeroDivisionError
        # Traceback 会显示"During handling of the above exception, another exception occurred"

# 访问异常链
try:
    risky()
except ValueError as e:
    print(f'原始异常: {e.__context__}')
    print(f'原因: {e.__cause__}')  # None(没有用 from)

【注解】

__cause__(显式 from)和 __context__(隐式)的区别:

  • raise X from YX.__cause__ = YX.__suppress_context__ = True
  • raise X(在 except 中):X.__context__ = 当前异常

from None 实际上是 raise X from None,将 __cause__ 设为 None 并设 __suppress_context__ = True

追问方向

  • 如何在捕获异常后清除异常上下文避免内存泄漏?
  • sys.exc_info() 和异常链的关系?

Q44: Python 3.11+ ExceptionGroup

核心答案

ExceptionGroup 允许一个异常携带多个子异常,配合 except* 语法分别处理不同类型的子异常,主要用于并发任务错误聚合。

代码示例

# 基本 ExceptionGroup
try:
    raise ExceptionGroup('多个网络错误', [
        ConnectionError('A 连接失败'),
        TimeoutError('B 超时'),
        ConnectionError('C 连接失败'),
    ])
except* ConnectionError as eg:
    # eg 是一个 ExceptionGroup,包含所有匹配的 ConnectionError
    print(f'处理连接错误: {len(eg.exceptions)} 个')
    for exc in eg.exceptions:
        print(f'  - {exc}')
except* TimeoutError as eg:
    print(f'处理超时: {eg.exceptions}')

# asyncio.TaskGroup(Python 3.11+):自动收集任务异常
import asyncio

async def failing_task(name, should_fail=False):
    await asyncio.sleep(0.1)
    if should_fail:
        raise ValueError(f'{name} 失败')
    return name

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            t1 = tg.create_task(failing_task('A'))
            t2 = tg.create_task(failing_task('B', should_fail=True))
            t3 = tg.create_task(failing_task('C', should_fail=True))
        # 所有任务完成后,若有失败则抛出 ExceptionGroup
    except* ValueError as eg:
        print(f'任务失败: {[str(e) for e in eg.exceptions]}')

asyncio.run(main())

【注解】

except* 不是 except 的替代品,两者不能混用(同一个 try 块中)。except* 会捕获 ExceptionGroup 中所有匹配类型的子异常,剩余的重新抛出(如果还有的话)。这对于 asyncio.TaskGroup 等并发场景非常重要——多个任务可能同时失败,需要批量处理错误。

追问方向

  • ExceptionGroup 可以嵌套吗?
  • asyncio.gatherasyncio.TaskGroup 的错误处理有何不同?

Q45: functools 实用工具

核心答案

functools 模块提供高阶函数工具,lru_cache 自动缓存、partial 部分应用、reduce 归约等是最常用的。

深入分析

主要工具:

工具 用途
lru_cache(maxsize) 基于 LRU 策略的自动缓存
cache (3.9+) 无限缓存,等价于 lru_cache(maxsize=None)
partial 固定部分参数,创建新函数
reduce 归约操作
wraps 装饰器中保留函数元数据
total_ordering 自动推导比较方法
singledispatch 函数重载(基于第一个参数类型)

代码示例

import functools

# lru_cache:自动缓存(底层是 LRU 字典 + 双向链表)
@functools.lru_cache(maxsize=128)
def fib(n):
    if n < 2:
        return n
    return fib(n-1) + fib(n-2)

print(fib(50))               # 瞬间返回
print(fib.cache_info())      # CacheInfo(hits=48, misses=51, maxsize=128, currsize=51)
fib.cache_clear()            # 清空缓存

# partial:部分应用
def power(base, exp):
    return base ** exp

square = functools.partial(power, exp=2)
cube = functools.partial(power, exp=3)
print(square(5))  # 25
print(cube(3))    # 27

# total_ordering:只需实现 __eq__ 和一个比较方法
@functools.total_ordering
class Temperature:
    def __init__(self, celsius):
        self.celsius = celsius

    def __eq__(self, other):
        return self.celsius == other.celsius

    def __lt__(self, other):
        return self.celsius < other.celsius

    # __le__, __gt__, __ge__ 自动推导

t1 = Temperature(20)
t2 = Temperature(30)
print(t1 < t2)   # True
print(t1 >= t2)  # False(自动推导)

# singledispatch:基于类型的函数分发
@functools.singledispatch
def process(data):
    raise TypeError(f'不支持的类型: {type(data)}')

@process.register(int)
def _(data):
    return f'处理整数: {data * 2}'

@process.register(str)
def _(data):
    return f'处理字符串: {data.upper()}'

@process.register(list)
def _(data):
    return f'处理列表: {len(data)} 个元素'

print(process(42))         # 处理整数: 84
print(process('hello'))   # 处理字符串: HELLO
print(process([1, 2, 3])) # 处理列表: 3 个元素

# reduce
from functools import reduce
total = reduce(lambda acc, x: acc + x, [1, 2, 3, 4, 5], 0)  # 15
product = reduce(lambda acc, x: acc * x, [1, 2, 3, 4, 5], 1)  # 120

【注解】

lru_cache 要求被缓存函数的参数必须是可哈希的(因为缓存键是参数的元组)。传入列表、字典等可变对象会报 TypeError。解决方案:将可变参数转为不可变类型(如 tuple(lst))或使用 functools.cache(Python 3.9+,仍需可哈希参数)。

追问方向

  • lru_cache 用于类方法时有什么问题?
  • singledispatch 和面向对象多态有什么区别?

最佳实践

答题遵循"核心答案 → 原理 → 代码演示 → 追问"四层结构:面试官首先需要判断你是否知道答案(核心答案 15 秒内),然后考察你是否理解原理,最后看能否在代码中实际运用。照搬模板背诵而不能延伸会在追问环节露馅。

GIL / 协程 / 元类 三大核心必须能画图解释:这三个主题在 Python 高级面试中出现频率最高,仅能口述不够,能用图示(时序图、调用栈图)解释更能体现深度理解。

用真实项目案例回答"你用过 X 解决什么问题":面试官问"asyncio 有什么用"时,背教科书定义不如描述一个真实场景:例如"我们的爬虫系统原来用多线程,切换到 asyncio + aiohttp 后并发量从 50 提升到 500,CPU 使用率下降了 40%"。

弱引用、内存模型等细节题先确认面试层级:部分题目(如 weakref、CPython 内存池)只在资深 / Python 核心开发岗位考察,普通后端岗位答出基本原理即可,避免过度准备低优先级内容而忽略高频基础。


常见陷阱

陷阱:混淆并发(Concurrency)和并行(Parallelism)

现象: 描述 GIL 时说"Python 不支持并发",被面试官追问导致无法自圆其说。

原因: GIL 只限制同时执行字节码的线程数为 1,但 I/O 密集型任务(网络请求、文件读写)等待时会释放 GIL,允许其他线程运行。asyncio 的并发也不受 GIL 影响(单线程事件循环)。

解决: 精确表述:GIL 阻止了 CPU 密集型 任务的多线程并行;I/O 密集型任务可以通过多线程或 asyncio 并发。真正需要 CPU 并行要用 multiprocessing


陷阱:把装饰器说成"语法糖"就止步

现象: 被追问"如何保留被装饰函数的 __name____doc__"时答不上来。

原因: 装饰器本质是函数替换,若不使用 functools.wraps,包装函数会覆盖原函数的元数据。

解决: 永远在装饰器内层函数上加 @functools.wraps(func)

import functools
def my_decorator(func):
    @functools.wraps(func)      # 保留 __name__、__doc__、__wrapped__ 等
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

陷阱:用可变默认参数

现象: 函数定义了 def foo(items=[]):,多次调用后 items 意外积累了之前调用的数据。

原因: Python 函数默认参数在定义时求值一次,可变对象(list、dict)在所有调用间共享同一个实例。

解决: 可变类型的默认参数用 None 代替,函数体内判断并初始化。

# 错误
def append_to(element, to=[]):
    to.append(element)
    return to

# 正确
def append_to(element, to=None):
    if to is None:
        to = []
    to.append(element)
    return to

参见

Python爬虫工程师面试题
装饰器与函数高级
数据类型
内置函数完全参考

阅读更多

Web 安全基础

1. HTML 转义(服务端渲染必须): 2. CSP(Content Security Policy): 3. HttpOnly Cookie:防止 JS 读取会话 Cookie: 4. 前端框架防护: 攻击者在第三方网站构造一个表单,诱导已登录用户提交,浏览器会自动携带目标站的 Cookie。 触发条件: 1. 用户已登录目标网站(Cookie 有效) 2. 目标 API 仅凭 Cookie 识别用户身份 3. 请求来源未验证 1. CSRF Token(推荐): 2. SameSite Cookie: 3. 验证 Origin/Referer 头:

By yellowdog

HTTP 协议深度指南

HTTP(HyperText Transfer Protocol)是 Web 的基础传输协议,基于 TCP/IP,采用请求/响应模型。 相关文档:Web安全基础(/web-an-quan-ji-chu/) FastAPI完全指南(/fastapi-wan-quan-zhi-nan/) Nginx完全指南(/nginx-wan-quan-zhi-nan/) 幂等性:多次执行相同请求,服务器状态结果相同。PUT /users/1 多次执行结果一致;POST /users 每次创建新资源,非幂等。 浏览器直接从本地缓存读取,不向服务器发送请求。 缓存命中时,状

By yellowdog

系统设计基础

SLA 对照表: 选择建议:无状态服务(Web 层、API 层)优先水平扩展;数据库初期垂直扩展,达到瓶颈后考虑分库分表或读写分离。 缓存穿透(查询不存在的 key,每次都打到 DB): 缓存击穿(热点 key 过期,瞬间大量请求打到 DB): 缓存雪崩(大量 key 同时过期,或缓存服务宕机): 令牌桶 Python 实现: Redis 实现分布式限流(滑动窗口): URL 命名规则: Cursor 分页响应格式: 雪花算法结构(64 bit): 定义:分布式系统不能同时满足以下三个特性: 在分布式环境中 P 是必须保证的,所以实际是 CP vs AP

By yellowdog

算法思路与模板

二分查找要求序列有序,每次将搜索范围缩减一半,时间复杂度 O(log n)。 两个指针从两端向中间收缩,常用于有序数组。 滑动窗口维护一个满足条件的区间 left, right,right 不断向右扩张,条件不满足时收缩 left。 滑动窗口通用框架: 1. 确定"子问题":原问题可以分解为哪些规模更小的同类问题 2. 定义 dpi 或 dpij 的含义,要足够清晰 3. 推导状态转移方程 4. 确定初始状态(边界条件) 5. 确定计算顺序(确保依赖的子问题先计算) 每件物品最多选一次。dpj = 容量为 j 时的最大价值,逆序遍历容量防止重复选取。 每

By yellowdog