Python高级编程之消息队列(Queue)与进程池(Pool)实例详解

2019-11-25 11:29:47

字体：大中小

来源：转载

供稿：网友

本文实例讲述了Python高级编程之消息队列(Queue)与进程池(Pool)。分享给大家供大家参考，具体如下：

Queue消息队列

1.创建

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

2.方法

方法	描述
put	变量名.put(数据)，放入数据(如队列已满，则程序进入阻塞状态，等待队列取出后再放入)
put_nowait	变量名.put_nowati(数据)，放入数据(如队列已满，则不等待队列信息取出后再放入，直接报错)
get	变量名.get(数据)，取出数据(如队列为空，则程序进入阻塞状态，等待队列防如数据后再取出)
get_nowait	变量名.get_nowait(数据)，取出数据(如队列为空，则不等待队列放入信息后取出数据，直接报错)，放入数据后立马判断是否为空有时为True，原因是放入值和判断同时进行
qsize	变量名.qsize()，消息数量
empty	变量名.empty()(返回值为True或False)，判断是否为空
full	变量名.full()(返回值为True或False)，判断是否为满

3.进程通信

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

import multiprocessingimport timedef 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 multiprocessingpool = multiprocessing.Pool(最大进程数)

2.方法

方法	描述
apply()	以同步方式添加进程
apply_async()	以异步方式添加进程
close()	关闭Pool，使其不接受新任务（还可以使用）
terminate()	不管任务是否完成，立即终止
join()	主进程阻塞，等待子进程的退出，必须在close和terminate后使用

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 osimport 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, "拷贝成功")          breakif __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()

运行结果：