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

import multiprocessing
queue = multiprocessing.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()

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

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()

# 导入模块
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()