当前位置：首页 > news >正文

Python多进程(multiprocessing)(mp)(三) —— 进程间通信 Queue队列(Manager)

news 来源：原创 2024/5/19 5:49:46

进程间通信

全局变量在多个进程中不共享，进程之间的数据是独立的，默认情况下互不影响
from multiprocessing import Process

num=1

def work1():
	global num
	num+=5
	print('子进程 1 运行， num:',num)

def work2():
	global num
	num += 10
	print('子进程 2 运行， num： ',num)

if __name__=='__main__':
	print('父进程开始运行')
	p1=Process(target=work1)
	p2=Process(target=work2)
	p1.start()
	p2.start()
	p1.join()
	p2.join()
用 Queue 实现多进程之间的数据传递

Queue 是多进程安全的队列，可以使用 Queue 实现多进程之间的数据传递。

put 方法用以插入数据到队列中， put 方法还有两个可选参数： blocked 和 timeout。如果 blocked 为 True（默认值），并且 timeout 为正值，该方法会阻塞 timeout 指定的时间，直到该队列有剩余的空间。如果超时，会抛出 Queue.full 异常。如果 blocked 为 False，但该 Queue 已满，会立即抛出 Queue.full 异常。

get 方法可以从队列读取并且删除一个元素。同样， get 方法有两个可选参数： blocked和 timeout。如果 blocked 为 True（默认值），并且 timeout 为正值，那么在等待时间内没有取到任何元素，会抛出 Queue.Empty 异常。如果 blocked 为 False，有两种情况存在，如果Queue 有一个值可用，则立即返回该值，否则，如果队列为空，则立即抛出Queue.Empty 异常
from multiprocessing import Queue

q=Queue(3)
q.put('消息 1')
q.put('消息 2')
print('消息队列是否已满： ',q.full())
q.put('消息 3')
print('消息队列是否已满： ',q.full())
#q.put('消息 4')因为消息队列已满，需要直接写入需要等待，如果超时会抛出异常，

#所以写入时候需判断，消息队列是否已满
if not q.full():
	q.put('消息 4')

#同理读取消息时，先判断消息队列是否为空，再读取
if not q.empty():
	for i in range(q.qsize()):
		print(q.get())
Queue 队列实现进程间通信
from multiprocessing import *
import time

def write(q):
	#将列表中的元素写入队列中
	for i in ["a","b","c"]:
		print('开始写入值%s' % i)
		q.put(i)
		time.sleep(1)

#读取
def read(q):
	print('开始读取')
	while True:
		if not q.empty():
			print('读取到:',q.get())
			time.sleep(1)
		else:
			break

if __name__=='__main__':
	#创建队列
	q=Queue()
	#创建写入进程
	pw=Process(target=write,args=(q,))
	pr=Process(target=read,args=(q,))
	#启动进程
	pw.start()
	pw.join()
	pr.start()
	pr.join()
如果使用 Pool 创建进程，就需要使用 multiprocessing.Manager()中的 Queue()来完成进程间的通信，而不是 multiprocessing.Queue()，否则会抛出异常
from multiprocessing import Manager,Pool
import time

def write(q):
	#将列表中的元素写入队列中
	for i in ["a","b","c"]:
		print('开始写入值%s' % i)
		q.put(i)
		time.sleep(1)

#读取
def read(q):
	print('开始读取')
	while True:
		if not q.empty():
			print('读取到:',q.get())
			time.sleep(1)
		else:
			break

if __name__=='__main__':
	#创建队列
	q=Manager().Queue()
	#创建进程池
	p=Pool(3)
	#使用apply阻塞模式创建进程
	p.apply(write,(q,))
	p.apply(read,(q,))
	p.close()
	p.join()