启用for循环后Python多线程康威生命游戏输出异常排查

核心问题分析 #

你遇到的打印次数不足且不稳定的问题，根源在于三个线程同步错误：

1. 竞态条件导致计数错误：numberOfNeighborsWhoRead的修改没有加锁，多个线程同时执行neighbor.numberOfNeighborsWhoRead +=1时，会出现数值覆盖，最终计数永远达不到预期的numberOfNeighbors，线程卡在while循环中无法退出。
2. 状态读取逻辑错误：外层循环启用后，neighborStates只在第一次迭代前获取邻居状态，后续迭代复用旧状态，既不符合生命游戏规则，也会导致部分线程的状态更新逻辑异常。
3. 忙等逻辑不可靠：用time.sleep(0.01)轮询等待的方式，无法保证线程能及时响应计数变化，大概率会出现线程永久阻塞。

修复后的代码实现 #

import threading
import time

class Cell(threading.Thread):
    def __init__(self, threadId, numberOfNeighbors, state):
        threading.Thread.__init__(self)
        self.threadId = threadId
        self.numberOfNeighbors = numberOfNeighbors 
        self.state = state 
        self.numberOfNeighborsWhoRead = 0
        # 用Condition替代单纯Lock，支持等待通知机制
        self.cond = threading.Condition()
        self.neighbors = None

    def calculateState(self, neighborStates):
        # 实现康威生命游戏核心规则
        alive_neighbors = sum(neighborStates)
        if self.state == 1:
            return 1 if alive_neighbors in (2, 3) else 0
        else:
            return 1 if alive_neighbors == 3 else 0

    def run(self):
        for x in range(3):
            neighborStates = []
            # 1. 线程安全地读取邻居状态并更新邻居的读取计数
            for neighbor in self.neighbors:
                with neighbor.cond:
                    neighbor.numberOfNeighborsWhoRead += 1
                    neighborStates.append(neighbor.state)
                    neighbor.cond.notify()  # 通知等待的邻居计数更新
            
            # 2. 等待所有邻居完成对当前细胞的状态读取
            with self.cond:
                while self.numberOfNeighborsWhoRead < self.numberOfNeighbors:
                    self.cond.wait()  # 阻塞直到计数达标，自动释放锁
                self.numberOfNeighborsWhoRead = 0  # 重置计数准备下一轮
            
            # 3. 线程安全地更新当前细胞状态
            with self.cond:
                self.state = self.calculateState(neighborStates)
            
            print(f"Iteration {x+1} done for cell {self.threadId}")

# 测试用例：创建10×10细胞网格并启动线程
if __name__ == "__main__":
    grid = []
    # 初始化细胞
    for i in range(10):
        row = []
        for j in range(10):
            state = (i + j) % 2
            # 计算当前细胞的邻居数量（角落3个、边缘非角落5个、内部8个）
            if i in (0, 9) and j in (0, 9):
                neighbor_count = 3
            elif i in (0, 9) or j in (0, 9):
                neighbor_count = 5
            else:
                neighbor_count = 8
            cell = Cell(threadId=i*10 + j, numberOfNeighbors=neighbor_count, state=state)
            row.append(cell)
        grid.append(row)
    
    # 为每个细胞设置邻居
    for i in range(10):
        for j in range(10):
            neighbors = []
            for di in (-1, 0, 1):
                for dj in (-1, 0, 1):
                    if di == 0 and dj == 0:
                        continue
                    ni, nj = i + di, j + dj
                    if 0 <= ni < 10 and 0 <= nj < 10:
                        neighbors.append(grid[ni][nj])
            grid[i][j].neighbors = neighbors
    
    # 启动所有线程并等待完成
    for row in grid:
        for cell in row:
            cell.start()
    for row in grid:
        for cell in row:
            cell.join()
    
    print("All 300 iterations completed!")

关键修改点说明 #

• 替换Lock为Condition：结合锁机制和等待通知，避免低效的忙等，线程只有在计数达标时才会被唤醒，提升效率和可靠性。
• 线程安全的计数操作：所有对numberOfNeighborsWhoRead的读写都在with self.cond块中执行，彻底解决竞态条件问题。
• 每轮迭代重新读取状态：在for x in range(3)循环内每次都重新获取邻居的最新状态，符合生命游戏每轮基于当前状态计算下一轮的规则。
• 处理虚假唤醒：用while循环检查计数条件，避免Condition的wait被虚假唤醒导致的逻辑错误。

内容的提问来源于stack exchange，提问作者baneballet