Python中消息队列与进程池的示例分析

这篇文章主要介绍了Python中消息队列与进程池的示例分析，具有一定借鉴价值，感兴趣的朋友可以参考下，希望大家阅读完这篇文章之后大有收获，下面让小编带着大家一起了解一下。

Queue消息队列

1.创建

import multiprocessing queue = multiprocessing.Queue(队列长度)

2.方法

3.进程通信

因为进程间不共享全局变量，所以使用Queue进行数据通信，可以在父进程中创建两个字进程，一个往Queue里写数据，一个从Queue里取出数据。
例：

import multiprocessing import time def write_queue(queue):   # 循环写入数据   for i in range(10):     if queue.full():       print("队列已满!")       break     # 向队列中放入消息     queue.put(i)     print(i)     time.sleep(0.5) def read_queue(queue):   # 循环读取队列消息   while True:     # 队列为空，停止读取     if queue.empty():       print("---队列已空---")       break     # 读取消息并输出     result = queue.get()     print(result) if __name__ == '__main__':   # 创建消息队列   queue = multiprocessing.Queue(3)   # 创建子进程   p1 = multiprocessing.Process(target=write_queue, args=(queue,))   p1.start()   # 等待p1写数据进程执行结束后，再往下执行   p1.join()   p1 = multiprocessing.Process(target=read_queue, args=(queue,))   p1.start()

执行结果：

Pool进程池

初始化Pool时，可以指定一个最大进程数，当有新的请求提交到Pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到指定的最大值，那么该请求就会等待，直到池中有进程结束，才会用之前的进程来执行新的任务。

1.创建

import multiprocessing pool = multiprocessing.Pool(最大进程数)

2.方法

3.进程池内通信

创建进程池内Queue消息队列通信

import multiprocessing  Queue：queue = multiprocessing.Manager().Queue()

例：

import multiprocessing  import time

写入数据的方法

def write_data(queue):  # for循环 向消息队列中写入值  for i in range(5):  # 添加消息  queue.put(i)  print(i)  time.sleep(0.2)  print("队列已满~")

创建读取数据的方法

def read_data(queue):   # 循环读取数据   while True:     # 判断队列是否为空     if queue.qsize() == 0:       print("队列为空～")       break     # 从队列中读取数据     result = queue.get()     print(result) if __name__ == '__main__':   # 创建进程池   pool = multiprocessing.Pool(2)   # 创建进程池队列   queue = multiprocessing.Manager().Queue()   # 在进程池中的进程间进行通信   # 使用线程池同步的方式，先写后读   # pool.apply(write_data, (queue, ))   # pool.apply(read_data, (queue, ))   # apply_async() 返回ApplyResult 对象   result = pool.apply_async(write_data, (queue, ))   # ApplyResult对象的wait() 方法，表示后续进程必须等待当前进程执行完再继续   result.wait()   pool.apply_async(read_data, (queue, ))   pool.close()   # 异步后，主线程不再等待子进程执行结束，再结束   # join() 后，表示主线程会等待子进程执行结束后，再结束   pool.join()

运行结果：

4.案例(文件夹copy器)

代码：

# 导入模块 import os import multiprocessing # 拷贝文件函数 def copy_dir(file_name, source_dir, desk_dir):   # 要拷贝的文件路径   source_path = source_dir+'/'+file_name   # 目标路径   desk_path = desk_dir+'/'+file_name   # 获取文件大小   file_size = os.path.getsize(source_path)   # 记录拷贝次数   i = 0   # 以二进制度读方式打开原文件   with open(source_path, "rb") as source_file:     # 以二进制写入方式创建并打开目标文件     with open(desk_path, "wb") as desk_file:       # 循环写入       while True:         # 读取1024字节         file_data = source_file.read(1024)         # 如果读到的不为空，则将读到的写入目标文件         if file_data:           desk_file.write(file_data)           # 读取次数+1           i += 1           # 拷贝百分比进度等于拷贝次数*1024*100/文件大小           n = i*102400/file_size           if n >= 100:             n = 100           print(file_name, "拷贝进度%.2f%%" % n)         else:           print(file_name, "拷贝成功")           break if __name__ == '__main__':   # 要拷贝的文件夹   source_dir = 'test'   # 要拷贝到的路径   desk_dir = 'C:/Users/Administrator/Desktop/'+source_dir   # 存在文件夹则不创建   try:     os.mkdir(desk_dir)   except:     print("目标文件夹已存在，未创建")   # 获取文件夹内文件目录，存到列表里   file_list = os.listdir(source_dir)   print(file_list)   # 创建进程池，最多同时运行3个子进程   pool = multiprocessing.Pool(3)   for file_name in file_list:     # 异步方式添加到进程池内     pool.apply_async(copy_dir, args=(file_name, source_dir, desk_dir))   # 关闭进程池(停止添加，已添加的还可运行)   pool.close()   # 让主进程阻塞，等待子进程结束   pool.join()

运行结果：

感谢你能够认真阅读完这篇文章，希望小编分享的“Python中消息队列与进程池的示例分析”这篇文章对大家有帮助，同时也希望大家多多支持亿速云，关注亿速云行业资讯频道，更多相关知识等着你来学习!