Python高级编程之消息队列(Queue)与进程池(Pool)实例详解
本文实例讲述了Python高级编程之消息队列(Queue)与进程池(Pool)。分享给大家供大家参考,具体如下:
Queue消息队列
1.创建
importmultiprocessing queue=multiprocessing.Queue(队列长度)
2.方法
3.进程通信
因为进程间不共享全局变量,所以使用Queue进行数据通信,可以在父进程中创建两个字进程,一个往Queue里写数据,一个从Queue里取出数据。
例:
importmultiprocessing importtime defwrite_queue(queue): #循环写入数据 foriinrange(10): ifqueue.full(): print("队列已满!") break #向队列中放入消息 queue.put(i) print(i) time.sleep(0.5) defread_queue(queue): #循环读取队列消息 whileTrue: #队列为空,停止读取 ifqueue.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.创建
importmultiprocessing pool=multiprocessing.Pool(最大进程数)
2.方法
3.进程池内通信
创建进程池内Queue消息队列通信
importmultiprocessing Queue:queue=multiprocessing.Manager().Queue()
例:
importmultiprocessing importtime
写入数据的方法
defwrite_data(queue): #for循环向消息队列中写入值 foriinrange(5): #添加消息 queue.put(i) print(i) time.sleep(0.2) print("队列已满~")
创建读取数据的方法
defread_data(queue): #循环读取数据 whileTrue: #判断队列是否为空 ifqueue.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器)
代码:
#导入模块 importos importmultiprocessing #拷贝文件函数 defcopy_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 #以二进制度读方式打开原文件 withopen(source_path,"rb")assource_file: #以二进制写入方式创建并打开目标文件 withopen(desk_path,"wb")asdesk_file: #循环写入 whileTrue: #读取1024字节 file_data=source_file.read(1024) #如果读到的不为空,则将读到的写入目标文件 iffile_data: desk_file.write(file_data) #读取次数+1 i+=1 #拷贝百分比进度等于拷贝次数*1024*100/文件大小 n=i*102400/file_size ifn>=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) forfile_nameinfile_list: #异步方式添加到进程池内 pool.apply_async(copy_dir,args=(file_name,source_dir,desk_dir)) #关闭进程池(停止添加,已添加的还可运行) pool.close() #让主进程阻塞,等待子进程结束 pool.join()
运行结果:
更多关于Python相关内容感兴趣的读者可查看本站专题:《Python进程与线程操作技巧总结》、《Python数据结构与算法教程》、《Python函数使用技巧总结》、《Python字符串操作技巧汇总》、《Python入门与进阶经典教程》、《Python+MySQL数据库程序设计入门教程》及《Python常见数据库操作技巧汇总》
希望本文所述对大家Python程序设计有所帮助。
声明:本文内容来源于网络,版权归原作者所有,内容由互联网用户自发贡献自行上传,本网站不拥有所有权,未作人工编辑处理,也不承担相关法律责任。如果您发现有涉嫌版权的内容,欢迎发送邮件至:czq8825#qq.com(发邮件时,请将#更换为@)进行举报,并提供相关证据,一经查实,本站将立刻删除涉嫌侵权内容。