当前位置:   article > 正文

Python高级——消息队列(Queue)与进程池(Pool)_python multiprocessing queue和pool

python multiprocessing queue和pool

Queue消息队列

1.创建

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

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 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()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40

执行结果:
这里写图片描述

Pool进程池

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

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

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()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30

运行结果:
这里写图片描述
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()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60

运行结果:
这里写图片描述

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/正经夜光杯/article/detail/860600
推荐阅读
相关标签
  

闽ICP备14008679号