Python 以 "简洁优雅" 著称,但优雅的背后隐藏着不少与内存和性能相关的陷阱。GIL 让多线程的 CPU 并行成为泡影,copy.copy 和 copy.deepcopy 的细微差别可能导致数据被意外篡改,而引用计数加循环检测的垃圾回收机制又让 "对象何时被销毁" 变得扑朔迷离。
本章把这三大主题摊开来讲,每个都配上可验证的代码,让你真正理解 Python 在内存层面的行为。
GIL(Global Interpreter Lock)是 CPython 解释器的一把互斥锁。它保证同一时刻只有一个线程在执行 Python 字节码。无论你有多少个 CPU 核心、开了多少个线程,真正在解释器里跑字节码的永远只有一个。例如:
import threading
import time
def cpu_bound_task(n: int) -> int:
"""纯 CPU 计算,没有 IO"""
count = 0
for i in range(n):
count += i * i
return count
# 单线程执行
start = time.time()
cpu_bound_task(10_000_000)
cpu_bound_task(10_000_000)
print(f"单线程: {time.time() - start:.2f}s")
# 多线程执行
start = time.time()
t1 = threading.Thread(target=cpu_bound_task, args=(10_000_000,))
t2 = threading.Thread(target=cpu_bound_task, args=(10_000_000,))
t1.start(); t2.start()
t1.join(); t2.join()
print(f"多线程: {time.time() - start:.2f}s")执行代码,输出如下:
单线程: 1.20s
多线程: 1.22s运行结果会让你失望:多线程和单线程的时间几乎一样,甚至多线程更慢。因为两个线程在抢 GIL,上下文切换反而增加了开销。
GIL 不是设计缺陷,而是历史权衡。CPython 的内存管理基于引用计数,多个线程同时修改对象的引用计数会导致数据竞态。加一把全局锁是最简单的线程安全方案。没有 GIL,要么重写整个内存管理系统(像 Jython 或 IronPython 那样),要么给每个对象加锁(性能暴跌)。
注意:在 C 扩展层面(如 NumPy 的矩阵运算),代码可以释放 GIL,此时多线程能真正并行。但纯 Python 代码永远受 GIL 约束。
GIL 在遇到 IO 操作(网络、文件、睡眠)时会主动释放,让其他线程有机会执行。所以 IO 密集型任务用多线程仍然能提升吞吐量:
import threading
import time
def io_task(url: str):
"""模拟网络请求"""
time.sleep(1) # sleep 会释放 GIL
print(f"Done: {url}")
urls = ["https://a.com", "https://b.com", "https://c.com", "https://d.com"]
# 单线程串行
start = time.time()
for url in urls:
io_task(url)
print(f"单线程: {time.time() - start:.2f}s")
# 多线程并发
start = time.time()
threads = [threading.Thread(target=io_task, args=(url,)) for url in urls]
for t in threads: t.start()
for t in threads: t.join()
print(f"多线程: {time.time() - start:.2f}s")运行代码,输出:
Done: https://a.com
Done: https://b.com
Done: https://c.com
Done: https://d.com
单线程: 4.00s
Done: https://a.comDone: https://d.com
Done: https://c.com
Done: https://b.com
多线程: 1.00s多线程版本大约 1 秒完成,单线程版本需要 4 秒。因为 time.sleep 让出 GIL 时,其他线程立刻接手。对于网络爬虫、数据库查询等场景,多线程仍然有效。
如果你要进行 CPU 密集型任务,可以使用多进程的方式绕过 GIL,例如:
# multiprocess_demo.py
import multiprocessing
import time
def cpu_task(n: int) -> int:
count = 0
for i in range(n):
count += i * i
return count
if __name__ == '__main__':
start = time.time()
# 多进程
with multiprocessing.Pool(processes=2) as pool:
pool.map(cpu_task, [10_000_000, 10_000_000])
print(f"多进程: {time.time() - start:.2f}s")运行代码,输出:
多进程: 1.47s多进程每个进程有独立的 Python 解释器和 GIL,可以真正利用多核。代价是进程间通信开销比线程大得多,共享数据需要用 multiprocessing.Queue 或 Manager。
这是多线程编程中最隐蔽的 bug:
import threading
counter = 0
def increment():
global counter
for _ in range(100000):
counter += 1
threads = [threading.Thread(target=increment) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
print(f"Expected 1000000, got {counter}")
# 大概率输出小于 1000000 的数counter += 1 在 CPython 中被编译为多条字节码:
LOAD_GLOBAL counter — 读取全局变量
LOAD_CONST 1 — 加载常量 1
INPLACE_ADD — 相加
STORE_GLOBAL counter — 写回全局变量
GIL 在这四条字节码之间可能切换线程。线程 A 读取到 100,还没写回,线程 B 也读取到 100,两者都写回 101,漏掉了一次增量。这就是为什么多线程共享计数器必须加锁:
import threading
# 次数
counter = 0
# 全局锁
lock = threading.Lock()
def safe_increment():
global counter
for _ in range(100000):
# 自动加锁和释放锁
with lock:
counter += 1
threads = [threading.Thread(target=safe_increment) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
print(f"Expected 1000000, got {counter}")注意:GIL 保证的是字节码执行的原子性,不是表达式执行的原子性。只有单条字节码指令(如 list.append)是原子的,复合表达式永远不是。
浅拷贝表示创建一个新容器对象,但容器内部的元素只是复制引用,不递归拷贝内层对象。修改嵌套内容,两边同步变化。
深拷贝表示递归拷贝所有层级对象,完全独立副本。新旧对象完全隔离,修改任意一方互不影响。
简单说,浅拷贝仅复制最外层容器,嵌套对象共用;深拷贝递归复制所有层级,得到完全独立的副本。
import copy
original = [[1, 2], [3, 4], 5]
# 赋值:引用同一个对象
assigned = original
print(f"赋值后 id(original[0]) == id(assigned[0]): {id(original[0]) == id(assigned[0])}")
# 浅拷贝:最外层新建,内层共享引用
shallow = copy.copy(original)
print(f"浅拷贝 id(original) == id(shallow): {id(original) == id(shallow)}") # False
print(f"浅拷贝 id(original[0]) == id(shallow[0]): {id(original[0]) == id(shallow[0])}") # True
# 深拷贝:完全独立,递归复制所有层级
deep = copy.deepcopy(original)
print(f"深拷贝 id(original[0]) == id(deep[0]): {id(original[0]) == id(deep[0])}") # False运行示例,输出:
赋值后 id(original[0]) == id(assigned[0]): True
浅拷贝 id(original) == id(shallow): False
浅拷贝 id(original[0]) == id(shallow[0]): True
深拷贝 id(original[0]) == id(deep[0]): False下表介绍赋值、浅拷贝、深拷贝的区别:
| 操作 | 最外层对象 | 内层对象 | 使用场景 |
| 赋值 (=) | 共享 | 共享 | 别名,同一份数据 |
| 浅拷贝 (copy.copy) | 新建 | 共享 | 列表/字典的顶层复制,内层不常修改 |
| 深拷贝 (copy.deepcopy) | 新建 | 递归新建 | 完全隔离,避免意外修改 |
浅拷贝只复制了外层容器的引用,内层对象仍然是共享的。这在处理嵌套数据结构时是巨大的隐患。例如:
import copy
users = [{"name": "Alice", "age": 30}, {"name": "Bob", "age": 25}]
# 复制后,列表前两个数据是共享的
users_copy = copy.copy(users)
# 修改外层列表:不影响原列表
# 新增的对象,原列表不存在
users_copy.append({"name": "Charlie", "age": 35})
print(f"原列表长度: {len(users)}") # 2
print(f"拷贝列表长度: {len(users_copy)}") # 3
# 修改内层字典:原列表也被改!
users_copy[0]["age"] = 31
print(f"原列表: {users[0]}") # {'name': 'Alice', 'age': 31} —— 被意外修改!
print(f"拷贝列表: {users_copy[0]}") # {'name': 'Alice', 'age': 31}运行代码,输出:
原列表长度: 2
拷贝列表长度: 3
原列表: {'name': 'Alice', 'age': 31}
拷贝列表: {'name': 'Alice', 'age': 31}使用上面同样的数据,但是使用 deepcopy 进行深度复制,例如:
import copy
users = [{"name": "Alice", "age": 30}, {"name": "Bob", "age": 25}]
users_deep = copy.deepcopy(users)
users_deep[0]["age"] = 31
print(f"原列表: {users[0]}") # {'name': 'Alice', 'age': 30} —— 安全!
print(f"深拷贝: {users_deep[0]}") # {'name': 'Alice', 'age': 31}运行代码,输出:
原列表: {'name': 'Alice', 'age': 30}
深拷贝: {'name': 'Alice', 'age': 31}deepcopy 会递归遍历对象的每个属性,遇到容器就新建,遇到普通对象就复制。对于自定义类,它会调用 __deepcopy__ 方法(如果定义了),否则默认行为是递归复制 __dict__。
我们可以通过实现 __deepcopy__() 方法自定义深度拷贝的行为,例如:
import copy
class Config:
def __init__(self, name: str, settings: dict):
self.name = name
self.settings = settings
self._cache = {} # 内部缓存,不需要深拷贝
def __deepcopy__(self, memo):
"""自定义深拷贝:不复制 _cache,只复制核心数据"""
new = copy.copy(self) # 浅拷贝,共享 _cache
new.settings = copy.deepcopy(self.settings, memo)
return new
def __repr__(self):
return f"Config({self.name!r}, {self.settings!r})"
cfg = Config("app", {"debug": True, "timeout": 30})
cfg_copy = copy.deepcopy(cfg)
cfg_copy.settings["debug"] = False
print(cfg) # Config('app', {'debug': True, 'timeout': 30})
print(cfg_copy) # Config('app', {'debug': False, 'timeout': 30})运行代码,输出:
Config('app', {'debug': True, 'timeout': 30})
Config('app', {'debug': False, 'timeout': 30})__deepcopy__ 接收 memo 字典,用于记录已复制的对象,防止循环引用导致无限递归。在自定义拷贝时,记得把新对象存入 memo[id(self)] = new,并在递归调用 deepcopy 时传递 memo。
对于字符串、元组、数字等不可变对象,copy 不会创建新对象,而是直接返回原对象引用。这是安全的,因为不可变对象不能被修改,共享没有风险。例如:
import copy
# 不可变对象:copy.copy 和 copy.deepcopy 都返回原对象
t = (1, 2, 3)
print(copy.copy(t) is t) # True
print(copy.deepcopy(t) is t) # True
s = "hello"
print(copy.copy(s) is s) # True
print(copy.deepcopy(s) is s) # TrueCPython 使用引用计数(Reference Counting)作为内存管理的主要手段。每个对象都有一个 ob_refcnt 字段,记录有多少个引用指向它。当引用计数降到 0 时,对象立即被销毁,内存被回收。
例如:
import sys
a = [1, 2, 3]
print(f"引用计数: {sys.getrefcount(a)}") # 输出 2:a 本身 + getrefcount 的参数
b = a
print(f"赋值后: {sys.getrefcount(a)}") # 输出 3
c = b
print(f"再赋值: {sys.getrefcount(a)}") # 输出 4
del b
print(f"del b 后: {sys.getrefcount(a)}") # 输出 3
del a, c
# 此时引用计数为 0,列表被销毁sys.getrefcount(obj) 返回的计数总是比实际多 1,因为传入 getrefcount 的参数本身也增加了一个临时引用。
引用计数有一个致命缺陷:无法处理循环引用。例如:
import gc
class Node:
def __init__(self, name: str):
self.name = name
self.next = None
def __del__(self):
print(f"[__del__] {self.name} 被销毁")
# 创建循环引用
a = Node("A")
b = Node("B")
a.next = b
b.next = a
# 手动断开外部引用
ref_a = a
ref_b = b
a = None
b = None
# 此时 A 和 B 互相引用,引用计数都是 1,但外部已经不可达
print("尝试手动 gc...")
gc.collect() # 强制垃圾回收
print("gc 完成")运行后你会发现 __del__ 没有被调用。因为 A 引用 B,B 引用 A,即使外部变量已经置为 None,两者的引用计数仍然是 1,引用计数机制无法回收它们。
CPython 的第二个垃圾回收机制是分代回收(Generational GC),专门处理循环引用。它通过 gc 模块提供:
import gc
# 查看当前垃圾回收配置
print(f"分代数量: {gc.get_threshold()}") # 默认 (700, 10, 10)
# 手动触发回收
count = gc.collect()
print(f"回收了 {count} 个不可达对象")
# 查看当前被循环引用困住的对象
for obj in gc.garbage:
print(f"垃圾对象: {obj}")运行示例,输出:
分代数量: (2000, 10, 10)
回收了 0 个不可达对象gc.collect() 的参数 generation 指定回收哪一代:
0 代:新创建的对象,检查最频繁;
1 代:经历过一次 0 代回收还存活的对象;
2 代:经历过多次回收的老对象,检查最稀疏。
gc.set_threshold(700, 10, 10) 的意思是:0 代对象超过 700 个时触发一次 0 代回收;0 代回收 10 次后触发一次 1 代回收;1 代回收 10 次后触发一次 2 代回收。
普通强引用会增加对象引用计数,而弱引用(weakref)只可以访问对象,不会增加引用计数。
只要对象不存在任何强引用,Python GC 就可以直接销毁它,不受弱引用影响。
例如:
import weakref
import sys
class DataCache:
def __init__(self, data: dict):
self.data = data
def __del__(self):
print(f"[__del__] DataCache 被销毁")
# 强引用:引用计数 +1
cache = DataCache({"key": "value"})
strong_ref = cache
print(f"强引用计数: {sys.getrefcount(cache) - 1}") # 2(cache + strong_ref)
# 弱引用:引用计数不变
weak_ref = weakref.ref(cache)
print(f"弱引用存活: {weak_ref() is not None}") # True
# 删除强引用后,弱引用自动失效
del cache
del strong_ref
# 此时 DataCache 被销毁,即使 weak_ref 还存在
print(f"弱引用存活: {weak_ref() is not None}") # False运行示例,输出:
强引用计数: 2
弱引用存活: True
[__del__] DataCache 被销毁
弱引用存活: False弱引用常用于缓存和观察者模式,避免对象因为被缓存而永不被回收。
循环引用指的是两个或多个对象之间互相形成引用指向的关系,对象 A 持有对象 B 的引用,同时对象 B 又反过来持有对象 A 的引用,在部分场景下还会出现更长的引用链条,比如对象 A 引用对象 B、对象 B 引用对象 C、对象 C 再回头引用对象 A。这类引用关系会出现在编程开发、数据建模、文档公式等不同场景当中,往往容易引发内存无法正常释放、逻辑报错、计算结果异常等各类问题。
例如:
import gc
# DEBUG_LEAK:打印可回收对象;注意会产生大量日志,调试结束务必关闭
gc.set_debug(gc.DEBUG_LEAK)
class Node:
def __init__(self, name):
self.name = name
self.next = None
def __repr__(self):
return f"Node({self.name!r})"
print("构建循环引用 a <-> b")
a = Node("A")
b = Node("B")
a.next = b
b.next = a
# 查看当前对象引用者
# get_referents:当前对象直接持有的下级对象
print("a 直接引用的对象:", gc.get_referents(a))
# get_referrers:指向a的上层引用(列表、变量、其他对象)
print("指向a的引用方:", gc.get_referrers(a)[:3])
# 解除全局变量 a、b 的外部引用
a = b = None
# 执行 gc.collect() 强制回收
recycle_num = gc.collect()
print(f"本次成功回收对象总数: {recycle_num}")
# 获取所有处于待回收状态的垃圾对象
garbage_objs = gc.garbage
print(f"\ngc.garbage 列表长度:{len(garbage_objs)}")
print("待清理垃圾对象列表:", garbage_objs)
# 关闭GC调试输出,防止持续刷屏
gc.set_debug(0)运行示例,输出:
构建循环引用 a <-> b
a 直接引用的对象: ['A', Node('B'), <class '__main__.Node'>]
指向a的引用方: [Node('B'), {'__name__': '__main__', '__doc__': None, '__package__': None, '__loader__': <_frozen_importlib_external.SourceFileLoader object at 0x00000270749F8D10>, '__spec__': None, '__annotations__': {}, '__builtins__': <module 'builtins' (built-in)>, '__file__': 'd:\\share_dir\\workspace\\5.demo\\python_demo\\demo.py', '__cached__': None, 'gc': <module 'gc' (built-in)>, 'Node': <class '__main__.Node'>, 'a': Node('A'), 'b': Node('B')}]
本次成功回收对象总数: 2
gc.garbage 列表长度:2
待清理垃圾对象列表: [Node('A'), Node('B')]
gc: collectable <Node 0x00000270747870E0>
gc: collectable <Node 0x00000270749FCB90>如果类定义 __del__ 析构方法,循环引用将产生内存泄漏,例如:
import gc
# 存在 __del__ 方法导致循环引用无法回收(内存泄漏)
gc.set_debug(gc.DEBUG_LEAK)
class BadNode:
def __init__(self, name):
self.name = name
self.next = None
def __del__(self):
# 存在析构方法,GC无法确定调用顺序,循环引用变成不可回收垃圾
print(f"调用 __del__ : BadNode({self.name})")
x = BadNode("X")
y = BadNode("Y")
x.next = y
y.next = x
x = y = None
count = gc.collect()
print(f"回收数量:{count}")
print("gc.garbage 内滞留无法释放的对象:", gc.garbage)
gc.set_debug(0)运行示例,输出:
调用 __del__ : BadNode(X)
调用 __del__ : BadNode(Y)
回收数量:2
gc.garbage内滞留无法释放的对象: [<__main__.BadNode object at 0x000001FAB09C70E0>, <__main__.BadNode object at 0x000001FAB0BACB90>]
gc: collectable <BadNode 0x000001FAB09C70E0>
gc: collectable <BadNode 0x000001FAB0BACB90>对于内存泄漏调试,可以用 gc.get_referrers(obj) 查看谁引用了某个对象,或者用 objgraph 第三方库画引用图。
使用 getsizeof() 方法获取对象内存。
注意:sys.getsizeof () 仅测量对象自身占用内存,不递归统计它引用的内部对象。只能作为粗略参考,不能得到整个容器完整内存占用。对于容器,总内存占用需要递归计算。
例如:
import sys
# 内置类型的大小
print(f"int 0: {sys.getsizeof(0)} bytes")
print(f"int 1000000: {sys.getsizeof(1_000_000)} bytes")
print(f"str '': {sys.getsizeof('')} bytes")
print(f"str 'hello': {sys.getsizeof('hello')} bytes")
print(f"list []: {sys.getsizeof([])} bytes")
print(f"list [1,2,3]: {sys.getsizeof([1,2,3])} bytes")
# 对象列表:注意列表本身只存引用,元素在别处
big_list = [0] * 1000
print(f"list of 1000 ints: {sys.getsizeof(big_list)} bytes")
# 结果约 8056 bytes,不是 1000 * 28 = 28000,因为列表存的是指针运行示例,输出:
int 0: 28 bytes
int 1000000: 28 bytes
str '': 41 bytes
str 'hello': 46 bytes
list []: 56 bytes
list [1,2,3]: 88 bytes
list of 1000 ints: 8056 bytesCPython 对小整数(-5 到 256)和短字符串有缓存池,频繁创建这些小对象时不会真的分配新内存。
示例:
a = 100
b = 100
print(a is b) # True,小整数来自缓存池
a = 1000
b = 1000
print(a is b) # False,大整数不在缓存池
# 字符串驻留(intern)
s1 = "hello_world"
s2 = "hello_world"
print(s1 is s2) # True,编译时合并的字符串
s1 = "hello " + "world"
s2 = "hello world"
print(s1 is s2) # 可能 False,运行时拼接的字符串不一定驻留运行示例,输出:
True
True
True
True注意:is 判断的是对象身份(内存地址是否相同),== 判断的是值是否相等。对于不可变对象,驻留优化让 is 在某些情况下成立,但不应该依赖这个行为。