如何避免过山车多线程场景的偶发死锁？附经典问题编程需求

针对你描述的场景——n名乘客反复乘坐容量为C的过山车（C < n），必须坐满才发车——我用Python写了一个线程安全的模拟实现，注释很详细，你可以直接运行看效果：

import threading
import time
import random

class RollerCoaster:
    def __init__(self, capacity):
        self.capacity = capacity  # 车厢最大容量
        self.passengers_on_board = 0  # 当前车上的乘客数
        self.condition = threading.Condition()  # 用于同步的条件变量
        self.ride_in_progress = False  # 标记过山车是否正在运行

    def board(self, passenger_id):
        """乘客上车的方法"""
        with self.condition:
            # 等待车厢有空位且未处于运行状态
            while self.passengers_on_board >= self.capacity or self.ride_in_progress:
                print(f"乘客{passenger_id}：排队等待上车...")
                self.condition.wait()
            
            # 执行上车操作
            self.passengers_on_board += 1
            print(f"乘客{passenger_id}：成功上车！当前车上已有{self.passengers_on_board}人")
            
            # 如果刚好坐满，通知车厢可以发车
            if self.passengers_on_board == self.capacity:
                print("=== 车厢坐满，即将发车 ===")
                self.condition.notify()

    def wait_for_ride_end(self, passenger_id):
        """乘客等待过山车行程结束"""
        with self.condition:
            # 等待过山车开始运行
            while not self.ride_in_progress:
                self.condition.wait()
            print(f"乘客{passenger_id}：过山车启动啦！太爽了！")
            # 等待行程结束的通知
            while self.ride_in_progress:
                self.condition.wait()

    def alight(self, passenger_id):
        """乘客下车的方法"""
        with self.condition:
            self.passengers_on_board -= 1
            print(f"乘客{passenger_id}：已下车，当前车上剩余{self.passengers_on_board}人")
            # 如果所有乘客都下车了，通知车厢可以开始下一轮
            if self.passengers_on_board == 0:
                self.condition.notify()

    def run_ride(self):
        """车厢线程的主逻辑"""
        while True:
            with self.condition:
                # 等待乘客坐满车厢
                while self.passengers_on_board < self.capacity:
                    print("车厢：等待乘客中...")
                    self.condition.wait()
                
                # 启动过山车
                self.ride_in_progress = True
                print("=== 过山车出发！呼呼呼～ ===")
                self.condition.notify_all()  # 通知所有乘客行程开始
            
            # 模拟过山车运行时间（2秒）
            time.sleep(2)
            
            with self.condition:
                # 结束行程，通知乘客下车
                self.ride_in_progress = False
                print("=== 过山车到站啦！请乘客下车 ===")
                self.condition.notify_all()  # 通知所有乘客可以下车
                
                # 等待所有乘客下车
                while self.passengers_on_board > 0:
                    self.condition.wait()
                
                print("=== 所有乘客已下车，车厢准备迎接下一批乘客 ===")

def passenger_loop(roller_coaster, passenger_id):
    """乘客线程的循环逻辑：反复乘坐过山车"""
    while True:
        roller_coaster.board(passenger_id)
        roller_coaster.wait_for_ride_end(passenger_id)
        roller_coaster.alight(passenger_id)
        # 模拟乘客下车后的休息时间（1-3秒）
        time.sleep(random.randint(1, 3))

if __name__ == "__main__":
    NUM_PASSENGERS = 10  # 总乘客数
    CAR_CAPACITY = 4     # 车厢容量

    # 初始化过山车对象
    coaster = RollerCoaster(CAR_CAPACITY)

    # 启动车厢线程（守护线程，主程序退出时自动结束）
    coaster_thread = threading.Thread(target=coaster.run_ride)
    coaster_thread.daemon = True
    coaster_thread.start()

    # 启动所有乘客线程
    passenger_threads = []
    for i in range(NUM_PASSENGERS):
        thread = threading.Thread(target=passenger_loop, args=(coaster, i+1))
        thread.daemon = True
        passenger_threads.append(thread)
        thread.start()

    # 让模拟运行20秒后结束
    time.sleep(20)
    print("=== 模拟结束，感谢乘坐！===")

代码关键点说明： #

• 用threading.Condition实现线程同步，它内置锁和等待/通知机制，比单独用锁更灵活。
• 所有共享资源的操作都在with self.condition块中执行，确保线程安全，同时自动处理锁的获取和释放。
• 乘客上车后如果坐满车厢，会主动通知车厢线程发车；车厢到站后通知所有乘客下车，等所有人下车后再启动下一轮。
• 乘客下车后会随机休息一段时间，更贴近现实场景，避免所有乘客立刻再次排队。

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