Python多进程通信:使用Queue实现数据共享详解
本文详解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的使用方法。
相关代码如下:
# _*_ coding:utf-8 _*_
from multiprocessing import Queue
if __name__ == "__main__":
q = Queue(3)
q.put("消息1")
q.put("消息2")
print(q.full()) # 返回False
q.put("消息3")
print(q.full()) # 返回True
# 因为消息队列已满,下面的try都会抛出异常
# 第一个try会等待2秒再抛出异常,第二个try会立刻抛出异常
try:
q.put("消息4", True, 2)
except:
print("消息队列已满,现有消息数量:%s" % q.qsize())
try:
q.put_nowait("消息4")
except:
print("消息队列已满,现有消息数量:%s" % q.qsize())
# 读取消息时,先判断消息队列是否为空,再读取
if not q.empty():
print("-----从队列中获取消息-------")
for i in range(q.qsize()):
print(q.get_nowait())
# 先判读消息队列是否已满,再写入:
if not q.full():
q.put_nowait("消息4")程序运行结果如下图所示:

四、使用队列在进程间通信
我们知道,使用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("-------父进程结束-----------")程序运行结果如下:

到此,关于Python通过队列实现进程间通信的详细介绍已完毕。更多相关Python进程间通信内容,请搜索本站以前的文章或继续浏览下面的相关文章,希望大家以后多多支持本站!
免责声明:本文内容仅供参考,不代表本站观点。转载或引用请注明出处。









