Python的 GIL

背景

一个进程可以有好几个线程。这些线程会共享进程内的所有内存。比如可以读写同样的变量。

当一个进程有不止一个线程的时候，就会出现一个情况叫racing。因为这些进程中的线程有可能同时运行，也有可能交替运行。但不管是同时运行还是交替运行，都没有办法控制他们之间的相对顺序。

a=1
if a>0:
    a-=1

当俩个程序同时执行如上代码时就会出现a被减了俩次。发生了多线程竞争问题。

python使用的垃圾收集算法是引用计数。如果一个进程有多个线程在运行的话就会有竞争问题。因为--操作实际上是进行了多个步骤。就可能会使得计数错误。就没有办法保证每一个python对象被正确释放了。最常用的解决方法是加锁，每次只有一个人能进入代码块运行。因此当时python的设计者们决定给python设计一个全局锁，也就是GIL。python通过这种机制保证每个程序在执行的时候都是拿到线程锁的。也就是说没有bytecode能被其他线程打断，因此在bytecode中运行的每一个c程序都是线程安全的。

这种全局锁的好处是，简单，避免死锁，对于多线程的程序或者是没法并行多线程的程序这种全局锁的性能是非常优秀的。因为要锁是非常花时间的场景，全局锁就保证了每一个字节码运行的时候，最多只要一次锁，如果每个object都要访问很多的锁，那可能需要很多的拿锁时间。这也让给python写c扩展的时候方便了很多，因为可以确定在每一个bytecode运行的时候，没有线程竞争冒险的问题。这样在c代码中修改python对象的时候就不用管乱七八糟的锁。

许多人尝试从python中拿走GIL，但是每次尝试都失败了，特别是先前兼容的问题。

其实python比java还要大。是上个世纪90年代初的语言（早java4年）当时多核根本不存在，多线程存在的意义就是这些线程可以轮流执行，不会因为某个线程计算量过大，而卡住其他的事情。但是在21世纪后多核已经是电脑的标配。因此现在GIL出现了严重的水土不服。python的多线程在这种情况下没有办法利用多核来增加自己的计算速度。

python如今使用多进程来解决这个问题。虽然一个进程没有办法用多个cpu，但是我可以有很多个进程。

from multiprocessing import Pool

def f(x):
    return x * x

if __name__ == '__main__':
    with Pool(5) as p:
        print(p.map(f, [1, 2, 3]))

或者写c程序 在c语言中做多线程。当然理论上也可以尝试用一些python中没有GIL的解释器。jython就是没有GIL的。

线程竞争

import threading

counter = 0  # 全局变量

def increment():
    global counter
    for _ in range(100000):
        temp = counter  # 1. 读取当前值
        temp = temp + 1  # 2. 增加值
        counter = temp   # 3. 写回

# 创建两个线程
t1 = threading.Thread(target=increment)
t2 = threading.Thread(target=increment)

t1.start()
t2.start()
t1.join()
t2.join()

print(f"Final counter: {counter}")  # 期望是 200000，但实际通常小于 200000

在python3.11中运行代码，上述代码的值就是200w，但是如果用3.9来运行这个程序就会发现这个值不是200w，竞争冒险就出现了

  1. 线程A执行完temp=counter后可能释放GIL（比如时间片到了）
  2. 线程B获得GIL，执行完整的counter+=1
  3. 线程 A 重新获得 GIL，继续从之前的状态执行，覆盖了线程 B 的更新

GIL 只保证字节码执行的原子性，但不保证多行字节码操作的原子性！

但是当前这个情况是一个行为而不是一个特性，所以任何人都不应该这样写代码。在未来的python版本中随时有可能会变化。

要安全地访问共享的全局变量，你需要显式地使用线程锁：

import threading

counter = 0
lock = threading.Lock()  # 创建一个同步锁

def increment():
    global counter
    for _ in range(100000):
        with lock:  # 获取锁
            counter += 1
        # 离开 with 块时自动释放锁

# 或者显式地：
def increment_explicit():
    global counter
    for _ in range(100000):
        lock.acquire()  # 获取锁
        try:
            counter += 1
        finally:
            lock.release()  # 确保释放锁

同时执行一条字节码≠现成安全

那么python的GIL真的一无是处吗？对于在执行过程中频繁需要等待的IO密集型任务，多线程还是有一定优势的。当线程碰到IO操作时，线程会自动释放锁，让其任务函数执行，避免CPU资源浪费。IO密集型任务就是要进行频繁等待的任务。

如果是CPU密集型任务建议使用多进程，因为多进程不受全局解释器锁的影响。

python未来

python3.14取消了GIL，但是者是一个可选的特性。

可以通过 --disable-gil编译标志来构建无GIL的Python解释器，新增了 sys.set_gil_enabled()函数，可以在运行时动态控制GIL的启用状态

4核CPU上性能可达5倍提升，在CPU密集型任务中执行时间从27秒缩短到9秒。

但是单线程性能有5-10%的损失，增加了百分之15%的内存占用。

在自由线程模式下，Python 不再有 GIL 这把“万能大锁”来简单粗暴地保证线程安全，而是需要为许多内置对象（如字典、列表）引入更精细化的锁机制。每次访问或修改这些对象时，解释器都可能需要执行获取锁、检查锁状态、释放锁等操作。这些操作在GIL模式下是不需要的，因此带来了额外的开销

python的最终走向应该是最终只会出现一个no-gil的build，不会出现永久的俩个build。cpython给出的时间是五年以上。

FastAPI

GIL 的存在意味着在单进程内，Python 解释器任一时刻只能执行一个线程的字节码。这对于 CPU 密集型（计算密集型）任务是一个瓶颈。FastAPI 通过以下策略应对：

FASTAPI构建在Starlette之上，Starlette是一个高性能的ASGI框架

通过异步处理，fastAPI可以在I/O操作（如数据库查询、外部API请求等）等待期间释放线程，从而处理其他请求，从而提高并发和响应速度。

uvicorn 和fastapi是什么关系
Uvicorn 和 FastAPI 是两个独立但高度互补的 Python 工具，它们的关系可以概括为：FastAPI 是一个基于 ASGI 标准构建的 Web 框架，而 Uvicorn 是一个支持 ASGI 的 ASGI 服务器，用于运行 FastAPI 应用。

Async IO

async本质上还是python运行的单线程单进程程序，核心中是eventloop，在python中能执行的任务只有一个，不存在系统级的上下文切换。因此它有个好处是不存在竞争冒险的问题。

import asyncio
async def main():
    print("hello")
    await asyncio.sleep(1)
    print("world")
coro=main()
asyncio.run(coro)

corouting:

这个async def main就是一个corouting function，所有async开头的都是corouting函数，一般函数调用的时候返回的是这个函数的返回值。而corouting函数调用的时候，返回的是corouting的object。为了运行我们要进行俩个事情。

一个就是进入async模式。正常python的代码是在同步模式下运行的。eventloop建立之后就会去找哪个任务可以执行

import asyncio
import time

async def say_after(delay, what):
    await asyncio.sleep(delay)
    print(what)

async def main():
    print(f"started at {time.strftime('%X')}")

    await say_after(1, 'hello')
    await say_after(2, 'world')

    print(f"finished at {time.strftime('%X')}")

asyncio.run(main())

asyncio.sleep返回的也是corouting。这个corouting被包装成了task并且告诉eventloop。并且会告诉eventloop当前task无法执行了，可以先执行其他的task。

这里会在一秒后输出hello 再一秒后输出world。

await关键字必须在async定义的异步函数中使用

所有控制器的返回都是显示的。eventloop没办法主动拿回控制权，必须task主动交出控制器。

俩中交回方式：运行await会交回，函数运行完毕后会交回。如果一个task里面有死循环，那么整个eventloop就卡死了

第二个world需要等第一个hello执行完毕了之后才能继续执行。

为了这个问题，asyncio给我们提供了create_task函数。

import asyncio
import time

async def say_after(delay, what):
    await asyncio.sleep(delay)
    print(what)

async def main():
    task1 = asyncio.create_task(
        say_after(1, 'hello')
    )

    task2 = asyncio.create_task(
        say_after(2, 'world')
    )

    print(f"started at {time.strftime('%X')}")

    await task1
    await task2

    print(f"finished at {time.strftime('%X')}")

# 运行主函数
if __name__ == "__main__":
    asyncio.run(main())

其中create_task，这个函数的参数是一个corouting，它不会运行任何corouting中的代码。create_task会把这个corouting注册到异步编程里面。但是现在eventloop还没有办法执行这个task。因为task还在main的手里。当await后面是一个task而不是corouting的时候它就省略了把corouting变成task的步骤。因此async很适合解决网络通讯的问题，因为网络通讯很多时间是在等待上的，其他任务就可以干活。

但是如果有10个task呢？我们可以使用一个叫asyncio.gather的方法。最终执行哪个部分永远是由eventloop来决定的，task没办法告诉eventloop去执行等待的某个特定的task。如果代码中并没有等待这件事情的话，协程对程序加速是没有帮助的。拿到corouting的返回值要用await。

await

eventloop在调度的时候最小单位是task。eventloop没办法执行一个corouting

import asyncio

async def main():
    await asyncio.create_task(
        asyncio.sleep(0.1)
    )

asyncio.run(main())

asyncio.run会建立一个task。

代码中顶层的task是main

Last modification：December 6, 2025
© Allow specification reprint
如果觉得我的文章对你有用，请随意赞赏