32歲的Python依然沒有真正的并行性/并發性。然而，這種情況即將發生改變，因為在即將發布的Python 3.12中引入了名為"Per-Interpreter GIL"的新特性。在距離發布還有幾個月的時間（預計2023年10月發布），但相關代碼已經有了，因此，我們可以提前了解如何使用子解釋器API編寫真正的并發Python代碼。
子解釋器
首先，讓我們來解釋一下如何通過“Per-Interpreter GIL”來解決Python缺乏適當的并發性問題。
在Python中，GIL是一個互斥鎖，它只允許一個線程控制Python解釋器。這意味著即使在Python中創建多個線程（例如使用線程模塊），也只有一個線程會運行。
隨著“Per-Interpreter GIL”的引入，各個Python解釋器不再共享同一個GIL。這種隔離級別允許每個子解釋器可以同時運行。這意味著我們可以通過生成額外的子解釋器來繞過Python的并發限制，其中每個子解釋器都有自己的GIL（全局狀態）。
更詳細的說明請參見PEP 684，該文檔描述了此功能/更改：https://peps.python.org/pep-0684/#per-interpreter-state
上手體驗
安裝
為使用這項最新功能，我們必須安裝最新版的Python，并且需要從源碼上進行構建：
# https://devguide.python.org/getting-started/setup-building/#unix-compiling
git clone https://github.com/python/cpython.git
cd cpython
./configure --enable-optimizations --prefix=$(pwd)/python-3.12
make -s -j2
./python
# Python 3.12.0a7+ (heads/main:22f3425c3d, May 10 2023, 12:52:07) [GCC 11.3.0] on linux
# Type "help", "copyright", "credits" or "license" for more information.
C-API在哪里？
既然已經安裝了最新的版本，那我們該如何使用子解釋器呢？可以直接導入嗎？不，正如PEP-684中所提到的：“這是一個高級功能，專為C-API的一小部分用戶設計?！?/span>
目前，Per-Interpreter GIL特性只能通過C-API使用，因此Python開發人員沒有直接的接口可以使用。這樣的接口預計將隨著PEP 554一起推出，如果被采納，則應該會在Python 3.13中實現，在那之前，我們必須自己想辦法實現子解釋器。
通過CPython代碼庫中的一些零散記錄，我們可以采用下面兩種方法：
使用_xxsubinterpreters模塊，該模塊是用C實現的，因此名稱看起來有些奇怪。由于它是用C實現的，所以開發者無法輕易地檢查代碼（至少不是在 Python 中）；
或者可以利用CPython的測試模塊，該模塊具有用于測試的示例 Interpreter（和 Channel）類。
# Choose one of these:
import _xxsubinterpreters as interpreters
from test.support import interpreters
在接下來的演示中，我們將主要采用第二種方法。我們已經找到了子解釋器，但還需要從Python的測試模塊中借用一些輔助函數，以便將代碼傳遞給子解釋器：
from textwrap import dedent
import os
# https://github.com/python/cpython/blob/
# 15665d896bae9c3d8b60bd7210ac1b7dc533b093/Lib/test/test__xxsubinterpreters.py#L75
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"Main interpreter ID: {main}")
# Main interpreter ID: Interpreter(id=0, isolated=None)
interp = interpreters.create()
print(f"Sub-interpreter: {interp}")
# Sub-interpreter: Interpreter(id=1, isolated=True)
# https://github.com/python/cpython/blob/
# 15665d896bae9c3d8b60bd7210ac1b7dc533b093/Lib/test/test__xxsubinterpreters.py#L236
code = dedent("""
from test.support import interpreters
cur = interpreters.get_current()
print(cur.id)
""")
out = _run_output(interp, code)
print(f"All Interpreters: {interpreters.list_all()}")
# All Interpreters: [Interpreter(id=0, isolated=None), Interpreter(id=1, isolated=None)]
print(f"Output: {out}") # Result of 'print(cur.id)'
# Output: 1
生成和運行新解釋器的一個方法是使用create函數，然后將解釋器與要執行的代碼一起傳遞給_run_output輔助函數。
更簡單點的方法是：
interp = interpreters.create()
interp.run(code)
使用解釋器中的run方法?？墒?，如果我們運行上述代碼中的任意一個，都會得到如下錯誤：
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: # main
continue
try:
print(f"Cleaning up interpreter: {i}")
i.close()
except RuntimeError:
pass # already destroyed
cleanup_interpreters()
# Cleaning up interpreter: Interpreter(id=1, isolated=None)
# Cleaning up interpreter: Interpreter(id=2, isolated=None)
線程
雖然使用上述輔助函數運行代碼是可行的，但使用線程模塊中熟悉的接口可能更加方便：
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函數傳遞給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)
# Can't print from here, won't bubble-up to main interpreter
assert isinstance(cur, _interpreters.InterpreterID)
""")))
print(f"Created Thread: {t}")
t.start()
return t
t1 = run_in_thread()
print(f"First running Thread: {t1}")
t2 = run_in_thread()
print(f"Second running Thread: {t2}")
time.sleep(4) # Need to sleep to give Threads time to complete
cleanup_interpreters()
這里，我們演示了如何使用_xxsubinterpreters模塊而不是test.support中的模塊。我們還在每個線程中睡眠2秒鐘來模擬一些“工作”。請注意，我們甚至不必調用join()函數等待線程完成，只需在線程完成時清理解釋器即可。
Channels
如果我們深入研究CPython測試模塊，我們還會發現有 RecvChannel 和 SendChannel 類的實現，它們類似于 Golang 中的通道（Channel）。要使用它們：
# https://github.com/python/cpython/blob/
# 15665d896bae9c3d8b60bd7210ac1b7dc533b093/Lib/test/test_interpreters.py#L583
r, s = interpreters.create_channel()
print(f"Channel: {r}, {s}")
# Channel: RecvChannel(id=0), SendChannel(id=0)
orig = b'spam'
s.send_nowait(orig)
obj = r.recv()
print(f"Received: {obj}")
# Received: b'spam'
cleanup_interpreters()
# Need clean up, otherwise:
# free(): invalid pointer
# Aborted (core dumped)
這個例子展示了如何創建一個帶有receiver（r）和sender（s）端的通道。我們可以使用send_nowait將數據傳遞給發送方，并使用recv函數在另一側讀取它。這個通道實際上只是另一個子解釋器 - 所以與之前一樣 - 我們需要在完成后進行清理。
深入挖掘
最后，如果我們想要干擾或調整在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)
這個函數是一個Python API，用于調用C函數。它提供了一些子解釋器選項，如own_gil，指定子解釋器是否應該擁有自己的GIL。
總結
話雖如此——也正如你所看到的，API調用并不簡單，除非你已具備C語言專業知識，并且又迫切想要使用字解釋器，否則建議還是等待Python 3.13的發布?；蛘吣梢試L試extrainterpreters項目，該項目提供更友好的Python API以便使用子解釋器。
掃碼二維碼 獲取免費視頻學習資料
- 本文固定鏈接: http://www.stbrigidsathleticclub.com/post/11373/
- 轉載請注明：轉載必須在正文中標注并保留原文鏈接
- 掃碼： 掃上方二維碼獲取免費視頻資料
查 看2022高級編程視頻教程免費獲取