Python多进程同步与通信机制详解：Lock、Queue、Pipe、Semaphore及Event

蓝鲸

发布时间：2026-10-12 | 4人浏览过

原创

本文深入解析Python多进程开发中的同步与通信机制。详细讲解如何利用Lock解决资源竞争，通过Queue实现多进程安全消息传递，使用Pipe进行双向通信，借助Semaphore控制并发数量，以及利用Event实现进程间同步。涵盖核心API用法、参数说明及代码示例，帮助开发者掌握多进程协作的关键技术。

前文已阐述进程的基本概念、进程与程序的关联、进程创建及进程池的使用，本节将重点介绍多进程环境下的同步控制与通信方式。

进程同步

当多个进程共同访问同一份数据资源时，由于进程执行顺序不可控，若不采取干预措施，极易引发数据不一致或顺序混乱。因此，在多线程或多进程读写共享资源时，必须引入适当的同步策略，以确保数据的一致性。

Lock（锁）

Lock对象提供acquire()和release()两个方法，用于控制对共享数据的读写权限。在多进程场景中，常出现多个进程争抢输出资源的情况，例如共享打印终端导致输出信息交错错乱，如上图所示。

此时可引入Lock机制：

import multiprocessing
import random
import time
def work(lock, i):
    lock.acquire()
    print("work'{}'执行中......".format(i), multiprocessing.current_process().name, multiprocessing.current_process().pid)
    time.sleep(random.randint(0, 2))
    print("work'{}'执行完毕......".format(i))
    lock.release()
if __name__ == '__main__':
    lock = multiprocessing.Lock()
    for i in range(5):
        p = multiprocessing.Process(target=work, args=(lock, i))
        p.start()

引入Lock后，同一时刻仅允许一个进程获取输出资源，其他进程需等待当前进程执行完毕并释放锁后，方可继续抢占。这解决了多进程资源竞争导致的数据错乱问题，但将并发执行转变为串行执行，从而牺牲了部分运行效率。

进程通信

前文提及，进程间数据默认相互隔离，互不影响。那么如何实现进程间的数据交换？Python提供了多种进程通信机制，下文将逐一介绍。

Queue（队列）

通过multiprocessing模块提供的Queue类，可以创建多进程安全的消息队列，从而实现多进程间的数据传递。

说明

初始化Queue对象时（如q=Queue()），若未指定最大消息数量或设为负值，则表示消息数量无上限（受限于内存大小）。
Queue.qsize()：返回当前队列中包含的消息数量。
Queue.empty()：若队列为空返回True，否则返回False。
Queue.full()：若队列已满返回True，否则返回False。
Queue.get(block, timeout)：从队列中获取一条消息并将其移除，block参数默认为True。若使用默认值且未设置timeout（秒），当队列为空时程序将阻塞直至获取到消息；若设置了timeout，则等待指定秒数，超时未获取到消息则抛出Queue.Empty异常；若block设为False，队列为空时将立即抛出Queue.Empty异常。
Queue.get_nowait()：等效于Queue.get(False)。
Queue.put(item, block, timeout)：将item消息写入队列，block参数默认为True。若使用默认值且未设置timeout，当队列已满无法写入时程序将阻塞直至腾出空间；若设置了timeout，则等待指定秒数，超时仍无空间则抛出Queue.Full异常；若block设为False，队列已满时将立即抛出Queue.Full异常。
Queue.put_nowait(item)：等效于Queue.put(item, False)。

from multiprocessing import Process, Queue
import time
def write_task(queue):
    """
    向队列中写入数据
    :param queue: 队列
    :return:
    """
    for i in range(5):
        if queue.full():
            print("队列已满!")
        message = "消息{}".format(str(i))
        queue.put(message)
        print("消息{}写入队列".format(str(i)))
def read_task(queue):
    """
    从队列读取数据
    :param queue: 队列
    :return:
    """
    while True:
        print("从队列读取:{}".format(queue.get(True)))
if __name__ == '__main__':
    print("主进程执行......")
    # 主进程创建Queue，最大消息数量为3
    queue = Queue(3)
    pw = Process(target=write_task, args=(queue, ))
    pr = Process(target=read_task, args=(queue, ))
    pw.start()
    pr.start()

运行结果为：

从结果可见，队列最大容量为3条消息，后续写入的消息需等待read_task从队列中取出后才能继续写入。

Pipe（管道）

Pipe通常用于两个进程间的通信，两端分别位于管道的不同位置。Pipe(duplex)方法返回(conn1, conn2)代表管道的两端，duplex参数默认为True（全双工模式）；若设为False，则conn1仅负责接收，conn2仅负责发送。

send()和recv()方法分别用于发送和接收消息。

import multiprocessing
import time
import random
def proc_send(pipe):
    """
    发送消息
    :param pipe:管道一端
    :return:
    """
    for i in range(10):
        print("process send:{}".format(str(i)))
        pipe.send(i)
        time.sleep(random.random())
def proc_recv(pipe):
    """
    接收消息
    :param pipe:管道一端
    :return:
    """
    while True:
        print("Process recv:{}".format(pipe.recv()))
        time.sleep(random.random())
if __name__ == '__main__':
    # 主进程创建pipe
    pipe = multiprocessing.Pipe()
    p1 = multiprocessing.Process(target=proc_send,args=(pipe[0], ))
    p2 = multiprocessing.Process(target=proc_recv,args=(pipe[1], ))
    p1.start()
    p2.start()
    p1.join()
    p2.terminate()

执行结果为：

Semaphore（信号量）

Semaphore用于控制对共享资源的访问数量，其概念类似于进程池的最大连接数限制。

import multiprocessing
import random
import time
def work(s, i):
    s.acquire()
    print("work'{}'执行中......".format(i), multiprocessing.current_process().name, multiprocessing.current_process().pid)
    time.sleep(i*2)
    print("work'{}'执行完毕......".format(i))
    s.release()
if __name__ == '__main__':
    s = multiprocessing.Semaphore(2)
    for i in range(1, 7):
        p = multiprocessing.Process(target=work, args=(s, i))
        p.start()

上述代码中，Semaphore限制了最多2个进程同时执行。每有一个进程获得锁，计数加1；当计数达到2时，后续进程需等待前序进程释放锁后方可获取。

信号量在概念上与进程池类似，但需明确区分：信号量涉及具体的加锁机制。

Event（事件）

Event用于实现进程间的同步通信。其运行机制为：全局定义一个flag标志，若flag为False，程序执行event.wait()时会阻塞；若flag为True，执行event.wait()时不会阻塞，继续执行。

Event常用函数：

event.wait()：在进程中插入一个标记（flag），默认为False，可设置timeout。
event.set()：将flag设置为True。
event.clear()：将flag设置为False。
event.is_set()：判断flag是否为True。

import multiprocessing
import time
def wait_for_event(e):
    print("wait_for_event执行")
    e.wait()
    print("wait_for_event: e.is_set():{}".format(e.is_set()))
def wait_for_event_timeout(e, t):
    print("wait_for_event_timeout执行")
    # 只会阻塞2s
    e.wait(t)
    print("wait_for_event_timeout:e.is_set:{}".format(e.is_set()))
if __name__ == "__main__":
    e = multiprocessing.Event()
    p1 = multiprocessing.Process(target=wait_for_event, args=(e,))
    p1.start()
    p2 = multiprocessing.Process(target=wait_for_event_timeout, args=(e, 2))
    p2.start()
    time.sleep(4)
    # 4s之后使用e.set()将flag设为Ture
    e.set()
    print("主进程:flag设置为True")

执行结果如下：