让 Python 真正支持多线程

Python 诞生至今已经32年了！如今它是 TOBIE 编程语言排行榜 排名第一的语言，被广泛用于各种应用程序的开发。 然而遗憾的是——Python 至今都缺乏对多线程的原生支持。 好消息是 Python 3.12 将引入的“Per-Interpreter GIL”，彻底改变对多线程缺乏原生支持的情况。尽管距离 Python 3.12 的发布还有几个月的时间，但我已经通过最新的 CPython 代码实现了真正的多线程！

在本文中，我将深入探讨 Python 无法引入多线程的背后机制，以及如何使用子解释器 API 编写真正并发的 Python 代码。

什么是多线程

要想理解为什么 Python 不支持多线程，首先需要了解什么是多线程，以及它是如何工作的。

简而言之，多线程是一种同时执行多个任务操作的方法。每个线程独立于其他线程运行，并且能够同时运行软件程序的特定部分。在可以同时执行多个任务的情况下，这种方法对于提高程序的整体性能和响应能力尤其适用。

很多人对多线程和多进程混淆分不清，这里做一个简单对比;

多线程

• 在现有进程中产生一个新线程
• 启动线程比启动进程快
• 内存在所有线程之间共享
• 通常需要互斥锁来控制对共享数据的访问
• Python 中所有线程的一个 GIL（全局解释器锁）【这条非常重要，是导致 Python 不支持真正多线程的罪魁祸首】
• 线程并不能并行作业，而是在线程之间来回切换。

多进程

• 可以独立于其他进程启动一个新进程
• 启动进程比启动线程慢
• 进程之间不共享内存（与线程相比，启动进程时涉及更多内存，尽管子进程可以访问父进程的内存）
• 不需要互斥体
• Python 中每个进程一个 GIL（全局解释器锁）

为什么Python不支持多线程

Python 缺少多线程支持是由于全局解释器锁 (GIL)导致的。GIL 是一种保证一次只有一个线程可以执行 Python 字节码的机制。从本质上讲，这意味着即使 Python 进程中可以存在多个线程，它们也不能并发执行 Python 代码。 这是因为 GIL 将解释器限制为单个线程，这会阻止其他线程执行 Python 代码。

全局解释器锁的存在是因为 Python 采用了引用计数内存管理模型，当对象的引用计数为零时，它会删除对象。 GIL 保证即便是多个线程同时访问同一个对象的情况下，引用计数也是准确的。 如果没有 GIL ，就会出现竞争，两个线程可能会同时尝试修改同一对象的引用计数，从而导致诸多内存问题。

一个让我吃惊的事实是：很多 Python 程序员居然不知道 Python 不支持真正的多线程。可能大多数 Python 的应用场景不需要用到多线程。Python 曾经被设计成一种用户友好的和可读性强的高级语言。因此，Python 缺少许多其他编程语言中常见的复杂和低级元素。虽然它是一个强大的工具，但现在 Python 应用越来越广，越来越多的场景和用例需要用多线程来提高效率或优化体验，因此引入真正的多线程是 Python 必须要面对并解决的紧急问题。

让Python真正支持多线程

子解释器

“Per-Interpreter GIL” 的引入让 Python 可以摆脱 GIL，真正实现多线程。

简单地说，GIL 或 全局解释锁是一种互斥体，只允许一个线程控制 Python 解释器。 这意味着即使您在 Python 中创建多个线程（例如使用 threading 模块），一次也只会运行一个线程。

随着“Per-Interpreter GIL”的引入，各个 Python 解释器不再共享同一个 GIL。这种隔离级别允许每个子解释器真正并行。这意味着，我们可以通过创建新的子解释器来绕过 Python 的并发限制，其中每个子解释器都有自己的 GIL（全局状态）。

有关子解释器和 Per-Interpreter GIL 更深入的解释，请参阅 PEP 684 。

安装

要使用这项最前沿的功能，我们需要安装最新的 Python 解释器，由于 Python 3.12 尚未发布，我们从最新的源代码构建：

git clone https://github.com/python/cpython.git
cd cpython

./configure --enable-optimizations --prefix=$(pwd)/python-3.12
make -s -j2
./python

C-API

安装好最新版本的 Python 解释器后，如何创建子解释器呢？

这里可能不要上点科技与狠活，因为子解释器不像 Python 模块一样直接导入可用，而是需要调用 C-API 才能使用。

正如 PEP-684 中指出的那样：

“…这是一项高级功能，适用于 C-API 的一小部分用户。”

Per-Interpreter GIL 目前只能通过 C-API 使用，因此没有面向 Python 开发人员的直接接口。Python 接口预计将随 PEP 554 一起提供。PEP 554目前还在讨论中，并没有决议通过。如果通过的话，应该会出现在 Python 3.13 中。在那之前我们将不得不用科技与狠活来实现子解释器。

虽然这部分没有文档，也没有可以直接导入的文档化模块，但 CPython 代码库中的一些零碎信息向我们展示了很多如何使用它的信息。

这里有 2 个选择：

• 我们可以使用 C 语言实现的 _xxsubinterpreters 模块（因为是用 C 实现的内部模块，所以名字很奇怪）。

  import _xxsubinterpreters as interpreters
  

• 或者我们可以用 CPython 的测试模块，里面有用于测试的示例解释器（和通道）类。

  from test.support import interpreters
  

下面演示中我将采用第二种方式。

我们已经找到了子解释器，但我们还需要从 Python 的测试模块中引入一些辅助函数，我们将使用这些函数将代码传递给子解释器：

from textwrap import dedent
import os

def _captured_script(script):
    r, w = os.pipe()
    indented = script.replace('\n', '\n                ')
    wrapped = dedent(f"""
        import contextlib
        with open({w}, 'w', encoding="utf-8") as spipe:
            with contextlib.redirect_stdout(spipe):
                {indented}
        """)
    return wrapped, open(r, encoding="utf-8")


def _run_output(interp, request, channels=None):
    script, rpipe = _captured_script(request)
    with rpipe:
        interp.run(script, channels=channels)
        return rpipe.read()

将 interpreters 模块和上面的辅助函数放在一起，我们即可生成我们的第一个子解释器：

from test.support import interpreters

main = interpreters.get_main()
print(f"主解释器ID: {main}")
# 主解释器ID: Interpreter(id=0, isolated=None)

interp = interpreters.create()

print(f"子解释器ID: {interp}")
# 主解释器ID: Interpreter(id=1, isolated=True)


code = dedent("""
            from test.support import interpreters
            cur = interpreters.get_current()
            print(cur.id)
            """)

out = _run_output(interp, code)

print(f"所有解释器: {interpreters.list_all()}")
# 所有解释器: [Interpreter(id=0, isolated=None), Interpreter(id=1, isolated=None)]
print(f"Output: {out}")  # 'print(cur.id)'的执行结果
# Output: 1

如上面代码所示，生成和运行新解释器的一种方法是使用 create 函数，然后将解释器与我们要执行的代码一起传递给 _run_output 辅助函数。

interpreters 中有个 run 方法，可以直接调用 interpreters 的 run 方法执行代码，这样代码会更加直观：

interp = interpreters.create()
interp.run(code)

然而，当我们再次尝试运行上述代码片段，程序会报错：

Fatal Python error: PyInterpreterState_Delete: remaining subinterpreters
Python runtime state: finalizing (tstate=0x000055b5926bf398)

错误信息可以看到，是因为已经存在子解释器造成的。为了解决这个问题，我们还需要清理所有用完的解释器：

def cleanup_interpreters():
    for i in interpreters.list_all():
        if i.id == 0:  # 主解释器
            continue
        try:
            print(f"清理解释器: {i}")
            i.close()
        except RuntimeError:
            pass  # 已经销毁

cleanup_interpreters()
# 清理解释器: Interpreter(id=1, isolated=None)
# 清理解释器: Interpreter(id=2, isolated=None)

线程化

虽然上面的代码直接使用工具函数可以正常启动子解释器，但是代码稍微有点复杂。实际使用中我们希望子解释器运行独立任务，所以将子解释器封装成 thread 调用会更加方便好理解。

import threading

def run_in_thread():
    t = threading.Thread(target=interpreters.create)
    print(t)
    t.start()
    print(t)
    t.join()
    print(t)

run_in_thread()
run_in_thread()

# <Thread(Thread-1 (create), initial)>
# <Thread(Thread-1 (create), started 139772371633728)>
# <Thread(Thread-1 (create), stopped 139772371633728)>
# <Thread(Thread-2 (create), initial)>
# <Thread(Thread-2 (create), started 139772371633728)>
# <Thread(Thread-2 (create), stopped 139772371633728)>

上面的代码主要将 interpreters.create 函数传递给 threading.Thread，它会自动在线程内生成新的子解释器。

我们还可以将这两种方法结合起来，并将辅助函数传递给 threading.Thread：

import time

def run_in_thread():
    interp = interpreters.create(isolated=True)
    t = threading.Thread(target=_run_output, args=(interp, dedent("""
            import _xxsubinterpreters as _interpreters
            cur = _interpreters.get_current()
            
            import time
            time.sleep(2)
           
            assert isinstance(cur, _interpreters.InterpreterID)
            """)))
    print(f"创建线程: {t}")
    t.start()
    return t

t1 = run_in_thread()
print(f"第一次运行线程: {t1}")
t2 = run_in_thread()
print(f"第二次运行线程: {t2}")
time.sleep(4)  # 休眠一会儿，让线程执行结束

cleanup_interpreters()

上面代码中，我们没有使用 test.support 模块，而是演示了如何使用 _xxsubinterpreters 模块。我们还在每个线程中休眠 2 秒来模拟一些“工作”。注意，我们这里我们没有在线程上调用 join，只是在解释器完成时清理它们。

通道

如果我们更深入地研究 CPython 测试模块，我们还会发现 RecvChannel 和 SendChannel 类。这两个类实现了类似 Go 语言的通道。 使用它们：

r, s = interpreters.create_channel()

print(f"Channel: {r}, {s}")
# Channel: RecvChannel(id=0), SendChannel(id=0)

orig = b'hello'
s.send_nowait(orig)
obj = r.recv()
print(f"Received: {obj}")
# Received: b'hello'

cleanup_interpreters()

上面代码创建一个带有接收端（r）和发送端（s）的通道。然后我们使用 send_nowait 将数据传递出去，并在另一端使用 recv 函数读取它。这个通道来自子解释器，用完后需要进行清理。

子解释器配置

最后，我们可以通过 test.support 模块调整子解释器的选项。这些选项通常需要在 C 代码中调整，我发现 test.support 模块提供了 run_in_subinterp_with_config 方法可以轻松设置解释器的选项：

import test.support

def run_in_thread(script):
    test.support.run_in_subinterp_with_config(
        script,
        use_main_obmalloc=True,
        allow_fork=True,
        allow_exec=True,
        allow_threads=True,
        allow_daemon_threads=False,
        check_multi_interp_extensions=False,
        own_gil=True,
    )

code = dedent(f"""
            from test.support import interpreters
            cur = interpreters.get_current()
            print(cur)
            """)

run_in_thread(code)
# Interpreter(id=7, isolated=None)
run_in_thread(code)
# Interpreter(id=8, isolated=None)

这个函数是 C 函数的 Python API 封装。它提供了一些子解释器选项，例如 own_gil，用于指定子解释器是否有自己的 GIL。

结论

我很高兴 Python 终于能够实现真正的线程了！

然而——正如你看到的那样——目前的API 并不易用，因此除非你具有 C 专业知识或非常迫切需要子解释器，否则我不建议你将上面的方法用于生产环境。或者，你可以尝试一下 extrainterpreters 项目，它为子解释器提供了更友好的 Python API。

虽然子解释器对于普通 Python 开发人员来说很难使用，但我相信工具/库开发人员会很好地利用它，我们期待看到更多库借助子解释器有出色的性能优化。