Python多进程通信：使用Queue实现数据共享详解

巷口听风

发布时间：2026-10-15 | 5人浏览过

原创

本文详解Python中如何利用multiprocessing模块的Queue实现多进程间的数据通信。通过对比全局变量无法共享的特性，引出Queue作为消息队列的核心作用，并分步解析Queue的初始化、常用方法（如get/put及其阻塞参数）及实际代码示例，帮助开发者掌握进程间高效数据交换技巧。

一、前言

在多进程环境下，各进程相互独立，各自拥有独立的地址空间、内存、数据栈及记录运行状态的辅助数据。为了直观理解进程间的数据隔离性，我们可以通过一个具体案例来验证进程间是否可以直接共享信息。

首先定义一个全局变量g_num，随后创建两个子进程，分别对g_num执行不同的数值操作，并打印操作后的结果以进行观察。

具体实现代码如下：

# _*_ coding:utf-8 _*_
from multiprocessing import Process
def plus():
    print("-------子进程1开始----------")
    global g_num
    g_num += 50
    print("g_num is %d" % g_num)
    print("-------子进程1结束----------")
def minus():
    print("-------子进程2开始----------")
    global g_num
    g_num -= 50
    print("g_num is %d" % g_num)
    print("-------子进程2结束----------")
g_num = 100  # 定义一个全局变量
if __name__ == "__main__":
    print("-------主进程开始----------")
    print("g_num is %d" % g_num)
    p1 = Process(target=plus)  # 实例化进程p1
    p2 = Process(target=minus)  # 实例化进程p2
    p1.start()  # 开启p1进程
    p2.start()  # 开启p2进程
    p1.join()  # 等待p1进程结束
    p2.join()  # 等待p2进程结束
    print("-------主进程结束----------")

程序运行后的输出结果如下图所示：

从上述代码逻辑来看，虽然创建了两个子进程，一个将g_num加50，另一个将g_num减50，但运行结果显示，父进程与两个子进程中的g_num初始值均为100。这表明全局变量g_num在父进程中的修改并未同步到子进程，反之亦然，即进程间默认不共享全局变量信息。

各进程间的内存关系示意图如下：

既然全局变量无法共享，如何实现进程间通信呢？Python的multiprocessing模块封装了底层机制，提供了Queue（队列）和Pipes（管道）等多种数据交换方式。本文将重点讲解如何利用Queue实现进程间通信。

二、队列简介

队列（Queue）的概念类似于现实生活中的排队场景，例如学生在食堂打饭。新来的学生排在队伍末尾，而排在最前面的学生打完饭后离开，后续学生依次向前移动。

通过上述场景可以看出队列具备两个核心特征：

新加入的元素始终排在队尾。
位于队首的元素被处理并移除后，下一个元素成为新的队首。

基于上述特性，队列的数据结构模型如下图所示：

三、多进程队列的使用

在多进程开发中，进程间通信是常见需求。操作系统提供了多种机制，而在Python中，可以使用multiprocessing模块提供的Queue来实现多进程间的数据传递。Queue本质上是一个消息队列程序，下面详细介绍其用法。

在初始化Queue对象时（例如q=Queue(num)），如果括号内未指定最大消息数量或指定为负值，则表示消息数量无上限，直到耗尽系统内存为止。

Queue类提供了以下常用方法：

Queue.qsize()：返回当前队列中包含的消息数量。Queue.empty()：若队列为空返回True，否则返回False。Queue.full()：若队列已满返回True，否则返回False。Queue.get(block[,timeout])：从队列中获取一条信息并将其移除，block参数默认值为True。

当block使用默认值（True）且未设置timeout（单位：秒）时，若消息队列为空，程序将被阻塞（处于读取等待状态），直到从队列中读取到消息为止。若设置了timeout，则程序将等待timeout秒，若超时仍未读取到消息，则抛出“Queue.Empty”异常。

若block值为False，且消息队列为空，程序将立即抛出“Queue.Empty”异常。

Queue.get_nowait()：功能等同于Queue.get(False)。Queue.put(item,[block[,timeout]])：将item消息写入队列，block参数默认值为True。

当block使用默认值（True）且未设置timeout时，若消息队列已满且无空间写入，程序将被阻塞（处于写入等待状态），直到队列腾出空间。若设置了timeout，则程序将等待timeout秒，若超时仍无空间，则抛出“Queue.Full”异常。

若block值为False，且消息队列已满无空间可写入，程序将立即抛出“Queue.Full”异常。

Queue.put_nowait(item)：功能等同于Queue.put(item, False)。

接下来，通过一个具体示例来学习processing.Queue的使用方法。

四、使用队列在进程间通信

我们知道，使用multiprocessing.Process可以创建多进程，而使用multiprocessing.Queue可以实现队列操作。接下来，通过一个结合Process和Queue的示例，演示如何实现进程间通信。

创建两个子进程，其中一个负责向队列写入数据，另一个负责从队列读取数据。为了确保能正确读取数据，设置读取进程的等待时间为2秒。若2秒后仍无法读取到数据，则抛出异常。

实现代码如下：

# _*_ coding:utf-8 _*_
from multiprocessing import Process, Queue
import time
# 向队列中写入数据
def write_task(q):
    if not q.full():
        for i in range(5):
            message = "消息" + str(i)
            q.put(message)
            print("写入：%s" % message)
# 从队列中读取数据
def read_task(q):
    time.sleep(1)  # 休眠1秒
    while not q.empty():
        print("读取：%s" % q.get(True, 2))  # 等待2秒中，如果没有读取到任何信息，则抛出异常
if __name__ == "__main__":
    print("--------父进程开始---------")
    q = Queue()  # 父进程创建Queue，并传给各个子进程
    pw = Process(target=write_task, args=(q,))  # 实例化写入队列的子进程，并传递给队列
    pr = Process(target=read_task, args=(q,))  # 实例化读取队列的子进程，并传递给队列
    pw.start()  # 启动子进程pw，写入
    pr.start()  # 启动子进程pr，读取
    pw.join()  # 等待pw结束
    pr.join()  # 等待pr结束
    print("-------父进程结束-----------")

程序运行结果如下：