Python 的生成器和協程提供優雅的迭代和非同步處理機制。生成器利用 yield 關鍵字產生值序列，無需一次性儲存所有結果。協程則透過 async 和 await 進行非同步操作，提升程式效能。asyncio 函式庫提供事件迴圈和任務管理，簡化非同步程式設計。本文也探討了子直譯器，它允許在單一程式中建立多個獨立直譯器，進一步提升平行度，適用於 Web 服務和科學計算等場景。理解這些概念有助於編寫更有效率且易於維護的 Python 程式。
生成器的結構
生成器的結構包括以下幾個部分：
__code__: 編譯後的函式程式碼物件（PyCodeObject）__exc_state__: 異常狀態（_PyErr_StackItem）__frame__: 當前的框架物件（PyFrameObject）__name__: 生成器的名稱（str）__qualname__: 生成器的限定名稱（str）__running__: 生成器是否正在執行（char）__weakreflist__: 弱參照列表（list）
協程
協程（coroutine）是一種特殊的生成器，它可以用來實作非同步程式設計。協程可以暫停和還原執行，允許其他協程執行。
生成器和協程的區別
生成器和協程都是特殊的迭代器，但它們有以下區別：
- 生成器是透過
yield關鍵字定義的，而協程是透過async和await關鍵字定義的。 - 生成器只能用於同步程式設計，而協程可以用於非同步程式設計。
Python 中的生成器和協程
Python 中的生成器和協程是透過以下幾個類別實作的：
PyGenObject: 生成器物件PyCoroObject: 協程物件PyAsyncGenObject: 非同步生成器物件
這些類別提供了生成器和協程的基本屬性和方法，包括 __code__、__exc_state__、__frame__ 等。
圖表翻譯：
上述 Mermaid 圖表展示了生成器和協程之間的關係。生成器透過 yield 關鍵字產生值，而協程透過 await 關鍵字等待任務完成。圖表中，A 表示生成器，B 表示生成器產生的值，C 表示協程，D 表示協程等待的任務。
生成器與協程的實作
在 Python 中，生成器和協程是兩種特殊的物件，分別用於實作迭代器和非同步運算。當編譯器遇到包含 yield 的函式時，會將其編譯為一個生成器物件。
生成器的建立
生成器的建立過程涉及到 _PyEval_EvalCode 函式，這個函式會檢查編譯後的程式碼物件是否包含 CO_GENERATOR、CO_COROUTINE 或 CO_ASYNC_GENERATOR 標誌。如果包含其中任何一個標誌，則會建立一個新的生成器、協程或非同步生成器物件。
PyObject *
_PyEval_EvalCode(PyObject *_co, PyObject *globals, PyObject *locals,...)
{
//...
if (co->co_flags & (CO_GENERATOR | CO_COROUTINE | CO_ASYNC_GENERATOR)) {
PyObject *gen;
PyObject *coro_wrapper = tstate->coroutine_wrapper;
int is_coro = co->co_flags & CO_COROUTINE;
//...
if (is_coro) {
gen = PyCoro_New(f, name, qualname);
} else if (co->co_flags & CO_ASYNC_GENERATOR) {
gen = PyAsyncGen_New(f, name, qualname);
} else {
gen = PyGen_NewWithQualName(f, name, qualname);
}
//...
return gen;
}
//...
}
生成器的結構
生成器物件包含了以下結構：
gi_code: 指向編譯後的程式碼物件。gi_running: 表示生成器是否正在執行。gi_frame: 指向生成器的框架物件。
typedef struct {
PyObject_HEAD
PyObject *gi_code; /* 編譯後的程式碼物件 */
int gi_running; /* 生成器是否正在執行 */
PyObject *gi_frame; /* 生成器的框架物件 */
//...
} PyGenObject;
生成器的執行
當呼叫生成器的 __next__ 方法時，會執行 gen_send_ex 函式。這個函式會根據生成器的狀態和傳入的引數，執行不同的動作。
PyObject *
gen_send_ex(PyGenObject *gen, PyObject *arg, PyObject *kwargs)
{
//...
if (gen->gi_running) {
// 生成器正在執行，繼續執行
} else {
// 生成器尚未執行，初始化並開始執行
}
//...
}
協程的實作
協程是特殊的生成器，用於實作非同步運算。協程的建立和執行過程與生成器類別似，但有一些額外的限制和檢查。
PyObject *
PyCoro_New(PyObject *f, PyObject *name, PyObject *qualname)
{
//...
PyCoroObject *coro = PyObject_New(PyCoroObject, &PyCoro_Type);
coro->co_f = f;
coro->co_name = name;
coro->co_qualname = qualname;
//...
return (PyObject *)coro;
}
7. 函式呼叫框架中的 f_back 欄位
函式呼叫框架中的 f_back 欄位包含了呼叫端的資訊，該資訊用於傳遞傳回值。因此，當函式傳回時，f_back 欄位會被設定為目前的呼叫框架，以便傳回值可以正確地傳遞給呼叫端。
8. 註冊函式為正在執行中
當函式被呼叫時，它會被註冊為正在執行中，這樣可以確保函式的執行狀態被正確地追蹤。
9. 儲存最後一次異常資訊
函式的最後一次異常資訊會被儲存，以便在需要時可以正確地處理異常。
10. 記錄呼叫端的異常資訊
呼叫端的異常資訊會被記錄在函式的異常資訊中，這樣可以確保異常可以被正確地傳遞給呼叫端。
11. 執行函式內的程式碼
函式內的程式碼會在 Python 的主迴圈中被執行，然後傳回值會被傳遞給呼叫端。
12. 重置最後一次異常資訊
最後一次異常資訊會被重置為呼叫函式前的狀態。
13. 註冊函式為未執行中
當函式傳回時，它會被註冊為未執行中，這樣可以確保函式的執行狀態被正確地追蹤。
14. 處理傳回值和異常
根據傳回值的不同，可能會有多種異常情況出現，例如當函式傳回 None 時，可能會引發 StopIteration 異常。
15. 傳回結果給呼叫端
最終，函式的傳回結果會被傳遞給呼叫端的 __next__() 方法。
關於協程
協程是一種強大的語法結構，它可以透過 yield 關鍵字來建立一個唯一的物件，並將編譯後的程式碼作為屬性儲存。然而，協程有一個限制，即它只能將值傳回給其直接呼叫端。為瞭解決這個限制，Python 引入了 yield from 語法，可以將協程轉換為包含 yield from 的輔助函式。
範例：使用 yield from 建立協程
以下範例展示瞭如何使用 yield from 建立一個協程：
def gen_letters(start, x):
i = start
end = start + x
while i < end:
yield chr(i)
i += 1
def letters(upper):
if upper:
yield from gen_letters(65, 26) # A--Z
else:
yield from gen_letters(97, 26) # a--z
for letter in letters(False):
print(letter)
在這個範例中，gen_letters 是一個協程，它生成了一系列字母。letters 函式使用 yield from 來委託 gen_letters 協程，從而可以根據 upper 引數決定生成大寫還是小寫字母。
平行處理與競爭性
在 Python 中，asyncio 模組提供了一種方式來實作平行處理和競爭性。平行處理允許程式同時執行多個任務，而競爭性則是指多個任務競爭資源的能力。
生成器和協程
生成器是一種特殊的迭代器，可以用來產生一系列的值。協程則是指可以暫停和還原執行的函式。Python 的 asyncio 模組提供了一種方式來實作協程。
import asyncio
async def sleepy_alarm(time):
await asyncio.sleep(time)
return "wake up!"
alarm = sleepy_alarm(10)
print(alarm) # <coroutine object sleepy_alarm at 0x1041de340>
執行協程
可以使用 asyncio.run() 函式來執行協程。
import asyncio
async def sleepy_alarm(time):
await asyncio.sleep(time)
return "wake up!"
print(asyncio.run(sleepy_alarm(10))) # wake up!
任務和事件迴圈
任務是指可以被執行的協程。事件迴圈是指用來管理任務的物件。可以使用 asyncio.new_event_loop() 函式來建立一個新的事件迴圈。
import asyncio
loop = asyncio.new_event_loop()
回呼函式
回呼函式是指當任務完成或失敗時會被呼叫的函式。可以使用 asyncio.Task 類別來建立一個任務，並指定回呼函式。
import asyncio
async def sleepy_alarm(person, time):
await asyncio.sleep(time)
return "wake up!"
task = asyncio.Task(sleepy_alarm("John", 10))
平行執行
可以使用 asyncio.gather() 函式來平行執行多個任務。
import asyncio
async def sleepy_alarm(person, time):
await asyncio.sleep(time)
return "wake up!"
tasks = [sleepy_alarm("John", 10), sleepy_alarm("Mary", 5)]
results = asyncio.run(asyncio.gather(*tasks))
print(results) # ["wake up!", "wake up!"]
圖表翻譯：
graph LR
A[任務] --> B[事件迴圈]
B --> C[執行]
C --> D[完成]
D --> E[回呼函式]
圖表描述了任務、事件迴圈、執行、完成和回呼函式之間的關係。
內容解密：
上述程式碼示範瞭如何使用 asyncio 模組來實作平行處理和競爭性。sleepy_alarm 函式是一個協程，可以暫停和還原執行。asyncio.run() 函式用來執行協程。asyncio.new_event_loop() 函式用來建立一個新的事件迴圈。asyncio.Task 類別用來建立一個任務，並指定回呼函式。asyncio.gather() 函式用來平行執行多個任務。
非同步程式設計與 asyncio
在 Python 中，asyncio 是一個用於撰寫單執行緒併發程式的函式庫。它可以讓你的程式同時執行多個任務，而不需要使用多執行緒或多程式。
基本概念
- 協程（Coroutine）：是一種可以暫停和還原執行的函式。它可以在任意時候將控制權交給其他協程。
- 事件迴圈（Event Loop）：是一個負責管理協程的執行和切換的迴圈。它會不斷地檢查哪些協程已經準備好執行，並將控制權交給它們。
- 任務（Task）：是一個包裝了協程的物件。它可以被用來管理協程的執行和取消。
使用 asyncio 的好處
- 提高併發性：asyncio 可以讓你的程式同時執行多個任務，從而提高併發性和效率。
- 簡化程式設計：asyncio 提供了一個簡單的 API，可以讓你輕鬆地撰寫併發程式。
範例：非同步醒來
以下是使用 asyncio 實作的非同步醒來範例：
import asyncio
async def sleepy_alarm(person, delay):
await asyncio.sleep(delay)
print(f"{person} -- wake up!")
async def wake_up_gang():
tasks = [
asyncio.create_task(sleepy_alarm("Bob", 3), name="wake up Bob"),
asyncio.create_task(sleepy_alarm("Yudi", 4), name="wake up Yudi"),
asyncio.create_task(sleepy_alarm("Doris", 2), name="wake up Doris"),
asyncio.create_task(sleepy_alarm("Kim", 5), name="wake up Kim")
]
await asyncio.gather(*tasks)
asyncio.run(wake_up_gang())
這個範例會輸出：
Doris -- wake up!
Bob -- wake up!
Yudi -- wake up!
Kim -- wake up!
範例：非同步埠掃描
以下是使用 asyncio 實作的非同步埠掃描範例：
import asyncio
timeout = 1.0
async def check_port(host, port, results):
try:
future = asyncio.open_connection(host=host, port=port)
r, w = await asyncio.wait_for(future, timeout=timeout)
results.append(port)
w.close()
except OSError:
pass
except asyncio.TimeoutError:
pass
async def scan(start, end, host):
tasks = []
results = []
for port in range(start, end):
task = asyncio.create_task(check_port(host, port, results))
tasks.append(task)
await asyncio.gather(*tasks)
return results
async def main():
host = "example.com"
start = 1
end = 100
results = await scan(start, end, host)
print(results)
asyncio.run(main())
這個範例會掃描 example.com 的 1 到 100 埠，並輸出開啟的埠號碼。
非同步掃描技術
在網路安全領域中，掃描技術是一種重要的工具，用於檢測和識別網路伺服器上的開放埠。傳統的同步掃描方法存在效率低下的問題，因為它需要等待每個埠的掃描結果才會繼續掃描下一個埠。為瞭解決這個問題，我們可以使用非同步掃描技術。
非同步掃描原理
非同步掃描技術使用了非同步程式設計的概念，允許多個任務同時執行，而不需要等待前一個任務完成。這樣可以大大提高掃描的效率。
實作非同步掃描
以下是使用 Python 的 asyncio函式庫實作非同步掃描的例子：
import asyncio
async def check_port(host, port, timeout=1):
try:
future = asyncio.open_connection(host=host, port=port)
r, w = await asyncio.wait_for(future, timeout=timeout)
return port
except asyncio.TimeoutError:
return None
async def scan(start, end, host):
tasks = []
for port in range(start, end):
task = asyncio.create_task(check_port(host, port))
tasks.append(task)
results = await asyncio.gather(*tasks)
return [port for port in results if port is not None]
if __name__ == '__main__':
host = "localhost"
start = 80
end = 100
results = asyncio.run(scan(start, end, host))
for result in results:
print(f"Port {result} is open")
在這個例子中，我們定義了兩個非同步函式：check_port 和 scan。check_port 函式用於檢查一個埠是否開放，而 scan 函式用於掃描一系列的埠。
非同步生成器
非同步生成器是一種特殊的生成器，可以用於產生一系列的值。它們與普通的生成器類別似，但需要使用 async for 來迭代。
以下是使用非同步生成器實作掃描的例子：
async def check_ports(host, start, end, max=10):
found = 0
for port in range(start, end):
try:
future = asyncio.open_connection(host=host, port=port)
r, w = await asyncio.wait_for(future, timeout=1)
yield port
found += 1
if found >= max:
break
except asyncio.TimeoutError:
pass
if __name__ == '__main__':
host = "localhost"
start = 80
end = 100
async for port in check_ports(host, start, end):
print(f"Port {port} is open")
在這個例子中，我們定義了一個非同步生成器 check_ports，它產生一系列的開放埠。
瞭解 Python 中的子直譯器（Subinterpreters）
Python 中的子直譯器（Subinterpreters）是一種允許在單一程式中建立多個獨立直譯器例項的機制。每個子直譯器都有自己的全域性鎖（GIL），這意味著多個子直譯器可以平行執行，從而提高了整體的平行度。
建立子直譯器
要建立一個子直譯器，可以使用 Py_NewInterpreter() 函式。這個函式傳回一個新的直譯器指標，然後你可以使用 PyEval_InitThreads() 來初始化這個直譯器的執行緒支援。
PyObject* new_interpreter = Py_NewInterpreter();
if (new_interpreter) {
PyEval_InitThreads();
// 使用新的直譯器
} else {
// 處理錯誤
}
子直譯器的優點
子直譯器提供了幾個優點，包括：
- 低開銷：建立子直譯器比建立新的程式或執行緒需要更少的資源。
- 平行執行：由於每個子直譯器都有自己的 GIL，所以多個子直譯器可以平行執行，提高了整體的平行度。
- 獨立性：每個子直譯器都有自己的狀態和記憶體空間，所以它們之間不會相互幹擾。
子直譯器的應用場景
子直譯器可以用於以下場景：
- 平行計算：使用多個子直譯器來平行執行計算任務。
- Web 伺服器：使用子直譯器來處理多個請求，提高 Web 伺服器的平行度。
- 科學計算：使用子直譯器來平行執行科學計算任務。
從Python底層實作到高階應用的全面檢視顯示，生成器和協程提供了一種優雅且高效的處理迭代和非同步操作的機制。藉由yield和async/await關鍵字，Python賦予開發者更精細的控制流程的能力，得以實作惰性求值、非同步I/O等複雜功能。然而，深入剖析其核心架構可以發現，協程在傳回值傳遞上存在侷限性，yield from的引入有效地解決了這個問題，同時也展現了Python語言持續演進的活力。
技術堆疊的各層級協同運作中體現，生成器和協程並非孤立的概念，它們與Python的直譯器、函式呼叫框架、例外處理機制等緊密結合。理解這些底層機制，才能更有效地運用生成器和協程，例如，子直譯器的引入，為Python的平行處理能力開啟了新的可能性，允許在單一程式中實作真正的平行執行，進一步提升了程式效能。同時，非同步程式設計模型的應用，例如在網路掃描中的實踐，展現了協程在處理I/O密集型任務時的優勢，有效地解決了傳統同步模型的效率瓶頸。
展望未來3-5年的技術演進路徑預測，隨著Python在資料科學、機器學習等領域的廣泛應用，生成器和協程將扮演 increasingly 重要的角色。預計Python社群將持續探索更進階的非同步程式設計模式和最佳實務，例如結構化平行、非同步上下文管理等，以進一步簡化非同步程式碼的開發和維護。同時，子直譯器技術的發展和應用，也將為Python的平行計算能力帶來更大的提升。
玄貓認為，深入理解生成器和協程的底層機制，並掌握非同步程式設計的最佳實務，對於Python開發者而言至關重要。這不僅能提升程式碼的效能和可維護性，更能讓開發者更好地應對未來技術發展的挑戰。
