为交通灯模拟代码添加队列与锁的实现需求

针对你的交通灯模拟需求，下面是添加了队列机制和锁保护的完整代码，解决多线程环境下的竞态问题，确保车辆按顺序通行：

import threading
import random
import time
import concurrent.futures

x_squares = 5
y_squares = 5
coordinate_dictionary = {}
for x in range(0, x_squares + 1):
    for y in range(0, y_squares + 1):
        coordinate_dictionary[(x, y)] = {
            "right_queue": [],
            "down_queue": [],
            "where_to_go": "right" if x % 3 == 0 else "down",
            # 用一把锁保护当前路口的所有状态（信号灯+队列），简化锁管理
            "lock": threading.Lock(),
            "x": x,
            "y": y
        }

def semaphore(structure):
    while not is_the_program_over:
        x, y = structure["x"], structure["y"]
        with structure["lock"]:
            print(f"信号灯 ({x},{y}) 切换通行方向: {structure['where_to_go']} → {'down' if structure['where_to_go'] == 'right' else 'right'}")
            # 切换通行方向
            structure["where_to_go"] = "down" if structure["where_to_go"] == "right" else "right"
        # 给车辆足够时间响应并通行
        time.sleep(random.randrange(3,7))

def cars(id):
    x, y = 0, 0
    # 随机选择行驶路线
    route = "down" if random.randrange(1,3) == 1 else "right"
    print(f"车辆 {id} 启动，路线: {route}")
    
    while x <= x_squares and y <= y_squares:
        current_pos = (x, y)
        print(f"车辆 {id} 到达路口 ({x},{y})，等待通行")
        structure = coordinate_dictionary[current_pos]
        
        # 1. 加入对应方向的队列
        with structure["lock"]:
            if route == "down":
                structure["down_queue"].append(id)
            else:
                structure["right_queue"].append(id)
        
        # 2. 等待自己排到队首且当前方向允许通行
        while True:
            with structure["lock"]:
                # 检查是否满足通行条件：方向匹配 + 自己是队列第一个
                if (route == structure["where_to_go"] and 
                    ((route == "down" and structure["down_queue"][0] == id) or 
                     (route == "right" and structure["right_queue"][0] == id))):
                    # 从队列移除
                    if route == "down":
                        structure["down_queue"].pop(0)
                    else:
                        structure["right_queue"].pop(0)
                    break
            # 避免忙等，释放锁后休眠
            time.sleep(0.5)
        
        # 3. 执行移动
        if route == "down":
            y += 1
        else:
            x += 1
        print(f"车辆 {id} 离开路口 ({current_pos[0]},{current_pos[1]})，到达 ({x},{y})")
        # 模拟行驶到下一个路口的时间
        time.sleep(1.5)
    
    print(f"车辆 {id} 完成路线，到达终点")

is_the_program_over = False
amount_of_cars = 25
amount_of_semaphors = len(coordinate_dictionary)

# 启动信号灯线程池
with concurrent.futures.ThreadPoolExecutor(max_workers=amount_of_semaphors) as semaphore_executor:
    semaphore_futures = [semaphore_executor.submit(semaphore, struct) for struct in coordinate_dictionary.values()]
    # 启动车辆线程池
    with concurrent.futures.ThreadPoolExecutor(max_workers=amount_of_cars) as car_executor:
        car_futures = [car_executor.submit(cars, i) for i in range(amount_of_cars)]
        # 等待所有车辆完成
        for future in concurrent.futures.as_completed(car_futures):
            try:
                future.result()
            except Exception as e:
                print(f"车辆线程出错: {e}")
    # 通知信号灯线程结束
    is_the_program_over = True
    # 等待所有信号灯线程结束
    for future in concurrent.futures.as_completed(semaphore_futures):
        try:
            future.result()
        except Exception as e:
            print(f"信号灯线程出错: {e}")

关键修改说明 #

• 锁机制优化：将原有的两个队列锁合并为一个路口锁，确保对信号灯状态和队列的操作都是原子性的，避免竞态条件。
• 队列管理：车辆到达路口时必须加入对应方向的队列，只有当自己是队首且当前信号灯允许该方向通行时，才能继续行驶。
• 避免忙等：车辆等待时会周期性释放锁并休眠，减少CPU占用，同时保证及时响应信号灯状态变化。
• 线程结束逻辑优化：确保所有车辆线程完成后，再通知信号灯线程终止，避免提前终止导致的异常。

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