Python3 基础教程

Python3 内存与性能底层:GIL、深浅拷贝与垃圾回收

🎉摘要:深入解析Python内存管理三大陷阱:GIL如何限制多线程并行、copy.copy与copy.deepcopy的细微差别导致数据篡改、引用计数与循环垃圾回收机制。通过可运行代码展示,让你真正理解Python内存行为,避免踩坑,提升性能。

Python 以 "简洁优雅" 著称,但优雅的背后隐藏着不少与内存和性能相关的陷阱。GIL 让多线程的 CPU 并行成为泡影,copy.copy 和 copy.deepcopy 的细微差别可能导致数据被意外篡改,而引用计数加循环检测的垃圾回收机制又让 "对象何时被销毁" 变得扑朔迷离。

本章把这三大主题摊开来讲,每个都配上可验证的代码,让你真正理解 Python 在内存层面的行为。

GIL:全局解释器锁,Python 多线程的命门

什么是 GIL?

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?

GIL 不是设计缺陷,而是历史权衡。CPython 的内存管理基于引用计数,多个线程同时修改对象的引用计数会导致数据竞态。加一把全局锁是最简单的线程安全方案。没有 GIL,要么重写整个内存管理系统(像 Jython 或 IronPython 那样),要么给每个对象加锁(性能暴跌)。

注意:在 C 扩展层面(如 NumPy 的矩阵运算),代码可以释放 GIL,此时多线程能真正并行。但纯 Python 代码永远受 GIL 约束。

IO 密集型:多线程仍然有效

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

如果你要进行 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。

count += 1 不是原子操作

这是多线程编程中最隐蔽的 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 中被编译为多条字节码:

  1. LOAD_GLOBAL counter  — 读取全局变量

  2. LOAD_CONST 1         — 加载常量 1

  3. INPLACE_ADD          — 相加

  4. 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)是原子的,复合表达式永远不是。

深浅拷贝:copy.copy 与 copy.deepcopy

浅拷贝表示创建一个新容器对象,但容器内部的元素只是复制引用,不递归拷贝内层对象。修改嵌套内容,两边同步变化。

深拷贝表示递归拷贝所有层级对象,完全独立副本。新旧对象完全隔离,修改任意一方互不影响。

简单说,浅拷贝仅复制最外层容器,嵌套对象共用;深拷贝递归复制所有层级,得到完全独立的副本。

赋值、浅拷贝与深拷贝的区别

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)  # True

垃圾回收:引用计数与循环检测

引用计数:Python 的主要 GC 机制

CPython 使用引用计数(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,引用计数机制无法回收它们。

gc 模块:分代垃圾回收

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

普通强引用会增加对象引用计数,而弱引用(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 bytes

内存池与对象复用

CPython 对小整数(-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 在某些情况下成立,但不应该依赖这个行为。

说说我的看法
全部评论(
没有评论
关于
本网站专注于 Java、数据库(MySQL、Oracle)、Linux、软件架构及大数据等多领域技术知识分享。涵盖丰富的原创与精选技术文章,助力技术传播与交流。无论是技术新手渴望入门,还是资深开发者寻求进阶,这里都能为您提供深度见解与实用经验,让复杂编码变得轻松易懂,携手共赴技术提升新高度。如有侵权,请来信告知:hxstrive@outlook.com
其他应用
公众号