Python进程间通信Queue实例解析

周予安

发布时间：2026-10-13| 2人浏览过

原创

本文详解Python中Queue在进程间通信的应用，涵盖qsize、get、put等核心方法。通过实例演示普通进程与进程池下Queue的使用差异，重点解析Pool场景需使用Manager.Queue的原因及潜在资源混乱问题，帮助开发者掌握正确的队列通信模式。

本文主要探讨Python中进程间通信Queue的具体应用实例，内容如下。

1.Queue常用方法：

Queue.qsize()：用于返回当前队列中包含的消息数量；
Queue.empty()：若队列为空则返回True，否则返回False；
Queue.full()：若队列已满则返回True，否则返回False；
Queue.get()：从队列中获取一条消息并将其移除，该方法支持传入超时时长参数；
Queue.get_nowait()：等效于Queue.get(False)，当无法立即获取值时会触发Empty异常；
Queue.put()：将一个值添加到队列中，支持传入超时时长参数；
Queue.put_nowait()：等效于Queue.put(False)，当队列已满时会报错Full。

2.Queue使用示例：

具体代码实现如下：

#!/usr/bin/env python3

import time
from multiprocessing import Process,Queue

q = Queue() #创建列队，不传数字表示列队不限数量
for i in range(11):
  q.put(i)

def A():
  while 1:
    try:
      num = q.get_nowait()
      print('我是进程A,取出数字:%d'%num)
      time.sleep(1)
    except :
      break

def B():
  while 1:
    try:
      num = q.get_nowait()
      print('我是进程B,取出数字:%d'%num)
      time.sleep(1)
    except :
      break

p1 = Process(target = A)
p2 = Process(target = B)
p1.start()
p2.start()

该程序向队列中插入10个数字，随后使用2个进程进行读取操作。

运行结果如下：

我是进程A,取出数字:0 我是进程B,取出数字:1 我是进程A,取出数字:2 我是进程B,取出数字:3 我是进程A,取出数字:4 我是进程B,取出数字:5 我是进程B,取出数字:6 我是进程A,取出数字:7 我是进程B,取出数字:8 我是进程A,取出数字:9 我是进程B,取出数字:10

3.在使用进程池Pool时，直接使用Queue可能会出错，此时应使用Manager.Queue：

相关代码如下

#!/usr/bin/env python3

import time
from multiprocessing import Pool,Manager,Queue

q = Manager().Queue()
for i in range(11):
  q.put(i)

def A(i):
  num = q.get_nowait()
  print('我是进程%d,取出数字:%d'%(i,num))
  time.sleep(1)
      

pool = Pool(3)

for i in range(10):
  pool.apply_async(A,(i,))

pool.close()
pool.join()

运行结果：

我是进程1,取出数字:0 我是进程0,取出数字:1 我是进程2,取出数字:2 我是进程4,取出数字:3 我是进程3,取出数字:4 我是进程5,取出数字:5 我是进程6,取出数字:6 我是进程7,取出数字:7 我是进程8,取出数字:8 我是进程9,取出数字:9

若将Manager().Queue()直接替换为Queue()，可能会导致资源混乱或进程缺失。

4.主进程定义了一个Queue类型的变量，并将其作为Process的args参数传递给子进程processA和processB，其中两个进程分别负责向队列写入数据和读取数据。

import time
from multiprocessing import Process,Queue

MSG_QUEUE = Queue(5)

def startA(msgQueue):
  while True:
    if msgQueue.empty() > 0:
      print 'queue is empty %d' % (msgQueue.qsize())
    else:
      msg = msgQueue.get()
      print 'get msg %s' % (msg,)
    time.sleep(1)

def startB(msgQueue):
  while True:
    msgQueue.put('hello world')
    print 'put hello world queue size is %d' % (msgQueue.qsize(),)
    time.sleep(3)

if __name__ == '__main__':
  processA = Process(target=startA,args=(MSG_QUEUE,))
  processB = Process(target=startB,args=(MSG_QUEUE,))

  processA.start()
  print 'processA start..'

  processB.start()
  print 'processB start..'

其打印结果如下：

C:\Python27\python.exe E:/outofmemory/test/queuetest/queuetest.py processA start.. processB start.. queue is empty 0 put hello world queue size is 1 get msg hello world queue is empty 0 queue is empty 0 put hello world queue size is 1 get msg hello world queue is empty 0 queue is empty 0 put hello world queue size is 1

总结

以上就是本文关于Python进程间通信Queue实例解析的全部内容，希望对大家有所帮助。感兴趣的朋友可以继续参阅本站其他相关专题，如有不足之处，欢迎留言指出。感谢朋友们对本站的支持！