Skip to content

Python 进阶考察的是你对语言本质的理解深度。装饰器、生成器、描述符、元类——这些是中级工程师和高级工程师的分水岭。

Q1: 装饰器的原理和实现?如何实现带参数的装饰器? 「🟡 中级」

考察点:考察对 Python 函数式编程特性的理解,装饰器是 Python 中最常用的高级特性之一。筛掉只会用不会写、不理解闭包和高阶函数的开发者。

参考答案

装饰器的本质

装饰器本质是高阶函数 + 闭包,它接收一个函数作为参数,返回一个包装后的新函数。

python
# 装饰器的本质:高阶函数
def decorator(func):
    def wrapper(*args, **kwargs):
        # 前置操作
        result = func(*args, **kwargs)
        # 后置操作
        return result
    return wrapper

# 使用方式 1:手动调用
def func():
    pass
func = decorator(func)

# 使用方式 2:语法糖 @
@decorator
def func():
    pass

基础装饰器实现

python
import time
import functools

def timer(func):
    """统计函数执行时间的装饰器"""
    @functools.wraps(func)  # 保留原函数的元数据(__name__, __doc__等)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        end = time.time()
        print(f"{func.__name__} 执行时间: {end - start:.4f}秒")
        return result
    return wrapper

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

slow_func()
# slow_func 执行时间: 1.0012秒

为什么需要 functools.wraps

python
# 没有 wraps 的情况
def decorator(func):
    def wrapper():
        return func()
    return wrapper

@decorator
def hello():
    """打招呼函数"""
    pass

print(hello.__name__)  # 'wrapper' —— 丢失了原函数名
print(hello.__doc__)   # None —— 丢失了文档字符串

# 加上 @functools.wraps(func) 后
print(hello.__name__)  # 'hello' —— 正确
print(hello.__doc__)   # '打招呼函数' —— 正确

带参数的装饰器:三层嵌套

带参数的装饰器需要再多一层函数来接收参数,返回真正的装饰器。

python
def repeat(times):
    """让函数重复执行 times 次"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for _ in range(times):
                result = func(*args, **kwargs)
            return result
        return wrapper
    return decorator

@repeat(times=3)
def greet(name):
    print(f"Hello, {name}!")

greet("Alice")
# Hello, Alice!
# Hello, Alice!
# Hello, Alice!

执行流程:

  1. @repeat(times=3) → 先调用 repeat(3),返回 decorator 函数
  2. @decorator → 再调用 decorator(greet),返回 wrapper
  3. greet = wrapper → 最终 greet 指向 wrapper

类装饰器:实现 call

用类实现装饰器,通过 __call__ 方法让实例可调用。

python
class Timer:
    def __init__(self, func):
        functools.update_wrapper(self, func)  # 类似 wraps
        self.func = func
        self.count = 0

    def __call__(self, *args, **kwargs):
        self.count += 1
        start = time.time()
        result = self.func(*args, **kwargs)
        end = time.time()
        print(f"第{self.count}次调用,耗时: {end - start:.4f}秒")
        return result

@Timer
def slow_func():
    time.sleep(0.5)

slow_func()  # 第1次调用,耗时: 0.5001秒
slow_func()  # 第2次调用,耗时: 0.5002秒

装饰器的执行顺序

多个装饰器叠加时:从下往上装饰,从上往下执行

python
def deco1(func):
    print("deco1 装饰")
    def wrapper():
        print("deco1 执行前")
        func()
        print("deco1 执行后")
    return wrapper

def deco2(func):
    print("deco2 装饰")
    def wrapper():
        print("deco2 执行前")
        func()
        print("deco2 执行后")
    return wrapper

@deco1
@deco2
def func():
    print("func 执行")

# 装饰阶段输出(定义函数时就执行了):
# deco2 装饰
# deco1 装饰

# 调用函数时输出:
# deco1 执行前
# deco2 执行前
# func 执行
# deco2 执行后
# deco1 执行后

相当于func = deco1(deco2(func)) —— 洋葱模型

装饰器的常见应用场景

场景说明
日志记录记录函数调用、参数、返回值
性能统计统计函数执行时间
权限校验登录验证、权限检查
缓存LRU 缓存(functools.lru_cache
输入校验参数合法性检查
重试机制失败自动重试
事务管理自动提交/回滚
路由注册Flask/FastAPI 的路由装饰器

追问延伸

  • 装饰器可以装饰类吗?怎么实现?
  • functools.lru_cache 的原理是什么?
  • 如何移除一个装饰器?(一般做不到,但可以通过 __wrapped__ 属性访问原函数)
  • 装饰器和面向切面编程(AOP)有什么关系?
  • 异步函数的装饰器怎么写?(wrapper 也要是 async 的)
  • 装饰器是在函数定义时执行还是调用时执行?(定义时就执行装饰逻辑)

Q2: 生成器(Generator)的原理和应用场景? 「🟡 中级」

考察点:考察对 Python 迭代器协议和惰性计算的深入理解。生成器是协程的基础,也是 Python 高级特性的核心之一。筛掉只懂列表不懂惰性计算的开发者。

参考答案

什么是生成器

生成器是一种特殊的迭代器,它使用 yield 关键字(或生成器表达式)创建,能够惰性地产生值,而不是一次性生成所有值并存储在内存中。

两种创建方式:

python
# 方式 1:生成器函数(yield)
def fibonacci(n):
    a, b = 0, 1
    for _ in range(n):
        yield a       # 执行到 yield 暂停,返回值;下次 next 继续
        a, b = b, a + b

gen = fibonacci(5)
print(type(gen))  # <class 'generator'>

# 方式 2:生成器表达式
gen = (x ** 2 for x in range(10))

生成器的工作原理

python
def gen_func():
    print("开始执行")
    yield 1
    print("继续执行")
    yield 2
    print("执行结束")

gen = gen_func()  # 调用生成器函数,不执行函数体,只返回生成器对象

print(next(gen))
# 开始执行
# 1

print(next(gen))
# 继续执行
# 2

print(next(gen))
# 执行结束
# StopIteration 异常

核心机制

  1. 调用生成器函数时,函数体不执行,只返回一个生成器对象
  2. 每次调用 next() 时,函数执行到 yield 处暂停,返回 yield 后面的值
  3. 再次调用 next() 时,从暂停处继续执行
  4. 函数执行完毕(没有更多 yield)时,抛出 StopIteration

生成器的高级方法

方法作用
next(gen)获取下一个值(启动或恢复执行)
gen.send(value)向生成器发送值,作为 yield 表达式的结果
gen.throw(type[, value[, traceback]])在生成器暂停处抛出异常
gen.close()关闭生成器(在暂停处抛出 GeneratorExit)

send() 示例:

python
def echo():
    msg = "start"
    while True:
        incoming = yield msg  # yield 右边是返回值,左边接收 send 的值
        if incoming is None:
            break
        msg = f"收到: {incoming}"

gen = echo()
print(next(gen))     # start —— 必须先启动(或 gen.send(None))
print(gen.send("hi"))   # 收到: hi
print(gen.send(123))    # 收到: 123
gen.close()

yield from:委托子生成器

yield from 可以将一个生成器委托给另一个生成器,简化生成器嵌套。

python
def inner():
    yield 1
    yield 2

def outer():
    # 不使用 yield from
    for x in inner():
        yield x
    yield 3

# 使用 yield from
def outer2():
    yield from inner()  # 等价于上面的 for 循环
    yield 3

print(list(outer2()))  # [1, 2, 3]

yield from 的额外能力:

  • 自动处理 StopIteration
  • 透明地传递 send()throw()close() 给子生成器
  • 可以获取子生成器的返回值(return 的值)

应用场景

1. 大数据流处理(节省内存)

python
# 读取超大文件,逐行处理,不用一次性加载全部内容
def read_large_file(file_path):
    with open(file_path, "r") as f:
        for line in f:
            yield line.strip()

for line in read_large_file("big_data.txt"):
    process(line)

2. 无限序列

python
def infinite_counter():
    n = 0
    while True:
        yield n
        n += 1

counter = infinite_counter()
print(next(counter))  # 0
print(next(counter))  # 1
# ... 可以无限取下去

3. 管道(Pipeline)模式

python
def numbers():
    for i in range(10):
        yield i

def even(nums):
    for n in nums:
        if n % 2 == 0:
            yield n

def square(nums):
    for n in nums:
        yield n ** 2

# 数据像管道一样层层流过
result = square(even(numbers()))
print(list(result))  # [0, 4, 16, 36, 64]

4. 协程的基础

生成器的 send() 能力是 Python 协程的雏形,asyncio 早期就是基于生成器实现的。

生成器 vs 列表性能对比

特性列表生成器
内存占用O(n),所有数据在内存O(1),惰性计算
创建速度慢(一次性生成所有)快(几乎立即返回)
遍历次数多次一次
支持索引
适用场景数据量小、需要随机访问数据量大、流式处理

追问延伸

  • 生成器和迭代器有什么区别和联系?
  • yield from 和普通的 for y in x: yield y 有什么区别?(send/throw/close 的透传、返回值)
  • 生成器的 return 值怎么获取?(通过 StopIteration.value)
  • 什么是生成器的"预激"(prime)?为什么需要 next() 或 send(None)?
  • itertools 模块中的生成器工具了解多少?
  • 生成器表达式和列表推导式在内存上差多少?(数据量越大差异越显著)

Q3: 什么是协程?Python中的协程演进历史? 「🟡 中级」

考察点:考察对 Python 异步编程演进的理解,以及对并发模型的认识。筛掉只会写同步代码、对 asyncio 一知半解的开发者。

参考答案

什么是协程

协程(Coroutine)是一种用户态的轻量级线程,由程序自身控制调度(协作式调度),而非操作系统抢占式调度。

协程 vs 线程 vs 进程:

特性进程线程协程
调度者操作系统操作系统用户程序
调度方式抢占式抢占式协作式
开销大(独立地址空间)中(共享地址空间)极小(函数级)
数量级几十~几百几千几十万~几百万
切换开销大(内核态切换)中(内核态切换)极小(用户态切换)
数据共享复杂(IPC)简单(共享内存,需锁)最简单(同一线程内)
多核利用是(但 GIL 限制)否(单线程内)

Python 协程的演进历史

第一阶段:生成器协程(Python 2.5+)

利用生成器的 yieldsend() 实现协程的雏形。

python
def coro():
    print("启动")
    x = yield
    print(f"收到: {x}")
    y = yield x * 2
    print(f"收到: {y}")

c = coro()
next(c)        # 启动,输出"启动",停在第一个 yield
result = c.send(10)  # 输出"收到: 10",返回 20
print(result)  # 20

特点:

  • 本质是生成器的"副作用"用法
  • 需要手动预激(next 或 send(None))
  • 没有专门的语法和类型

第二阶段:@asyncio.coroutine + yield from(Python 3.4)

正式引入 asyncio 库,用装饰器标记协程函数,用 yield from 调用子协程。

python
import asyncio

@asyncio.coroutine
def fetch_data():
    yield from asyncio.sleep(1)  # 模拟 IO 操作
    return "data"

@asyncio.coroutine
def main():
    result = yield from fetch_data()
    print(result)

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

特点:

  • 有了标准库 asyncio
  • 协程和生成器仍然藕断丝连(基于生成器实现)
  • yield from 既可以委托生成器,也可以调用协程,容易混淆

第三阶段:async/await(Python 3.5+)

引入 async/await 语法,协程成为 Python 的一等公民,与生成器彻底分离。

python
import asyncio

async def fetch_data():
    await asyncio.sleep(1)  # 挂起点
    return "data"

async def main():
    result = await fetch_data()
    print(result)

asyncio.run(main())  # Python 3.7+ 简化入口

async/await 的核心概念:

概念说明
async def定义协程函数,调用返回协程对象(不会立即执行)
await挂起当前协程,等待另一个协程完成(只能在 async 函数内使用)
事件循环(Event Loop)调度协程执行的核心,负责 IO 事件监听和协程切换
Task协程的调度单元,将协程包装成任务加入事件循环

协程的执行模型:

调用 async 函数 → 返回协程对象(未执行)

await 协程 / asyncio.create_task() → 加入事件循环

事件循环调度执行 → 遇到 await 挂起 → IO 完成后恢复

协程执行完毕 → 返回结果

为什么协程适合 IO 密集型?

同步方式(顺序执行):
  请求1: [发送][等待响应...][接收]
  请求2:                     [发送][等待响应...][接收]
  总时间 = 2 * (发送 + 等待 + 接收)

协程方式(并发执行):
  请求1: [发送][等待响应...        ][接收]
  请求2:      [发送][等待响应...   ][接收]
  总时间 ≈ 发送 + max(等待) + 接收

IO 等待时,CPU 是空闲的。协程在 IO 等待时挂起,让其他协程执行,充分利用 CPU。

常见误区

  1. "用了 async 就会变快"——错。CPU 密集型任务用协程不会更快,反而因为调度开销更慢。协程只对 IO 密集型有效。

  2. "async 函数调用了就会执行"——错。调用 async 函数只返回协程对象,需要 await 或放入事件循环才会执行。

  3. "协程是多线程的"——错。默认情况下,asyncio 的所有协程都在单线程的事件循环中运行。

  4. "有了 async/await 就不需要多线程了"——错。CPU 密集型还是需要多进程,某些场景(如阻塞库无法改造成异步)还是需要线程池。

追问延伸

  • asyncio 的事件循环原理是什么?(基于 epoll/kqueue/iocp 的 IO 多路复用)
  • 协程和线程的调度有什么本质区别?(协作式 vs 抢占式)
  • 什么是"异步编程的传染性"?(一个 await 导致整条调用链都要 async)
  • 如何在 asyncio 中使用阻塞的同步函数?(loop.run_in_executor + 线程池/进程池)
  • asyncio.create_task 和直接 await 有什么区别?(前者并发执行,后者顺序执行)
  • 你用过哪些异步框架?(aiohttp、FastAPI、Tornado 等)
  • Python 3.11 中的 TaskGroup 了解吗?(结构化并发,类似 trio 的 nursery)

Q4: 描述符(Descriptor)是什么?有什么应用? 「🔴 高级」

考察点:考察对 Python 对象模型底层机制的理解深度。描述符是 Python 面向对象的基石之一,property、classmethod、staticmethod 都是基于描述符实现的。能区分初中高级工程师。

参考答案

什么是描述符

描述符是实现了描述符协议的类,即实现了 __get____set____delete__ 中任意一个方法的类。描述符的作用是控制属性访问的行为

描述符必须定义在类级别(作为类属性),不能定义在实例上。

描述符协议

python
class Descriptor:
    def __get__(self, instance, owner):
        # 读取属性时调用
        # instance: 被访问的实例(如果是类访问,则为 None)
        # owner: 拥有该描述符的类
        return self.value

    def __set__(self, instance, value):
        # 设置属性时调用
        self.value = value

    def __delete__(self, instance):
        # 删除属性时调用
        del self.value

数据描述符 vs 非数据描述符

类型定义优先级
数据描述符(Data Descriptor)同时实现了 __get____set__高于实例字典
非数据描述符(Non-data Descriptor)只实现了 __get__低于实例字典

属性查找顺序(从高到低):

  1. 数据描述符的 __get__
  2. 实例字典(instance.__dict__
  3. 非数据描述符的 __get__
  4. 类字典(type(instance).__dict__
  5. 父类字典(按 MRO 顺序)
  6. __getattr__(如果定义了)
python
# 数据描述符:优先级高于实例字典
class DataDesc:
    def __get__(self, inst, owner):
        return "data desc"
    def __set__(self, inst, val):
        pass

class A:
    x = DataDesc()

a = A()
a.__dict__["x"] = "instance"
print(a.x)  # "data desc" —— 数据描述符优先

# 非数据描述符:实例字典优先
class NonDataDesc:
    def __get__(self, inst, owner):
        return "non-data desc"

class B:
    x = NonDataDesc()

b = B()
b.__dict__["x"] = "instance"
print(b.x)  # "instance" —— 实例字典优先

手动实现一个简易版 property

python
class MyProperty:
    def __init__(self, fget=None, fset=None, fdel=None):
        self.fget = fget
        self.fset = fset
        self.fdel = fdel

    def __get__(self, instance, owner):
        if instance is None:
            return self  # 类访问时返回描述符自身
        if self.fget is None:
            raise AttributeError("unreadable attribute")
        return self.fget(instance)

    def __set__(self, instance, value):
        if self.fset is None:
            raise AttributeError("can't set attribute")
        self.fset(instance, value)

    def __delete__(self, instance):
        if self.fdel is None:
            raise AttributeError("can't delete attribute")
        self.fdel(instance)

    def setter(self, fset):
        return MyProperty(self.fget, fset, self.fdel)

    def deleter(self, fdel):
        return MyProperty(self.fget, self.fset, fdel)


# 使用示例
class Circle:
    def __init__(self, radius):
        self._radius = radius

    @MyProperty
    def radius(self):
        return self._radius

    @radius.setter
    def radius(self, value):
        if value <= 0:
            raise ValueError("半径必须为正数")
        self._radius = value

    @MyProperty
    def area(self):
        return 3.14 * self._radius ** 2

c = Circle(5)
print(c.radius)  # 5
print(c.area)    # 78.5
c.radius = 10
print(c.area)    # 314.0

描述符的应用场景

应用说明
property计算属性、属性验证、惰性属性(数据描述符)
classmethod将方法绑定到类而非实例(非数据描述符)
staticmethod普通函数,不绑定(非数据描述符)
ORM 字段Django ORM、SQLAlchemy 的字段类型
类型验证属性类型检查(类似 dataclass 字段验证)
延迟加载第一次访问才计算并缓存

classmethod 和 staticmethod 的描述符原理

python
# classmethod 简化实现
class classmethod:
    def __init__(self, func):
        self.func = func

    def __get__(self, instance, owner):
        # 类方法:绑定到类(owner),不管是实例访问还是类访问
        return lambda *args, **kwargs: self.func(owner, *args, **kwargs)

# staticmethod 简化实现
class staticmethod:
    def __init__(self, func):
        self.func = func

    def __get__(self, instance, owner):
        # 静态方法:不绑定,直接返回原函数
        return self.func

追问延伸

  • 为什么 property 是数据描述符而 classmethod 是非数据描述符?
  • 描述符为什么必须定义为类属性?(属性查找逻辑在 type.getattribute 中)
  • __getattribute__ 和描述符有什么关系?(__getattribute__ 内部调用描述符的 __get__
  • 你在项目中用过描述符吗?解决了什么问题?
  • Django ORM 的 CharField 是怎么通过描述符工作的?
  • 什么是 __set_name__?(Python 3.6+,描述符被赋值为类属性时调用,获取属性名)

Q5: property的原理?如何手动实现一个property? 「🔴 高级」

考察点:考察对描述符机制的掌握,以及对 Python 核心内置特性实现原理的理解。property 是描述符最经典的应用,这道题能检验候选人是否真正理解描述符。

参考答案

property 是什么

property 是 Python 内置的一个描述符类,用于将方法"伪装"成属性访问,实现计算属性、属性验证、惰性求值等功能。

基本用法

python
class Person:
    def __init__(self, first_name, last_name):
        self.first_name = first_name
        self.last_name = last_name

    @property
    def full_name(self):
        """只读计算属性"""
        return f"{self.first_name} {self.last_name}"

    @property
    def age(self):
        return self._age

    @age.setter
    def age(self, value):
        if not isinstance(value, int) or value < 0 or value > 150:
            raise ValueError("年龄不合法")
        self._age = value

    @age.deleter
    def age(self):
        del self._age

p = Person("张", "三")
print(p.full_name)  # 张 三 —— 像属性一样访问,实际调用方法
p.age = 25          # 触发 setter 验证
print(p.age)        # 25

property 的完整签名

python
property(fget=None, fset=None, fdel=None, doc=None)
参数作用
fget获取属性的方法
fset设置属性的方法
fdel删除属性的方法
doc属性的文档字符串

手动实现 property

基于描述符协议,实现一个功能完整的 property:

python
class MyProperty:
    def __init__(self, fget=None, fset=None, fdel=None, doc=None):
        self.fget = fget
        self.fset = fset
        self.fdel = fdel
        self.__doc__ = doc or (fget.__doc__ if fget else None)

    def __get__(self, instance, owner):
        if instance is None:
            # 类访问时,返回 property 对象自身
            return self
        if self.fget is None:
            raise AttributeError("unreadable attribute")
        return self.fget(instance)

    def __set__(self, instance, value):
        if self.fset is None:
            raise AttributeError("can't set attribute")
        self.fset(instance, value)

    def __delete__(self, instance):
        if self.fdel is None:
            raise AttributeError("can't delete attribute")
        self.fdel(instance)

    def getter(self, fget):
        """装饰器:设置 getter"""
        return type(self)(fget, self.fset, self.fdel, self.__doc__)

    def setter(self, fset):
        """装饰器:设置 setter"""
        return type(self)(self.fget, fset, self.fdel, self.__doc__)

    def deleter(self, fdel):
        """装饰器:设置 deleter"""
        return type(self)(self.fget, self.fset, fdel, self.__doc__)

为什么 property 是数据描述符

property 同时实现了 __get____set__,所以是数据描述符。这意味着:

python
class A:
    @property
    def x(self):
        return 42

a = A()
a.__dict__["x"] = 100  # 试图在实例字典中设置 x
print(a.x)  # 42 —— 数据描述符优先级更高,实例字典中的 x 被忽略

为什么要设计成数据描述符?

  • 如果是非数据描述符,实例字典的同名属性会覆盖 property,导致 setter 逻辑被绕过
  • 数据描述符确保了 property 的控制权,即使实例字典中有同名属性也会被忽略

property 的应用场景

1. 计算属性(Derived Attribute)

python
class Rectangle:
    def __init__(self, width, height):
        self.width = width
        self.height = height

    @property
    def area(self):
        return self.width * self.height

    @property
    def perimeter(self):
        return 2 * (self.width + self.height)

2. 属性验证(Validation)

python
class User:
    @property
    def email(self):
        return self._email

    @email.setter
    def email(self, value):
        if "@" not in value:
            raise ValueError("无效的邮箱地址")
        self._email = value

3. 惰性属性(Lazy Attribute)

python
class ExpensiveObject:
    def __init__(self):
        self._data = None

    @property
    def data(self):
        if self._data is None:
            print("加载数据中...")  # 第一次访问才加载
            self._data = self._load_expensive_data()
        return self._data

    def _load_expensive_data(self):
        # 模拟耗时操作
        return {"key": "value"}

4. 向后兼容(从属性升级为 getter/setter)

Python 的 property 使得"属性 → 计算属性"的升级无需修改调用方代码,这是 Python 的设计哲学之一。

property vs 直接属性访问的性能

python
# 直接访问属性最快(直接查字典)
# property 略慢(需要调用描述符的 __get__,再调用 fget 方法)
# 但差异通常可以忽略,除非在极热路径上

追问延伸

  • 只读 property 和只写 property 怎么实现?
  • cached_property(Python 3.8+ functools)了解吗?和普通 property 有什么区别?
  • 如何实现一个带缓存的 property?(非数据描述符 + 实例字典缓存)
  • property 和 __getattr__/__getattribute__ 有什么区别?
  • dataclass 中的 field 可以和 property 配合使用吗?
  • 什么情况下应该用 property,什么情况下应该用普通方法?

Q6: 什么是元类(Metaclass)?应用场景? 「🔴 高级」

考察点:考察对 Python 对象模型最核心机制的理解。元类是 Python 最晦涩也是最强大的特性之一,俗称"Python 的黑魔法"。能区分高级工程师和资深工程师。

参考答案

什么是元类

元类是创建类的类(Metaclass = Meta + Class)。

在 Python 中,一切皆对象,类本身也是对象。那么类这个对象是由谁创建的?答案是:元类。

对象  ←  类  ←  元类
实例     (class)   (metaclass)

type 是 Python 的默认元类。所有的类(包括 type 自己)都是由 type 创建的。

type 的双重身份

type 既是一个内置函数(用来获取对象的类型),也是一个元类(用来创建类)。

python
# type 作为函数:获取类型
print(type(42))          # <class 'int'>
print(type(int))         # <class 'type'>
print(type(type))        # <class 'type'> —— type 的类型是它自己

# type 作为元类:动态创建类
# type(类名, 父类元组, 属性字典)
MyClass = type("MyClass", (), {"x": 1, "hello": lambda self: "hello"})

obj = MyClass()
print(obj.x)          # 1
print(obj.hello())    # hello
print(type(MyClass))  # <class 'type'>

等价于:

python
class MyClass:
    x = 1
    def hello(self):
        return "hello"

自定义元类

自定义元类继承自 type,通过重写 __new____init__ 来控制类的创建过程。

python
class MyMeta(type):
    def __new__(mcs, name, bases, attrs):
        # mcs: 元类自身(MyMeta)
        # name: 正在创建的类名
        # bases: 父类元组
        # attrs: 类的属性字典

        # 在这里可以修改类的属性、添加方法、做验证等
        attrs["class_id"] = id(name)  # 给每个类添加 class_id 属性

        print(f"创建类: {name}")
        return super().__new__(mcs, name, bases, attrs)

    def __init__(cls, name, bases, attrs):
        # 类创建完成后的初始化
        print(f"初始化类: {name}")
        super().__init__(name, bases, attrs)


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

# 输出:
# 创建类: MyClass
# 初始化类: MyClass

元类的执行时机

元类的 __new____init__ 是在类定义时执行的(不是实例化时):

python
print("开始定义类")

class A(metaclass=MyMeta):
    print("类定义体执行中")
    x = 1

print("类定义完成")
print("开始创建实例")
a = A()

# 输出顺序:
# 开始定义类
# 类定义体执行中
# 创建类: A
# 初始化类: A
# 类定义完成
# 开始创建实例

元类的应用场景

1. ORM 框架(最经典的应用)

Django ORM 和 SQLAlchemy 都使用了元类来实现声明式的模型定义:

python
# Django ORM 的简化模拟
class ModelMeta(type):
    def __new__(mcs, name, bases, attrs):
        if name == "Model":  # 基类不处理
            return super().__new__(mcs, name, bases, attrs)

        # 收集字段定义
        fields = {}
        for k, v in list(attrs.items()):
            if isinstance(v, Field):
                fields[k] = v

        attrs["_fields"] = fields
        attrs["_table_name"] = name.lower()
        return super().__new__(mcs, name, bases, attrs)

class Model(metaclass=ModelMeta):
    pass

class User(Model):
    name = CharField(max_length=100)
    age = IntegerField()

# User 类自动拥有了 _fields 和 _table_name 属性
print(User._table_name)   # user
print(User._fields)       # {'name': CharField(...), 'age': IntegerField(...)}

2. 类的自动注册

python
class RegistryMeta(type):
    registry = {}

    def __new__(mcs, name, bases, attrs):
        cls = super().__new__(mcs, name, bases, attrs)
        if name != "Base":  # 排除基类
            mcs.registry[name] = cls
        return cls

class Base(metaclass=RegistryMeta):
    pass

class PluginA(Base):
    pass

class PluginB(Base):
    pass

print(RegistryMeta.registry)
# {'PluginA': <class 'PluginA'>, 'PluginB': <class 'PluginB'>}

3. 单例模式

python
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 Singleton(metaclass=SingletonMeta):
    pass

a = Singleton()
b = Singleton()
print(a is b)  # True

4. API 框架的路由注册(类似 Flask 的 @app.route,但用元类实现类级别的路由)

元类的争议:99% 的场景不需要

"元类是 99% 的人永远不需要关心的黑魔法。如果你犹豫是否需要元类,那你就不需要。" —— Python 社区共识

替代方案(优先考虑):

需求优先方案元类(最后手段)
修改类行为装饰器元类
类注册__init_subclass__ + 基类元类
单例模块级变量、装饰器元类
ORM 字段描述符 + __init_subclass__元类

__init_subclass__(Python 3.6+):元类的轻量替代

python
class Base:
    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        # 子类创建时自动调用
        print(f"子类 {cls.__name__} 被创建了")
        cls.registry = {}  # 给子类添加属性

class Sub(Base):
    pass
# 输出: 子类 Sub 被创建了

追问延伸

  • 元类的 __new____init____call__ 分别控制什么?
  • Python 中的 __metaclass__ 属性和 metaclass 参数有什么区别?(Python 2 vs 3)
  • 元类和装饰器有什么区别?各适合什么场景?
  • Django ORM 的 Model 元类做了哪些事情?
  • 什么是抽象基类(ABC)?它和元类有什么关系?(ABCMeta)
  • 你在项目中使用过元类吗?解决了什么问题?为什么不用其他方案?

Q7: Python中的MRO是什么?多继承的方法解析顺序? 「🟡 中级」

考察点:考察对 Python 多继承机制的理解。MRO 是多继承的核心,也是理解 super() 的基础。筛掉只会单继承、对多继承一知半解的开发者。

参考答案

什么是 MRO

MRO(Method Resolution Order,方法解析顺序)是 Python 中多继承情况下查找方法和属性的顺序。

当调用一个对象的方法时,Python 会按照 MRO 列表从左到右依次查找,直到找到第一个匹配的方法。

python
class D(B, C):
    pass

print(D.__mro__)      # 元组形式
print(D.mro())        # 列表形式
# D → B → C → A → object

C3 线性化算法

Python 3 使用 C3 线性化(C3 Linearization) 算法计算 MRO(也叫 MRO 2.3)。

C3 算法的三个原则:

  1. 子类优先于父类:子类在 MRO 中排在父类前面
  2. 父类顺序保留:如果在类定义中 class D(B, C),那么 B 在 C 前面
  3. 单调性(Monotonicity):任何类的 MRO 中,其父类的顺序也必须符合各父类自身的 MRO 顺序

经典示例:菱形继承

    A
   / \
  B   C
   \ /
    D
python
class A:
    def say(self):
        print("A")

class B(A):
    def say(self):
        print("B")

class C(A):
    def say(self):
        print("C")

class D(B, C):
    pass

print(D.mro())
# [D, B, C, A, object]

D().say()  # B —— 先找到 B 的 say 方法

如果是 class D(C, B),MRO 就是 [D, C, B, A, object]

更复杂的例子

      A
     / \
    B   C
   / \ / \
  D   E   F
   \ / \ /
    G   H
     \ /
      I
python
class A: pass
class B(A): pass
class C(A): pass
class D(B): pass
class E(B, C): pass
class F(C): pass
class G(D, E): pass
class H(E, F): pass
class I(G, H): pass

print([c.__name__ for c in I.mro()])
# ['I', 'G', 'D', 'H', 'E', 'B', 'F', 'C', 'A', 'object']

验证三个原则:

  1. 子类在父类前:I 在最前,A 在最后 ✓
  2. 父类顺序保留:G 定义时 D 在 E 前 → D 在 E 前 ✓
  3. 单调性:E 的 MRO 是 E→B→C→A,在 I 的 MRO 中 B 在 C 前 ✓

super() 的本质

super() 调用的不是父类的方法,而是 MRO 中下一个类的方法。

python
class A:
    def say(self):
        print("A")

class B(A):
    def say(self):
        print("B start")
        super().say()
        print("B end")

class C(A):
    def say(self):
        print("C start")
        super().say()
        print("C end")

class D(B, C):
    def say(self):
        print("D start")
        super().say()
        print("D end")

D().say()
# D start
# B start
# C start
# A
# C end
# B end
# D end

执行流程(沿着 MRO 链传递):

D.say() → super() → B.say() → super() → C.say() → super() → A.say()

因为 MRO 是 D → B → C → A → object,所以:

  • D 的 super 是 B
  • B 的 super 是 C(不是 A!因为 MRO 中 B 的下一个是 C)
  • C 的 super 是 A

这就是为什么 super() 被称为"super"而不是"parent"——它是 MRO 中的下一个,不一定是父类。

super() 的完整签名

python
super(type, obj_or_type=None)
  • super() 在实例方法中等价于 super(__class__, self)
  • super() 在类方法中等价于 super(__class__, cls)
  • 也可以显式指定:super(C, self).say() 跳过 C,从 C 的下一个开始找

多继承的最佳实践

  1. 优先用组合而非继承(Composition over Inheritance)
  2. 如果必须多继承,尽量让父类职责单一(Mixin 模式)
  3. 使用 super() 而非显式调用父类方法
  4. 保持方法签名一致(参数兼容)
  5. 菱形继承中确保最顶层基类的方法是"终点"(不再调用 super)

Mixin 模式示例:

python
# Mixin 类:不独立使用,只为其他类提供额外功能
class LoggingMixin:
    def save(self):
        print(f"保存前记录日志: {self}")
        super().save()  # 调用 MRO 中下一个类的 save

class ValidationMixin:
    def save(self):
        print(f"保存前验证: {self}")
        super().save()

class BaseModel:
    def save(self):
        print("实际保存数据")

# Mixin 顺序:从左到右依次"包裹"
class User(LoggingMixin, ValidationMixin, BaseModel):
    pass

User().save()
# 保存前记录日志: <User object>
# 保存前验证: <User object>
# 实际保存数据

追问延伸

  • Python 2 的经典类(Classic Class)和新式类(New-style Class)的 MRO 有什么区别?
  • 什么情况下 C3 算法会失败(无法计算出合法的 MRO)?
  • Mixin 模式的最佳实践有哪些?Mixin 类为什么通常不写 __init__
  • super()__init__ 中使用有什么坑?(参数不匹配问题)
  • 为什么说 Python 的多继承比 Java 的接口更灵活但也更危险?
  • 你在项目中用过哪些 Mixin?(比如 Django 的 LoginRequiredMixin)

Q8: 什么是猴子补丁(Monkey Patching)?有什么优缺点? 「🟡 中级」

考察点:考察对 Python 动态特性的理解,以及工程实践中的权衡能力。筛掉只会理论不懂实战、对动态语言特性没有实践经验的开发者。

参考答案

什么是猴子补丁

猴子补丁(Monkey Patching)是指在运行时动态修改类或模块的行为,而不修改原始源代码。

名字来源:"Monkey" 来自 "guerrilla"(游击战)的谐音,意为偷偷摸摸地修改。也有说法是因为 "monkey around with code"(胡乱摆弄代码)。

python
# 示例:给 list 动态添加一个方法
def prepend(self, item):
    self.insert(0, item)

list.prepend = prepend  # 猴子补丁:运行时修改内置类

my_list = [2, 3, 4]
my_list.prepend(1)
print(my_list)  # [1, 2, 3, 4]

常见应用场景

1. 紧急修复(Hotfix)

线上出 bug 了,第三方库有问题,来不及等官方修复,先用猴子补丁顶一下:

python
import some_lib

# 原始函数有 bug
original_buggy_func = some_lib.buggy_func

def fixed_func(*args, **kwargs):
    # 修复逻辑
    result = original_buggy_func(*args, **kwargs)
    result = result.replace("bug", "fix")
    return result

some_lib.buggy_func = fixed_func  # 打补丁

2. 测试中的 Mock

python
from unittest.mock import patch

# 临时替换函数/类,模拟特定行为
with patch("requests.get") as mock_get:
    mock_get.return_value.status_code = 200
    mock_get.return_value.json.return_value = {"data": "test"}
    result = my_function_that_uses_requests()

3. 给第三方库增加功能

第三方库缺个你需要的功能,又不想维护 fork 版本:

python
import pandas as pd

# 给 DataFrame 增加自定义方法
def to_markdown(df):
    # 实现转换为 Markdown 表格的逻辑
    pass

pd.DataFrame.to_markdown = to_markdown

df = pd.DataFrame(...)
df.to_markdown()  # 直接使用

4. 性能监控/日志注入

python
import time
import sqlalchemy

original_execute = sqlalchemy.engine.Connection.execute

def monitored_execute(self, *args, **kwargs):
    start = time.time()
    result = original_execute(self, *args, **kwargs)
    elapsed = time.time() - start
    if elapsed > 0.5:
        print(f"慢 SQL: {args[0]}, 耗时: {elapsed:.2f}s")
    return result

sqlalchemy.engine.Connection.execute = monitored_execute

优点

优点说明
灵活性高可以在不修改源码的情况下改变行为
快速修复线上紧急问题可以快速打补丁
扩展第三方库给无法修改的第三方库增加功能
测试便利Mock 就是猴子补丁的典型应用
动态适配根据运行环境动态调整行为

缺点和风险

缺点说明
代码难理解行为不在源码里,别人看代码时不知道被改过
调试困难出问题时很难定位是哪里打的补丁
版本不兼容第三方库升级后补丁可能失效或导致新 bug
多补丁冲突多个补丁修改同一个东西时可能互相影响
测试不充分补丁逻辑容易绕过正常测试
违反封装破坏了类的封装性,依赖内部实现细节

最佳实践

  1. 尽量避免使用——能用继承、装饰器、组合解决的,就不要用猴子补丁
  2. 作为最后手段——只有当其他方案都不可行时才考虑
  3. 集中管理——所有补丁放在一个地方(如 monkey_patches.py),不要散落在各处
  4. 添加注释——详细说明为什么打补丁、补丁做了什么、哪个版本开始需要
  5. 版本检查——检查第三方库版本,避免版本不兼容
  6. 及时跟进——等官方修复后及时移除补丁
  7. 写测试——为补丁逻辑编写专门的测试用例
python
# 良好实践:集中管理 + 版本检查 + 注释
# monkey_patches.py

import requests
import logging

logger = logging.getLogger(__name__)

def patch_requests():
    """
    给 requests 增加超时时间的默认值。
    原因:requests 默认没有超时,可能导致请求永久挂起。
    版本:适用于 requests 2.x
    跟踪:https://github.com/psf/requests/issues/2393
    """
    if requests.__version__.startswith("3."):
        logger.warning("requests 已升级到 3.x,请检查是否还需要这个补丁")
        return

    original_request = requests.Session.request

    def request_with_timeout(self, method, url, **kwargs):
        kwargs.setdefault("timeout", 30)  # 默认 30 秒超时
        return original_request(self, method, url, **kwargs)

    requests.Session.request = request_with_timeout
    logger.info("已应用 requests 超时补丁")

追问延伸

  • gevent 的猴子补丁了解吗?原理是什么?(gevent.monkey.patch_all() 替换标准库的阻塞 IO 为非阻塞版本)
  • Mock 和猴子补丁有什么关系?
  • 猴子补丁和装饰器有什么区别?各适合什么场景?
  • 什么情况下猴子补丁会导致难以调试的 bug?
  • 你在项目中打过猴子补丁吗?解决了什么问题?最后移除了吗?
  • Python 作为动态语言,还有哪些"动态特性"?(动态添加属性/方法、动态创建类、setattr/getattr 等)

Q9: Python中的垃圾回收机制?引用计数 + 标记清除 + 分代回收 「🟡 中级」

考察点:考察对 Python 内存管理机制的系统理解。垃圾回收是后端工程师的基本功,也是性能调优的基础。筛掉只写代码不关心内存的开发者。

参考答案

Python 的垃圾回收(Garbage Collection, GC)采用的是引用计数为主,标记-清除和分代回收为辅的策略。

一、引用计数(Reference Counting)

原理:每个对象都维护一个引用计数 ob_refcnt,当引用计数归零时,对象立即被回收。

python
import sys

a = [1, 2, 3]
print(sys.getrefcount(a))  # 2 —— a 和 getrefcount 的参数各一次引用

b = a
print(sys.getrefcount(a))  # 3 —— b 也引用了

del b
print(sys.getrefcount(a))  # 2 —— 删除 b 后减少

引用计数增加的场景:

  • 对象被赋值给变量:a = obj
  • 对象被放入容器:lst.append(obj)
  • 对象作为函数参数传递
  • 对象作为返回值返回

引用计数减少的场景:

  • 变量被显式删除:del a
  • 变量被重新赋值:a = None
  • 对象离开作用域(函数返回后,局部变量被销毁)
  • 容器被销毁或从中移除对象

优点:

  • 实时性:引用计数为零立即回收,没有延迟
  • 简单:实现简单,容易理解
  • 暂停时间短:不需要一次性扫描所有对象

缺点:

  • 无法处理循环引用(最大的问题)
  • 额外开销:每个对象都要维护引用计数,频繁加减
  • 空间开销:每个对象多一个计数字段(通常是一个指针大小)

二、标记-清除(Mark and Sweep)

标记-清除是为了解决循环引用问题而引入的补充机制。

循环引用示例:

python
a = [1]
b = [2]
a.append(b)  # a 引用 b
b.append(a)  # b 引用 a
del a
del b
# 此时 a 和 b 的引用计数都是 1(互相引用),但外部已经访问不到了
# 引用计数无法回收它们 —— 这就是循环引用导致的内存泄漏

算法流程:

  1. 标记阶段:从根对象(root object)出发,遍历所有可达对象并标记

    • 根对象包括:全局变量、栈上的变量、寄存器中的引用等
    • 能从根对象到达的对象 → 存活对象
    • 不能到达的对象 → 垃圾
  2. 清除阶段:遍历所有对象,回收未被标记的对象

    根集合


    ┌───┐    ┌───┐
    │ A │───▶│ B │    可达(存活)
    └───┘    └───┘

    ┌───┐⇄───┌───┐
    │ C │    │ D │    不可达(垃圾,循环引用)
    └───┘    └───┘

优点:

  • 可以回收循环引用
  • 没有额外的运行时开销(不像引用计数需要频繁加减)

缺点:

  • 需要 STW(Stop-The-World),暂停应用程序
  • 扫描所有对象,耗时与对象数量成正比

三、分代回收(Generational Collection)

原理:基于"弱分代假说"——大部分对象的生命周期都很短,存活越久的对象越不可能是垃圾

Python 将对象分为 3 代(Generation 0, 1, 2):

说明检查频率
0 代(年轻代)新创建的对象最频繁
1 代(中年代)在 0 代 GC 中存活下来的对象中等
2 代(老年代)在 1 代 GC 中存活下来的对象最少

对象晋升:每次 GC 后仍然存活的对象,会晋升到下一代。

触发条件(默认阈值):

python
import gc
print(gc.get_threshold())
# (700, 10, 10)
# 含义:
# - 0 代对象数 - 释放数 > 700 时,触发 0 代 GC
# - 每 10 次 0 代 GC,触发 1 次 1 代 GC
# - 每 10 次 1 代 GC,触发 1 次 2 代 GC

优点:

  • 减少每次 GC 的扫描范围,提高效率
  • 新生代对象多但回收也快,老年代对象少且稳定

缺点:

  • 实现更复杂
  • 老年代的对象可能长期不被回收

三种机制的关系

引用计数(主力,日常回收)
    ↓ 无法处理循环引用
标记-清除(补充,定期执行)
    ↓ 扫描全量对象太慢
分代回收(优化,按年龄分层回收)
  • 引用计数是主要机制,大部分对象靠它回收
  • 标记-清除只针对容器对象(可能产生循环引用的对象)
  • 分代回收是标记-清除的优化策略,提高 GC 效率

gc 模块常用 API

python
import gc

gc.enable()       # 启用自动 GC
gc.disable()      # 禁用自动 GC
gc.isenabled()    # 是否启用

gc.collect()      # 手动触发全量 GC(返回回收的对象数)
gc.collect(0)     # 只回收 0 代
gc.collect(1)     # 回收 0 代和 1 代
gc.collect(2)     # 回收所有代(默认)

gc.set_threshold(700, 10, 10)  # 设置阈值
gc.get_threshold()              # 获取阈值

gc.get_objects()  # 获取所有被跟踪的对象列表
gc.garbage        # 无法回收的垃圾对象列表(通常是有 __del__ 的循环引用)

追问延伸

  • 什么情况下会出现内存泄漏?Python 不是自动 GC 吗?(循环引用 + del、全局变量累积、C 扩展泄漏等)
  • __del__ 方法为什么会导致 GC 问题?(有 del 的循环引用对象无法确定调用顺序,被放入 gc.garbage)
  • 如何排查 Python 的内存泄漏?(gc.garbageobjgraphtracemallocmemory_profiler
  • 引用计数的增减操作是原子的吗?和 GIL 有什么关系?
  • CPython 为什么选择引用计数为主?(历史原因、简单实时、C 扩展简单)
  • PyPy 的 GC 和 CPython 有什么不同?(PyPy 用分代复制 GC,没有引用计数,性能更好)

Q10: 什么是循环引用?怎么检测和解决? 「🟡 中级」

考察点:考察对 Python 垃圾回收机制中核心问题的理解,以及实际排查内存问题的能力。筛掉理论懂但不会实际排查问题的开发者。

参考答案

什么是循环引用

循环引用(Circular Reference)是指两个或多个对象互相引用,形成一个闭环,导致它们的引用计数永远不会归零。

python
# 简单的循环引用
a = [1]
b = [2]
a.append(b)
b.append(a)
del a
del b
# a 和 b 互相引用,引用计数都为 1,但已经无法从外部访问
# 引用计数机制无法回收这两个对象

常见场景:

python
# 1. 双向链表
class Node:
    def __init__(self):
        self.prev = None
        self.next = None

a = Node()
b = Node()
a.next = b
b.prev = a  # 双向引用

# 2. 树结构的父子关系
class TreeNode:
    def __init__(self, parent=None):
        self.parent = parent
        self.children = []
        if parent:
            parent.children.append(self)  # 父子互相引用

# 3. 观察者模式
class Observer:
    def __init__(self, subject):
        self.subject = subject
        subject.attach(self)

class Subject:
    def __init__(self):
        self.observers = []

    def attach(self, observer):
        self.observers.append(observer)  # 互相引用

# 4. 闭包和回调
class Handler:
    def __init__(self):
        self.callback = self.handle  # 方法引用实例,实例引用方法

    def handle(self):
        pass

怎么检测循环引用

方法 1:使用 gc 模块

python
import gc

# 先手动触发一次 GC
gc.collect()

# gc.garbage 中存放的是无法回收的对象(通常是有 __del__ 的循环引用)
print(gc.garbage)

# 开启调试模式,打印循环引用的详细信息
gc.set_debug(gc.DEBUG_LEAK)
gc.collect()  # 会打印出被回收的循环引用对象详情

方法 2:使用 objgraph 库

bash
pip install objgraph
python
import objgraph

# 显示增长最多的类型
objgraph.show_growth()

# 显示某个对象的引用链(为什么还没被回收)
objgraph.show_backrefs(some_object, max_depth=5)

# 查找类型的所有实例
objgraph.by_type("MyClass")

方法 3:使用 tracemalloc(Python 3.4+)

python
import tracemalloc

tracemalloc.start()

# ... 你的代码 ...

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

for stat in top_stats[:10]:
    print(stat)  # 显示内存分配最多的前 10 个位置

怎么解决循环引用

方案 1:手动打破循环

在不需要使用时,手动将引用设为 None 或从容器中移除:

python
class Node:
    def __init__(self):
        self.next = None
        self.prev = None

    def destroy(self):
        """手动打破循环引用"""
        self.next = None
        self.prev = None

# 使用完后手动清理
a.destroy()
b.destroy()
del a
del b

方案 2:使用弱引用(weakref)

弱引用不会增加对象的引用计数,因此不会阻止对象被回收。这是最推荐的方案。

python
import weakref

class TreeNode:
    def __init__(self, parent=None):
        self.parent = weakref.ref(parent) if parent else None  # 父引用用弱引用
        self.children = []  # 子引用用强引用
        if parent:
            parent.children.append(self)

# 这样父子关系中,父→子是强引用,子→父是弱引用,不会形成循环引用

方案 3:依赖标记-清除算法自动回收

Python 的 GC(标记-清除)能够回收大部分循环引用的对象。如果对象不涉及 __del__ 方法,循环引用通常会被自动回收,不会造成内存泄漏。

python
import gc

# 创建循环引用
a = [1]
b = [2]
a.append(b)
b.append(a)
del a
del b

print(gc.collect())  # 手动触发 GC,返回回收的对象数(至少 2 个)

注意:标记-清除不是实时的,有一定延迟,而且会暂停程序。

方案 4:避免使用 __del__ 方法

__del__ 方法的对象如果形成循环引用,GC 无法确定调用顺序,会被放入 gc.garbage 列表中永远不被回收。

python
class Leaky:
    def __del__(self):
        print("被销毁了")

a = Leaky()
b = Leaky()
a.other = b
b.other = a
del a
del b

gc.collect()
print(gc.garbage)  # [<__main__.Leaky object at ...>, <__main__.Leaky object at ...>]
# 这些对象永远不会被回收!

弱引用详解

python
import weakref

class Foo:
    def __init__(self, name):
        self.name = name
    def __repr__(self):
        return f"Foo({self.name})"

# 1. weakref.ref:基础弱引用
obj = Foo("test")
ref = weakref.ref(obj)
print(ref())       # Foo(test) —— 通过 ref() 获取原对象
del obj
print(ref())       # None —— 对象已被回收,返回 None

# 2. WeakValueDictionary:值是弱引用的字典
cache = weakref.WeakValueDictionary()
obj = Foo("cached")
cache["key"] = obj
print(cache["key"])  # Foo(cached)
del obj
print(cache.get("key"))  # None —— 自动移除

# 3. WeakKeyDictionary:键是弱引用的字典
# 4. WeakSet:元素是弱引用的集合

哪些对象不能被弱引用?

  • 内建的不可变小对象:intstrfloattuplefrozenset(因为驻留/缓存机制)
  • 不是所有对象都支持弱引用,需要类支持(默认支持,但某些类如 listdict 不支持)

总结对比

方案适用场景优点缺点
手动打破循环生命周期明确的对象简单直接容易忘记,不优雅
弱引用父子关系、观察者、缓存优雅、自动处理需要设计时考虑
依赖 GC 自动回收__del__ 的循环引用不用改代码有延迟、有暂停开销
避免 __del__所有场景从根源避免问题减少了析构能力

追问延伸

  • __del__ 的循环引用为什么不能被回收?(GC 不知道先调用哪个的 __del__,怕破坏对方后再调用会出错)
  • weakref.refweakref.proxy 有什么区别?
  • 如何判断内存泄漏是循环引用导致的还是全局变量累积导致的?
  • Django 的 ORM 查询会不会产生循环引用?(QuerySet 和模型实例之间可能有,但会被 GC 回收)
  • 你在项目中遇到过内存泄漏吗?怎么排查的?最终原因是什么?
  • __del__ 和上下文管理器(with)各适合什么场景?为什么更推荐上下文管理器?

Q11: Python的内存池机制?pymalloc? 「🟡 中级」

考察点:考察对 Python 内存管理底层的理解深度。内存池是 Python 性能优化的重要手段,也是区别初级和中级开发者的知识点。

参考答案

为什么需要内存池

Python 作为高级语言,频繁创建和销毁小对象(如 int、list 节点、dict 条目等)。如果每次都直接调用系统的 malloc/free,会有两个问题:

  1. 性能开销大:系统调用(malloc/free)有开销,频繁调用效率低
  2. 内存碎片多:大量小对象的分配释放会产生很多外部碎片,降低内存利用率

pymalloc 就是 Python 为了解决小对象分配问题而实现的内存池机制。

pymalloc 的内存层次结构

Arena(256 KB)
  └── Pool(4 KB = 1 个内存页)
        └── Block(固定大小,8~512 字节)

三层结构详解:

层级大小说明
Arena256 KB最大单位,通过 malloc 向操作系统申请
Pool4 KB中等单位,等于一个内存页(page)
Block8~512 字节最小单位,实际分配给用户的内存块

Arena(竞技场):

  • 大小固定为 256 KB
  • 一个 Arena 包含 64 个 Pool(256KB / 4KB = 64)
  • 多个 Arena 组成链表管理

Pool(内存池):

  • 大小为 4 KB(一个系统内存页)
  • 每个 Pool 只分配固定大小的 Block
  • 大小分级:8, 16, 24, 32, ..., 512 字节(共 64 个级别)
  • 相同大小级别的 Pool 组成链表(usedpools)

Block(内存块):

  • 实际分配给用户的内存单元
  • 大小是 8 字节的倍数(8, 16, 24, ..., 512)
  • 一个 Pool 内的所有 Block 大小相同

分配策略

小对象(≤512 字节)→ pymalloc 分配

  • 计算请求大小对应的 size class(向上对齐到 8 的倍数)
  • 在对应 size class 的 Pool 链表中找可用 Pool
  • 找到后从 Pool 中分配一个 Block
  • 如果没有可用 Pool,创建新的 Pool(从 Arena 中取或新建 Arena)

大对象(>512 字节)→ 直接 malloc

  • 超过 512 字节的分配直接走系统 malloc
  • 不受 pymalloc 管理
python
import sys

# 小对象走 pymalloc
a = [1, 2, 3]  # 列表对象本身(list结构体)走 pymalloc

# 大对象走 malloc
big_list = [0] * 100000  # 列表内部的元素数组可能走 malloc

内存池的好处

好处说明
减少系统调用小对象从内存池分配,不用每次都 malloc
减少内存碎片相同大小的 Block 集中管理,碎片少
分配速度快内存池中有空闲 Block 时,几乎 O(1) 分配
缓存友好小对象在 Pool 中连续分配,缓存局部性好

slots:节省实例内存

Python 的实例默认有一个 __dict__ 字典来存储属性,字典的开销很大(每个实例一个字典)。

使用 __slots__ 可以限制实例的属性,不使用 __dict__,从而大幅节省内存。

python
class Point:
    __slots__ = ("x", "y")  # 声明允许的属性
    def __init__(self, x, y):
        self.x = x
        self.y = y

import sys

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

p1 = Regular(1, 2)
p2 = Point(1, 2)

print(sys.getsizeof(p1))  # 约 48 字节(实例本身)+ 字典的额外开销
print(sys.getsizeof(p2))  # 约 40 字节(没有 __dict__)

slots 的优点:

  • 节省内存(属性越多,节省越明显)
  • 属性访问更快(直接查偏移量,不用哈希查找)
  • 防止意外添加属性(一定程度上的封装)

slots 的限制:

  • 不能动态添加属性
  • 不能使用 __dict__(除非在 __slots__ 中加入 __dict__
  • 不能使用 __weakref__(除非在 __slots__ 中加入 __weakref__
  • 多继承时,多个父类都有非空 __slots__ 会有冲突

内存管理全景图

应用程序

Python 对象分配器(PyObject_New 等)

pymalloc(小对象内存池,≤512字节)
    ↓ 或 大对象直接调用
C 层 malloc/free(系统内存分配)

操作系统内存管理

追问延伸

  • 小整数池(-5 到 256)和 pymalloc 是什么关系?(小整数池是对象级缓存,pymalloc 是内存级分配器)
  • 字符串驻留(intern)机制了解吗?和内存池有什么区别?
  • Python 3 中整数对象的内存是怎么管理的?(PyLongObject,小整数缓存,大对象 malloc)
  • 列表的扩容机制是什么?(over-allocation,预留空间,减少 realloc)
  • 字典的扩容机制和列表有什么不同?(负载因子 2/3,扩容时 rehash)
  • 什么是内存碎片?内部碎片和外部碎片有什么区别?
  • sys.intern() 函数有什么用?什么场景下用?

Q12: 什么是弱引用?应用场景? 「🟡 中级」

考察点:考察对 Python 引用模型和内存管理的深入理解,以及实际工程中处理循环引用、缓存等问题的能力。筛掉只知道强引用不知道弱引用的开发者。

参考答案

什么是弱引用

弱引用(Weak Reference)是一种不增加对象引用计数的引用方式,因此不会阻止对象被垃圾回收

当对象的强引用计数归零时,即使还有弱引用指向它,对象也会被回收,此时弱引用自动失效(返回 None)。

python
import weakref

class Foo:
    def __init__(self, name):
        self.name = name
    def __repr__(self):
        return f"Foo({self.name})"
    def __del__(self):
        print(f"{self.name} 被销毁了")

# 强引用
obj = Foo("test")

# 弱引用
ref = weakref.ref(obj)
print(ref())       # Foo(test) —— 对象还活着,可以通过 ref() 访问

# 删除强引用
del obj            # test 被销毁了
print(ref())       # None —— 对象已被回收,弱引用返回 None

弱引用的类型

类型说明
weakref.ref基础弱引用,通过 ref() 访问对象
weakref.proxy代理弱引用,使用更像原对象(但访问时可能抛 ReferenceError)
weakref.WeakKeyDictionary键是弱引用的字典
weakref.WeakValueDictionary值是弱引用的字典
weakref.WeakSet元素是弱引用的集合
weakref.finalize对象销毁时的回调(替代 __del__ 的更好选择)

ref vs proxy:

python
obj = Foo("test")
ref = weakref.ref(obj)
proxy = weakref.proxy(obj)

# ref 需要调用
print(ref().name)   # test

# proxy 可以直接像原对象一样使用
print(proxy.name)   # test

# 对象销毁后
del obj
print(ref())        # None
print(proxy.name)   # ReferenceError: weakly-referenced object no longer exists

WeakValueDictionary(缓存常用):

python
cache = weakref.WeakValueDictionary()

def get_data(key):
    if key in cache:
        print("缓存命中")
        return cache[key]
    # 从数据库加载
    data = load_from_db(key)
    cache[key] = data
    return data

# 当外部没有强引用指向 data 时,cache 中的条目自动被移除
# 不会因为缓存而导致对象永远不被回收

finalize(比 del 更好的析构方式):

python
class Resource:
    def __init__(self, name):
        self.name = name
        weakref.finalize(self, self._cleanup, name)

    @staticmethod
    def _cleanup(name):
        print(f"清理资源: {name}")

r = Resource("file.txt")
del r  # 清理资源: file.txt

finalize 的优势:不会导致循环引用问题,且可以传入参数、可以取消注册。

应用场景

1. 缓存(Cache)

最常见的场景。普通字典做缓存会导致对象永远不会被回收(因为字典持有强引用),用 WeakValueDictionary 可以在内存不足时自动释放缓存。

python
class ImageCache:
    def __init__(self):
        self._cache = weakref.WeakValueDictionary()

    def get(self, path):
        if path not in self._cache:
            self._cache[path] = load_image(path)
        return self._cache[path]

2. 避免循环引用

双向引用、父子关系、观察者模式等场景下,将其中一方改为弱引用,打破循环。

python
class Parent:
    def __init__(self):
        self.children = []  # 父→子:强引用

class Child:
    def __init__(self, parent):
        self._parent_ref = weakref.ref(parent)  # 子→父:弱引用

    @property
    def parent(self):
        return self._parent_ref()

3. 观察者模式(Observer Pattern)

python
class EventManager:
    def __init__(self):
        self._listeners = weakref.WeakSet()  # 弱引用集合

    def subscribe(self, listener):
        self._listeners.add(listener)

    def notify(self, event):
        for listener in self._listeners:
            listener.on_event(event)

# 监听器被销毁后,自动从 _listeners 中移除,不会内存泄漏

4. 大对象的临时引用

处理大对象时,如果只想"观察"它而不想阻止它被回收,可以用弱引用。

限制和注意事项

1. 不是所有对象都支持弱引用

python
# 支持:自定义类、list子类、dict子类等
# 不支持:int、str、float、tuple、frozenset、list、dict 等内置类型
# 原因:内置小对象有驻留/缓存机制,生命周期不确定

weakref.ref(42)        # TypeError: cannot create weak reference to 'int' object
weakref.ref("hello")   # TypeError: cannot create weak reference to 'str' object
weakref.ref([1,2,3])   # TypeError: cannot create weak reference to 'list' object

# 但自定义类的实例可以
class MyList(list):
    pass

weakref.ref(MyList())  # OK

2. 弱引用本身也是对象,也有开销

弱引用对象本身需要内存存储,大量弱引用也有开销。

3. 线程安全

weakref 模块的操作是线程安全的(因为操作引用计数受 GIL 保护)。

强引用 vs 弱引用对比

特性强引用弱引用
增加引用计数
阻止对象被回收
普通赋值强引用
容器存放强引用WeakValueDictionary 等
适用场景正常使用缓存、观察者、打破循环引用

追问延伸

  • weakref.ref 创建的弱引用对象自己什么时候被回收?(当所有强引用和弱引用都没有时)
  • WeakKeyDictionaryWeakValueDictionary 各适合什么场景?
  • Python 的 @cached_property 和弱引用有什么关系?(没有直接关系,它用的是非数据描述符 + 实例字典缓存)
  • 什么是"幽灵引用"(Phantom Reference)?Python 有吗?(没有,Java 才有,用于更细粒度的回收控制)
  • 你在项目中用过弱引用吗?解决了什么问题?
  • 循环引用 + 弱引用的组合,和"自动垃圾回收"相比各有什么优劣?

Q13: Python中的GIL是什么?对程序性能有什么影响? 「🟡 中级」

考察点:考察对 Python 并发编程核心限制的理解。GIL 是 Python 最有名的"梗"之一,也是后端面试必考题。筛掉对 Python 并发模型一知半解的开发者。

参考答案

什么是 GIL

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

注意:GIL 是 CPython 的实现细节,不是 Python 语言的特性。Jython、IronPython 没有 GIL,PyPy 有 GIL 但因为有 JIT 所以影响较小。

为什么要有 GIL

1. 简化内存管理(引用计数)

Python 使用引用计数管理内存,引用计数的增减操作需要保证线程安全。如果没有 GIL,就需要给每个对象加锁,会导致:

  • 锁太多,死锁风险高
  • 性能开销大(频繁加解锁)

GIL 用一把大锁简化了问题。

2. C 扩展集成简单

很多 C 扩展不是线程安全的。有了 GIL,C 扩展可以假设自己在单线程环境中运行,简化了开发。

GIL 对性能的影响

CPU 密集型任务:多线程反而更慢

python
# CPU 密集型:计算密集任务
import threading
import time

def cpu_bound(n):
    result = 0
    for i in range(n):
        result += i ** 2
    return result

# 单线程
start = time.time()
cpu_bound(10_000_000)
cpu_bound(10_000_000)
print(f"单线程: {time.time() - start:.2f}秒")

# 多线程
start = time.time()
t1 = threading.Thread(target=cpu_bound, args=(10_000_000,))
t2 = threading.Thread(target=cpu_bound, args=(10_000_000,))
t1.start()
t2.start()
t1.join()
t2.join()
print(f"多线程: {time.time() - start:.2f}秒")  # 比单线程更慢!

原因:GIL 的存在使得多线程无法真正并行执行 CPU 密集任务,反而因为线程切换和锁竞争增加了开销。

IO 密集型任务:多线程有效

python
# IO 密集型:网络请求、文件读写等
import threading
import time
import requests

def fetch(url):
    return requests.get(url)

urls = ["https://example.com"] * 10

# 单线程:顺序请求
start = time.time()
for url in urls:
    fetch(url)
print(f"单线程: {time.time() - start:.2f}秒")

# 多线程:并发请求
start = time.time()
threads = [threading.Thread(target=fetch, args=(url,)) for url in urls]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"多线程: {time.time() - start:.2f}秒")  # 快很多!

原因:IO 等待时,线程会释放 GIL,让其他线程执行。所以 IO 密集型任务用多线程是有效的。

GIL 的释放时机

GIL 不是一直持有不放的,以下情况会释放 GIL:

时机说明
IO 操作文件读写、网络请求等阻塞操作
time.sleep()睡眠时主动释放
字节码计数每执行一定数量的字节码后主动让出(Python 3.2 前是 100 条)
时间片轮转Python 3.2+ 改用时间片机制(默认 5ms)
C 扩展中手动释放C 扩展可以在计算密集部分手动释放 GIL(如 numpy)

Python 3.2 的改进:

  • 之前:每 100 条字节码释放一次 GIL
  • 之后:基于时间片(默认 5ms),线程在持有 GIL 超过时间片后主动释放
  • 效果:减少了线程切换的频率,提高了 CPU 密集型多线程的性能(虽然还是比单线程慢)

不同类型任务的并发方案选择

任务类型推荐方案不推荐原因
CPU 密集型多进程、C扩展多线程GIL 限制,多线程无法并行
IO 密集型多线程、协程(asyncio)多进程进程切换开销大,没必要
混合类型多进程 + 多线程/协程按瓶颈选择

常见误区

  1. "Python 不能多线程"——错。Python 完全支持多线程,只是多线程不能并行执行 CPU 密集任务。IO 密集型多线程是有效的。

  2. "Python 不能利用多核"——错。用多进程可以利用多核,每个进程有独立的 GIL。

  3. "GIL 永远是性能瓶颈"——错。IO 密集型程序不受 GIL 影响。只有 CPU 密集型才会受影响。

  4. "所有 Python 解释器都有 GIL"——错。Jython、IronPython 没有 GIL。

追问延伸

  • 为什么不直接移除 GIL?(历史包袱、C 扩展依赖、单线程性能会下降)
  • Python 3.13 的 nogil(PEP 703)了解吗?进展如何?
  • 多进程和多线程在 Python 中有什么区别?各适合什么场景?
  • multiprocessing 模块和 threading 模块的 API 有什么异同?
  • numpy 为什么在多线程下能加速?(C 扩展中释放了 GIL)
  • 协程(asyncio)和 GIL 有什么关系?(单线程内的并发,完全不涉及 GIL 竞争)
  • 你做过 Python 性能优化吗?怎么判断是 GIL 导致的性能问题?

Q14: 怎么绕过GIL?有哪些方法? 「🔴 高级」

考察点:考察对 Python 性能优化方案的系统性理解,以及实际处理 CPU 密集型任务的经验。能区分中级和高级工程师。

参考答案

绕过 GIL 的核心思路:要么不用单解释器内的多线程,要么在 C 层面释放 GIL

方案一:多进程(最常用)

每个进程有独立的 Python 解释器和独立的 GIL,因此多进程可以真正并行利用多核 CPU。

python
from multiprocessing import Process, Pool, cpu_count
from concurrent.futures import ProcessPoolExecutor
import time

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

# 方式 1:Process 类
if __name__ == "__main__":
    p1 = Process(target=cpu_bound, args=(5_000_000,))
    p2 = Process(target=cpu_bound, args=(5_000_000,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

# 方式 2:进程池
if __name__ == "__main__":
    with Pool(cpu_count()) as pool:
        results = pool.map(cpu_bound, [5_000_000] * 8)

# 方式 3:ProcessPoolExecutor(推荐)
if __name__ == "__main__":
    with ProcessPoolExecutor() as executor:
        results = list(executor.map(cpu_bound, [5_000_000] * 8))

优点:

  • 真正的并行,能利用多核
  • 进程隔离,一个进程崩溃不影响其他
  • API 相对简单,与 threading 类似

缺点:

  • 进程间通信(IPC)开销大(数据需要序列化/反序列化)
  • 创建进程开销比线程大
  • 共享数据复杂(需要用 Queue、Pipe、Manager、SharedMemory 等)
  • 进程数不能太多(受 CPU 核数限制)

适用场景: CPU 密集型任务、计算密集型批处理。

方案二:C 扩展释放 GIL

将计算密集的部分用 C/C++ 编写,在 C 代码中手动释放 GIL。

c
// C 扩展中释放 GIL 的示例
#include "Python.h"

static PyObject* compute(PyObject* self, PyObject* args) {
    int n;
    if (!PyArg_ParseTuple(args, "i", &n)) {
        return NULL;
    }

    // 释放 GIL,进入无 GIL 区域
    Py_BEGIN_ALLOW_THREADS

    // 这里是纯 C 计算,不操作 Python 对象,可以并行执行
    long long result = 0;
    for (int i = 0; i < n; i++) {
        result += (long long)i * i;
    }

    // 重新获取 GIL
    Py_END_ALLOW_THREADS

    return PyLong_FromLongLong(result);
}

实际中更常用的方式:

  • Cython:编写类 Python 代码,编译为 C 扩展,用 with nogil: 释放 GIL
  • ctypes / cffi:调用已有的 C 动态库,调用期间会释放 GIL
  • NumPy / SciPy:底层用 C/Fortran 实现,运算时释放 GIL
  • Numba:JIT 编译 Python 代码,@njit 模式下释放 GIL

Numba 示例:

python
from numba import njit
import numpy as np

@njit  # 自动编译为机器码,释放 GIL
def fast_sum(arr):
    total = 0.0
    for x in arr:
        total += x
    return total

arr = np.random.rand(10_000_000)
fast_sum(arr)  # 比纯 Python 快几十到几百倍

优点:

  • 性能最高,接近 C 语言速度
  • 多线程下能真正并行

缺点:

  • 开发成本高,需要懂 C/C++ 或 Cython
  • 调试困难

适用场景: 核心计算逻辑、性能瓶颈代码。

方案三:使用无 GIL 的 Python 解释器

解释器有无 GIL特点
CPython最主流,生态最好
Jython运行在 JVM 上,Python 2.x,生态差
IronPython运行在 .NET 上,Python 2.x,生态差
PyPy有(但有 JIT)JIT 编译,速度快,但 C 扩展兼容差
CPython 3.13+可选移除(实验性)PEP 703,nogil 构建

PEP 703(Making the Global Interpreter Lock Optional in CPython):

  • Python 3.13 开始提供实验性的 --disable-gil 构建选项
  • 目标是长期移除 GIL
  • 采用"biased reference counting"(偏向引用计数)等技术
  • 短期(3.13-3.15):实验阶段,性能可能回退
  • 长期:可能成为默认选项

方案四:协程(单线程并发)

协程不是"绕过 GIL",而是"绕开 GIL 问题"——在单线程内实现高并发,完全没有 GIL 竞争。

python
import asyncio
import aiohttp

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.text()

async def main():
    urls = ["https://example.com"] * 100
    tasks = [fetch(url) for url in urls]
    results = await asyncio.gather(*tasks)

asyncio.run(main())

适用场景: IO 密集型高并发(爬虫、API 网关、实时通信等)。

方案对比总结

方案原理性能开发成本适用场景
多进程每个进程独立 GIL高(线性扩展到核数)CPU 密集型
C/Cython 扩展C 层面释放 GIL最高(接近原生)核心计算瓶颈
NumbaJIT 编译 + 释放 GIL很高低(加装饰器)数值计算
无 GIL 解释器移除 GIL中(多线程快但单线程可能慢)未来方案
协程单线程并发,无 GIL 竞争IO 场景极高中低IO 密集型
numpy 等库内部 C 实现释放 GIL数值、矩阵运算

选型建议

有性能问题?
├── 是 IO 瓶颈? → 用协程(asyncio)或多线程
└── 是 CPU 瓶颈?
    ├── 能用 numpy/pandas 解决? → 用(内部已释放 GIL)
    ├── 是数值计算? → 用 Numba
    ├── 逻辑通用、任务独立? → 用多进程
    └── 性能要求极高且核心代码量小? → Cython / C 扩展

追问延伸

  • 多进程之间如何共享数据?(Queue、Pipe、Manager、SharedMemory、mmap)
  • 多进程的序列化(pickle)有什么限制?哪些对象不能 pickle?
  • multiprocessing 的 spawn/fork/forkserver 三种启动方式有什么区别?
  • Cython 的 with nogil: 有什么限制?(不能操作 Python 对象)
  • Numba 的 JIT 和 PyPy 的 JIT 有什么区别?
  • 你做过的最有挑战性的 Python 性能优化是什么?从多少优化到多少?
  • 什么是"全局解释器锁"的"全局"?是全局唯一还是每个解释器一个?(每个解释器一个,所以多进程能绕过)