Python 協程實作康威生命遊戲平行模擬

協程在 Python 中提供了一種更簡潔的平行處理方案，尤其在模擬平行行為的場景下更具優勢。相較於多執行緒容易遇到的資源競爭、鎖定等問題，協程以單執行緒模擬平行，大幅降低了程式設計的複雜度。本文將以康威生命遊戲為例，逐步拆解如何使用 Python 協程模擬細胞的生死演變。生命遊戲的核心規則簡明扼要：活細胞周圍少於兩個活細胞會因孤獨而死，多於三個則因擁擠而死；死細胞周圍恰好三個活細胞則會復活。透過協程，我們可以將每個細胞的行為模擬成一個獨立的單元，並透過彼此互動來演繹整個生命遊戲的過程。首先，我們需要設計 count_neighbors 協程來統計每個細胞周圍的活細胞數量，作為細胞狀態轉變的依據。接著，step_cell 協程則根據生命遊戲的規則和鄰居細胞的狀態，決定細胞的生死。這些協程彼此協作，共同構成了生命遊戲的模擬引擎。

生命遊戲：Python協程的平行模擬藝術

為何我擁抱協程：告別多執行緒的複雜性

在追求高效能的道路上，Python開發者經常面臨多執行緒的挑戰。資源競爭、鎖定、以及難以追蹤的錯誤，都讓多執行緒程式設計變得複雜。但自從我開始深入研究協程（Coroutines）後，我發現了一種更優雅、更輕量級的平行處理方式。協程不僅能避免多執行緒的常見問題，還能以更直觀的方式模擬平行行為。

協程的魔力：暫停與還原的藝術

協程，本質上是一種特殊的生成器函式。與傳統函式不同，協程可以在執行過程中暫停（透過yield表示式），並在稍後還原執行（透過send方法）。這種暫停與還原的機制，讓協程能夠在單一執行緒中模擬平行，而無需多執行緒的複雜性。

def my_coroutine():
    print("協程開始")
    x = yield  # 暫停，等待外部傳入值
    print(f"協程收到值: {x}")
    yield x * 2  # 再次暫停，並傳回值
    print("協程結束")

# 啟動協程
coro = my_coroutine()
next(coro)  # 啟動協程至第一個yield

# 與協程互動
print(coro.send(10))  # 傳送值給協程，並取得下一個yield的傳回值

try:
    next(coro)  # 繼續執行協程
except StopIteration:
    print("協程已完成")

內容解密

• 首先，my_coroutine() 是一個協程函式，它使用 yield 關鍵字來暫停和還原執行。
• next(coro) 用於啟動協程，使其執行到第一個 yield 陳述式。
• coro.send(10) 將值 10 傳送給協程，協程接收到這個值並賦給變數 x。然後，協程繼續執行，直到下一個 yield 陳述式，並傳回 x * 2 的結果。
• 最後，next(coro) 再次嘗試執行協程，但由於協程已經執行完畢，會引發 StopIteration 異常。

生命遊戲：協程平行模擬的絕佳範例

為了展示協程的平行能力，我選擇了康威生命遊戲（Conway’s Game of Life）作為範例。這個簡單的遊戲，透過模擬細胞的生死演變，展現了複雜的行為模式。

生命遊戲的規則非常簡單：

1. 每個細胞可以是活的（ALIVE）或空的（EMPTY）。
2. 每個細胞會檢查其周圍八個鄰居的狀態。
3. 根據鄰居的數量，細胞決定下一刻的狀態：
   • 活細胞若鄰居少於 2 個，則死亡（太孤單）。
   • 活細胞若鄰居多於 3 個，則死亡（太擁擠）。
   • 死細胞若鄰居恰好有 3 個，則復活。

細胞的協程化：count_neighbors 的設計

為了模擬生命遊戲，我將每個細胞視為一個協程。首先，我需要一個方法來取得鄰居細胞的狀態。count_neighbors 協程負責查詢鄰居狀態，並傳回活細胞的數量。

from collections import namedtuple

Query = namedtuple('Query', ('y', 'x'))
ALIVE = '*'
EMPTY = '-'

def count_neighbors(y, x):
    n_  = yield Query(y + 1, x + 0)  # 北
    ne  = yield Query(y + 1, x + 1)  # 東北
    e_  = yield Query(y + 0, x + 1)  # 東
    se  = yield Query(y - 1, x + 1)  # 東南
    s_  = yield Query(y - 1, x + 0)  # 南
    sw  = yield Query(y - 1, x - 1)  # 西南
    w_  = yield Query(y + 0, x - 1)  # 西
    nw  = yield Query(y + 1, x - 1)  # 西北

    neighbor_states = [n_, ne, e_, se, s_, sw, w_, nw]
    count = 0
    for state in neighbor_states:
        if state == ALIVE:
            count += 1
    return count

內容解密

• Query = namedtuple('Query', ('y', 'x'))：定義一個具名元組 Query，用於表示查詢特定座標細胞狀態的請求。
• ALIVE = '*' 和 EMPTY = '-'：定義活細胞和空細胞的符號。
• count_neighbors(y, x) 協程函式：
  • 接收細胞的座標 y 和 x 作為引數。
  • 使用 yield Query(y + 1, x + 0) 等陳述式，向周圍八個鄰居發出查詢請求，請求它們的狀態。每次 yield 都會暫停協程的執行，直到透過 send() 方法接收到鄰居的狀態。
  • 將所有鄰居的狀態收集到 neighbor_states 列表中。
  • 遍歷 neighbor_states 列表，計算活細胞的數量。
  • 傳回活細胞的總數。

細胞狀態的轉變：step_cell 的設計

step_cell 協程負責決定細胞在下一個時間點的狀態。它首先查詢自身的狀態，然後使用 count_neighbors 取得鄰居狀態，最後根據生命遊戲的規則，決定下一個狀態。

Transition = namedtuple('Transition', ('y', 'x', 'state'))

def game_logic(state, neighbors):
    if state == ALIVE:
        if neighbors < 2:
            return EMPTY  # 死亡：太少鄰居
        elif neighbors > 3:
            return EMPTY  # 死亡：太多鄰居
        else:
            return state  # 存活：鄰居數量適中
    else:
        if neighbors == 3:
            return ALIVE  # 復活：恰好3個鄰居
        else:
            return EMPTY  # 維持空：鄰居數量不對

def step_cell(y, x):
    state = yield Query(y, x)
    neighbors = yield from count_neighbors(y, x)
    next_state = game_logic(state, neighbors)
    yield Transition(y, x, next_state)

內容解密

• Transition = namedtuple('Transition', ('y', 'x', 'state'))：定義一個具名元組 Transition，用於表示細胞狀態的轉變。
• game_logic(state, neighbors) 函式：
  • 接收細胞當前狀態 state 和鄰居數量 neighbors 作為引數。
  • 根據生命遊戲的規則，判斷細胞的下一個狀態，並傳回。
• step_cell(y, x) 協程函式：
  • 接收細胞的座標 y 和 x 作為引數。
  • 首先使用 yield Query(y, x) 查詢細胞當前的狀態。
  • 然後使用 neighbors = yield from count_neighbors(y, x) 呼叫 count_neighbors 協程，取得鄰居的數量。yield from 允許一個協程將部分操作委託給另一個協程，並接收其傳回值。
  • 接著呼叫 game_logic(state, neighbors) 函式，根據細胞的當前狀態和鄰居數量，計算出細胞的下一個狀態。
  • 最後使用 yield Transition(y, x, next_state) 發出一個轉變訊號，指示細胞應該轉變為新的狀態。