Python通过信号量控制多进程并发的完整示例

引言

在 Python 的 multiprocessing 模块中，信号量（Semaphore） 是一把控制并发访问的“阀门”，跟互斥锁（Lock）那种“一夫当关”的风格不同——互斥锁一次只放行一个进程，信号量则允许指定数量的进程同时进入临界区。你可以把它想象成一个有固定数量钥匙的柜子：进程要干活，得先 acquire()（取走一把钥匙），干完活必须 release()（归还钥匙）。钥匙全部被取走时，后续进程就得排队等候。这个“钥匙”的数量，就是信号量允许的最大并发数。

核心API

• Semaphore(value=1)：创建一个信号量对象。value 就是初始的“钥匙”数量，即允许的最大并发进程数。
• acquire(blocking=True, timeout=None)：请求一把“钥匙”。
  • 如果计数器 > 0，计数器减1，进程继续执行。
  • 如果计数器 = 0，进程阻塞，直到有可用的“钥匙”。
• release()：归还一把“钥匙”，计数器加1。
• 上下文管理器：强烈推荐使用 with semaphore: 语句，它会自动帮你调用 acquire() 和 release()，即使代码抛出异常也能保证锁被释放，省心又安全。

关键注意事项

在多进程场景下，信号量的创建位置至关重要——这可能是新手最容易踩的坑。

• 正确做法：必须在主进程中创建信号量，然后作为参数传递给子进程。
• 错误做法：把信号量定义成全局变量。在 Windows 或 macOS（使用 'spawn' 方式创建进程）上，每个子进程都会重新导入模块，从而创建出各自的信号量副本，导致同步完全失效。这一点务必警惕。

完整示例：模拟限流打印任务

下面这个例子模拟了一个最多允许3个进程同时执行任务的场景，直观展示信号量的工作方式。

import multiprocessing
import time
import random

def worker(process_id, semaphore):
    """
    模拟一个工作进程。
    在进入临界区前，必须获取信号量。
    """
    # 使用 with 语句管理信号量，更安全
    with semaphore:
        # 进入临界区
        start_time = time.strftime("%H:%M:%S")
        print(f"[{start_time}] 进程 {process_id} 开始执行任务...")
        # 模拟一个耗时任务，耗时 1~3 秒
        work_duration = random.randint(1, 3)
        time.sleep(work_duration)
        end_time = time.strftime("%H:%M:%S")
        print(f"[{end_time}] 进程 {process_id} 任务完成 (耗时 {work_duration}秒)。")

if __name__ == "__main__":
    # 1. 在主进程中创建信号量，指定最大并发数为 3
    # 这就像创建了一个有 3 把钥匙的锁
    max_concurrent = 3
    semaphore = multiprocessing.Semaphore(max_concurrent)

    # 2. 创建并启动 10 个工作进程
    processes = []
    for i in range(10):
        # 重要：将信号量对象作为参数传递给子进程
        p = multiprocessing.Process(target=worker, args=(i, semaphore))
        processes.append(p)
        p.start()

    # 3. 等待所有子进程结束
    for p in processes:
        p.join()

    print("所有任务执行完毕。")

代码详解

1. 创建信号量：semaphore = multiprocessing.Semaphore(3) 创建了一个允许 3 个进程同时进入临界区的信号量。
2. 传递信号量：在创建 Process 时，将 semaphore 作为参数传入 worker 函数。这是确保所有进程共享同一个信号量的关键。
3. 控制并发：在 worker 函数中，with semaphore: 语句块内的代码就是临界区。任何时候，最多只有3个进程能同时执行其中的代码。
4. 模拟任务：time.sleep() 模拟了进程的实际工作。

可能的输出结果（部分）

[14:23:01] 进程 0 开始执行任务...
[14:23:01] 进程 1 开始执行任务...
[14:23:01] 进程 2 开始执行任务...
[14:23:03] 进程 1 任务完成 (耗时 2秒)。
[14:23:03] 进程 3 开始执行任务...   # 进程1释放信号量，进程3立刻开始
[14:23:04] 进程 2 任务完成 (耗时 3秒)。
[14:23:04] 进程 4 开始执行任务...
...

从输出中可以清晰看到，开始执行任务 的消息不会连续出现超过3次。每当一个进程完成并释放信号量后，下一个等待的进程便会立即获取并开始执行。这就是信号量“限流”的直观体现。

信号量 vs 进程池 (multiprocessing.Pool)

这两种方式都能控制并发，但适用场景各有侧重：

• Semaphore：更灵活，可以精确控制代码中任意一个代码块的并发数，而不仅仅是整个任务函数。
• Pool：更高层，主要用于管理一组同质任务的并发执行。

如果你的需求是“同时最多运行N个任务”，Pool 是更简单的选择。如果你的需求是“在任务执行过程中，某个特定资源的访问并发数不能超过N”，那么 Semaphore 才是更合适的工具。

总结

1. 核心思想：信号量通过一个计数器来控制对共享资源的并发访问数量。
2. 关键用法：在 if __name__ == "__main__": 块中创建 multiprocessing.Semaphore(N)，并作为参数传递给子进程。
3. 最佳实践：使用 with semaphore: 语句管理信号量的获取和释放，避免死锁。
4. 适用场景：限制对有限资源（如数据库连接、网络带宽、特定硬件）的并发访问。

以上就是Python通过信号量控制多进程并发的完整示例。希望这个指南能帮你少踩坑，更优雅地管理并发资源。