Python中信号量控制主线程与子线程互斥运行的优化问询

你需要实现的逻辑是：主线程创建n个子线程执行循环任务，主线程休眠期间子线程正常运行；仅当所有子线程都进入锁定/等待状态时，主线程才执行自身代码，执行完成后唤醒子线程继续运行。原方案用Semaphore实现，但主线程需要循环n次调用sem.acquire()，以下是更优的替代方案：

方案一：使用Condition变量+计数器（推荐） #

这种方式通过条件变量同步主线程与子线程，用计数器跟踪等待中的子线程数量，无需循环获取Semaphore许可，逻辑直观清晰。

import threading
import time

def main_thread(n):
    cond = threading.Condition()
    waiting_thread_count = 0

    def thread_fun():
        nonlocal waiting_thread_count
        while True:
            # 执行子线程任务
            print(f"子线程 {threading.get_ident()} 执行任务中...")
            time.sleep(1)  # 模拟任务耗时

            # 任务完成后进入等待状态
            with cond:
                waiting_thread_count += 1
                cond.notify()  # 通知主线程：有子线程进入等待
                cond.wait()    # 等待主线程唤醒

    # 启动n个子线程（设为守护线程，避免主线程退出后子线程残留）
    for _ in range(n):
        threading.Thread(target=thread_fun, daemon=True).start()

    while True:
        time.sleep(60)  # 主线程休眠固定时长
        print("\n主线程准备执行操作，等待所有子线程进入等待状态...")

        with cond:
            # 等待所有子线程都进入等待
            while waiting_thread_count < n:
                cond.wait()
            
            # 所有子线程已锁定，执行主线程核心逻辑
            print("所有子线程已处于等待状态，主线程执行核心操作...")
            # --- 这里写入主线程需要执行的代码 ---
            time.sleep(2)  # 模拟主线程任务耗时

            # 重置计数器，唤醒所有子线程继续运行
            waiting_thread_count = 0
            cond.notify_all()
        print("主线程操作完成，子线程恢复运行\n")

if __name__ == "__main__":
    main_thread(3)

方案优势 #

• 无需循环调用acquire()，通过计数器精准判断所有子线程的状态
• 逻辑直接对应需求：子线程任务完成后主动进入等待，主线程确认所有子线程等待后再执行
• 避免Semaphore方案中可能出现的竞态问题（比如主线程获取许可时，子线程刚好释放许可导致的重复等待）

方案二：使用Barrier实现批量同步 #

如果你的场景允许子线程和主线程在固定屏障点同步，可以使用threading.Barrier，它会等待指定数量的线程到达屏障点后再一起继续执行。

import threading
import time

def main_thread(n):
    # 屏障设置为n+1：n个子线程 + 主线程
    barrier = threading.Barrier(n + 1)

    def thread_fun():
        while True:
            # 执行子线程任务
            print(f"子线程 {threading.get_ident()} 执行任务中...")
            time.sleep(1)  # 模拟任务耗时

            # 到达屏障点，等待主线程
            barrier.wait()
            # 主线程到达后，屏障释放，子线程继续下一轮任务

    # 启动子线程
    for _ in range(n):
        threading.Thread(target=thread_fun, daemon=True).start()

    while True:
        time.sleep(60)
        print("\n主线程准备执行操作，等待所有子线程到达屏障...")
        # 主线程到达屏障，此时所有子线程已在等待
        barrier.wait()
        
        # 执行主线程核心逻辑
        print("所有子线程已同步，主线程执行操作...")
        time.sleep(2)  # 模拟任务耗时
        
        print("主线程操作完成，子线程恢复运行\n")

if __name__ == "__main__":
    main_thread(3)

注意事项 #

• Barrier要求所有参与的线程（包括主线程）都调用wait()才能继续，适合需要双向同步的场景
• 如果有线程异常退出导致Barrier无法集齐足够数量的线程，会触发BrokenBarrierError，需要额外处理异常

原方案的问题总结 #

原Semaphore方案中，主线程循环n次acquire()本质是抢占所有许可迫使子线程阻塞，存在以下问题：

• 当n很大时，循环调用会增加不必要的代码复杂度
• 存在竞态风险：如果子线程在主线程获取许可的过程中释放许可，可能导致主线程需要多次等待才能集齐所有许可
• 逻辑不够直观，无法直接体现“等待所有子线程锁定”的业务意图

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