品牌建设网站有哪些,做棋牌网站团队,高佣金app软件推广平台,云服务器安装网站一、理论部分 一 什么是进程 进程#xff1a;正在进行的一个过程或者说一个任务。而负责执行任务则是cpu。 举例#xff08;单核多道#xff0c;实现多个进程的并发执行#xff09;#xff1a; egon在一个时间段内有很多任务要做#xff1a;python备课的任务#xff0c;…一、理论部分 一 什么是进程 进程正在进行的一个过程或者说一个任务。而负责执行任务则是cpu。 举例单核多道实现多个进程的并发执行 egon在一个时间段内有很多任务要做python备课的任务写书的任务交女朋友的任务王者荣耀上分的任务 但egon同一时刻只能做一个任务cpu同一时间只能干一个活如何才能玩出多个任务并发执行的效果 egon备一会课再去跟李杰的女朋友聊聊天再去打一会王者荣耀....这就保证了每个任务都在进行中. 二 进程与程序的区别 程序仅仅只是一堆代码而已而进程指的是程序的运行过程。 举例 想象一位有一手好厨艺的计算机科学家egon正在为他的女儿元昊烘制生日蛋糕。 他有做生日蛋糕的食谱 厨房里有所需的原料:面粉、鸡蛋、韭菜蒜泥等。 在这个比喻中 做蛋糕的食谱就是程序(即用适当形式描述的算法) 计算机科学家就是处理器(cpu) 而做蛋糕的各种原料就是输入数据。 进程就是厨师阅读食谱、取来各种原料以及烘制蛋糕等一系列动作的总和。 现在假设计算机科学家egon的儿子alex哭着跑了进来说XXXXXXXXXXXXXX。 科学家egon想了想处理儿子alex蛰伤的任务比给女儿元昊做蛋糕的任务更重要于是 计算机科学家就记录下他照着食谱做到哪儿了(保存进程的当前状态)然后拿出一本急救手册按照其中的指示处理蛰伤。这里我们看到处理机从一个进程(做蛋糕)切换到另一个高优先级的进程(实施医疗救治)每个进程拥有各自的程序(食谱和急救手册)。当蜜蜂蛰伤处理完之后这位计算机科学家又回来做蛋糕从他离开时的那一步继续做下去。 需要强调的是同一个程序执行两次那也是两个进程比如打开暴风影音虽然都是同一个软件但是一个可以播放苍井空一个可以播放饭岛爱。 三 并发与并行 无论是并行还是并发在用户看来都是同时运行的不管是进程还是线程都只是一个任务而已真是干活的是cpucpu来做这些任务而一个cpu同一时刻只能执行一个任务 一 并发是伪并行即看起来是同时运行。单个cpu多道技术就可以实现并发并行也属于并发 二 并行同时运行只有具备多个cpu才能实现并行 单核下可以利用多道技术多个核每个核也都可以利用多道技术多道技术是针对单核而言的 有四个核六个任务这样同一时间有四个任务被执行假设分别被分配给了cpu1cpu2cpu3cpu4 一旦任务1遇到I/O就被迫中断执行此时任务5就拿到cpu1的时间片去执行这就是单核下的多道技术 而一旦任务1的I/O结束了操作系统会重新调用它(需知进程的调度、分配给哪个cpu运行由操作系统说了算)可能被分配给四个cpu中的任意一个去执行 所有现代计算机经常会在同一时间做很多件事一个用户的PC无论是单cpu还是多cpu都可以同时运行多个任务一个任务可以理解为一个进程。 启动一个进程来杀毒360软件 启动一个进程来看电影暴风影音 启动一个进程来聊天腾讯QQ 所有的这些进程都需被管理于是一个支持多进程的多道程序系统是至关重要的 多道技术概念回顾内存中同时存入多道多个程序cpu从一个进程快速切换到另外一个使每个进程各自运行几十或几百毫秒这样虽然在某一个瞬间一个cpu只能执行一个任务但在1秒内cpu却可以运行多个进程这就给人产生了并行的错觉即伪并发以此来区分多处理器操作系统的真正硬件并行多个cpu共享同一个物理内存 四 同步与异步 同步执行一个进程在执行某个任务时另外一个进程必须等待其执行完毕才能继续执行异步执行一个进程在执行某个任务时另外一个进程无需等待其执行完毕就可以继续执行当有消息返回时系统会通知后者进行处理这样可以提高执行效率 举个例子打电话时就是同步通信发短息时就是异步通信。 同步\异步and阻塞\非阻塞重点 同步 #所谓同步就是在发出一个功能调用时在没有得到结果之前该调用就不会返回。按照这个定义其实绝大多数函数都是同步调用。但是一般而言我们在说同步、异步的时候特指那些需要其他部件协作或者需要一定时间完成的任务。
#举例
#1. multiprocessing.Pool下的apply #发起同步调用后就在原地等着任务结束根本不考虑任务是在计算还是在io阻塞总之就是一股脑地等任务结束
#2. concurrent.futures.ProcessPoolExecutor().submit(func,).result()
#3. concurrent.futures.ThreadPoolExecutor().submit(func,).result() 异步 #异步的概念和同步相对。当一个异步功能调用发出后调用者不能立刻得到结果。当该异步功能完成后通过状态、通知或回调来通知调用者。如果异步功能用状态来通知那么调用者就需要每隔一定时间检查一次效率就很低有些初学多线程编程的人总喜欢用一个循环去检查某个变量的值这其实是一 种很严重的错误。如果是使用通知的方式效率则很高因为异步功能几乎不需要做额外的操作。至于回调函数其实和通知没太多区别。
#举例
#1. multiprocessing.Pool().apply_async() #发起异步调用后并不会等待任务结束才返回相反会立即获取一个临时结果并不是最终的结果可能是封装好的一个对象。
#2. concurrent.futures.ProcessPoolExecutor(3).submit(func,)
#3. concurrent.futures.ThreadPoolExecutor(3).submit(func,) 阻塞 #阻塞调用是指调用结果返回之前当前线程会被挂起如遇到io操作。函数只有在得到结果之后才会将阻塞的线程激活。有人也许会把阻塞调用和同步调用等同起来实际上他是不同的。对于同步调用来说很多时候当前线程还是激活的只是从逻辑上当前函数没有返回而已。
#举例
#1. 同步调用apply一个累计1亿次的任务该调用会一直等待直到任务返回结果为止但并未阻塞住即便是被抢走cpu的执行权限那也是处于就绪态;
#2. 阻塞调用当socket工作在阻塞模式的时候如果没有数据的情况下调用recv函数则当前线程就会被挂起直到有数据为止。 非阻塞 #非阻塞和阻塞的概念相对应指在不能立刻得到结果之前也会立刻返回同时该函数不会阻塞当前线程。 小结 #1. 同步与异步针对的是函数/任务的调用方式同步就是当一个进程发起一个函数任务调用的时候一直等到函数任务完成而进程继续处于激活状态。而异步情况下是当一个进程发起一个函数任务调用的时候不会等函数返回而是继续往下执行当函数返回的时候通过状态、通知、事件等方式通知进程任务完成。#2. 阻塞与非阻塞针对的是进程或线程阻塞是当请求不能满足的时候就将进程挂起而非阻塞则不会阻塞当前进程 五 进程的创建了解 但凡是硬件都需要有操作系统去管理只要有操作系统就有进程的概念就需要有创建进程的方式一些操作系统只为一个应用程序设计比如微波炉中的控制器一旦启动微波炉所有的进程都已经存在。 而对于通用系统跑很多应用程序需要有系统运行过程中创建或撤销进程的能力主要分为4中形式创建新的进程 1. 系统初始化查看进程linux中用ps命令windows中用任务管理器前台进程负责与用户交互后台运行的进程与用户无关运行在后台并且只在需要时才唤醒的进程称为守护进程如电子邮件、web页面、新闻、打印 2. 一个进程在运行过程中开启了子进程如nginx开启多进程os.fork,subprocess.Popen等 3. 用户的交互式请求而创建一个新进程如用户双击暴风影音 4. 一个批处理作业的初始化只在大型机的批处理系统中应用 无论哪一种新进程的创建都是由一个已经存在的进程执行了一个用于创建进程的系统调用而创建的 1. 在UNIX中该系统调用是forkfork会创建一个与父进程一模一样的副本二者有相同的存储映像、同样的环境字符串和同样的打开文件在shell解释器进程中执行一个命令就会创建一个子进程 2. 在windows中该系统调用是CreateProcessCreateProcess既处理进程的创建也负责把正确的程序装入新进程。 关于创建的子进程UNIX和windows 1.相同的是进程创建后父进程和子进程有各自不同的地址空间多道技术要求物理层面实现进程之间内存的隔离任何一个进程的在其地址空间中的修改都不会影响到另外一个进程。 2.不同的是在UNIX中子进程的初始地址空间是父进程的一个副本提示子进程和父进程是可以有只读的共享内存区的。但是对于windows系统来说从一开始父进程与子进程的地址空间就是不同的。 六 进程的终止了解 1. 正常退出自愿如用户点击交互式页面的叉号或程序执行完毕调用发起系统调用正常退出在linux中用exit在windows中用ExitProcess 2. 出错退出自愿python a.py中a.py不存在 3. 严重错误非自愿执行非法指令如引用不存在的内存1/0等可以捕捉异常try...except... 4. 被其他进程杀死非自愿如kill -9 七 进程的层次结构 无论UNIX还是windows进程只有一个父进程不同的是 1. 在UNIX中所有的进程都是以init进程为根组成树形结构。父子进程共同组成一个进程组这样当从键盘发出一个信号时该信号被送给当前与键盘相关的进程组中的所有成员。 2. 在windows中没有进程层次的概念所有的进程都是地位相同的唯一类似于进程层次的暗示是在创建进程时父进程得到一个特别的令牌称为句柄,该句柄可以用来控制子进程但是父进程有权把该句柄传给其他子进程这样就没有层次了。 八 进程的状态 tail -f access.log |grep 404 执行程序tail开启一个子进程执行程序grep开启另外一个子进程两个进程之间基于管道|通讯将tail的结果作为grep的输入。 进程grep在等待输入即I/O时的状态称为阻塞此时grep命令都无法运行 其实在两种情况下会导致一个进程在逻辑上不能运行 1. 进程挂起是自身原因遇到I/O阻塞便要让出CPU让其他进程去执行这样保证CPU一直在工作 2. 与进程无关是操作系统层面可能会因为一个进程占用时间过多或者优先级等原因而调用其他的进程去使用CPU。 因而一个进程由三种状态 九 进程并发的实现了解 进程并发的实现在于硬件中断一个正在运行的进程把此时进程运行的所有状态保存下来为此操作系统维护一张表格即进程表process table每个进程占用一个进程表项这些表项也称为进程控制块 该表存放了进程状态的重要信息程序计数器、堆栈指针、内存分配状况、所有打开文件的状态、帐号和调度信息以及其他在进程由运行态转为就绪态或阻塞态时必须保存的信息从而保证该进程在再次启动时就像从未被中断过一样。 二、代码知识部分 一 multiprocessing模块介绍: python中的多线程无法利用多核优势如果想要充分地使用多核CPU的资源os.cpu_count()查看在python中大部分情况需要使用多进程。Python提供了multiprocessing。 multiprocessing模块用来开启子进程并在子进程中执行我们定制的任务比如函数该模块与多线程模块threading的编程接口类似。 multiprocessing模块的功能众多支持子进程、通信和共享数据、执行不同形式的同步提供了Process、Queue、Pipe、Lock等组件。 需要再次强调的一点是与线程不同进程没有任何共享状态进程修改的数据改动仅限于该进程内。 二 Process类的介绍 创建进程的类 Process([group [, target [, name [, args [, kwargs]]]]])由该类实例化得到的对象表示一个子进程中的任务尚未启动强调
1. 需要使用关键字的方式来指定参数
2. args指定的为传给target函数的位置参数是一个元组形式必须有逗号 参数介绍 group参数未使用值始终为None target表示调用对象即子进程要执行的任务 args表示调用对象的位置参数元组args(1,2,egon,) kwargs表示调用对象的字典,kwargs{name:egon,age:18} name为子进程的名称 方法介绍 p.start()启动进程并调用该子进程中的p.run() p.run():进程启动时运行的方法正是它去调用target指定的函数我们自定义类的类中一定要实现该方法 p.terminate():强制终止进程p不会进行任何清理操作如果p创建了子进程该子进程就成了僵尸进程使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放进而导致死锁p.is_alive():如果p仍然运行返回True p.join([timeout]):主线程等待p终止强调是主线程处于等的状态而p是处于运行的状态。timeout是可选的超时时间需要强调的是p.join只能join住start开启的进程而不能join住run开启的进程 属性介绍 1 p.daemon默认值为False如果设为True代表p为后台运行的守护进程当p的父进程终止时p也随之终止并且设定为True后p不能创建自己的新进程必须在p.start()之前设置
2
3 p.name:进程的名称
4
5 p.pid进程的pid
6
7 p.exitcode:进程在运行时为None、如果为–N表示被信号N结束(了解即可)
8
9 p.authkey:进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性这类连接只有在具有相同的身份验证键时才能成功了解即可 三 Process类的使用 注意在windows中Process()必须放到# if __name__ __main__:下 Since Windows has no fork, the multiprocessing module starts a new Python process and imports the calling module.
If Process() gets called upon import, then this sets off an infinite succession of new processes (or until your machine runs out of resources).
This is the reason for hiding calls to Process() insideif __name__ __main__
since statements inside this if-statement will not get called upon import.
由于Windows没有fork多处理模块启动一个新的Python进程并导入调用模块。
如果在导入时调用Process那么这将启动无限继承的新进程或直到机器耗尽资源。
这是隐藏对Process内部调用的原使用if __name__ “__main __”这个if语句中的语句将不会在导入时被调用。 创建并开启子进程的两种方式 #开进程的方法一:
import time
import random
from multiprocessing import Process
def piao(name):print(%s piaoing %name)time.sleep(random.randrange(1,5))print(%s piao end %name)p1Process(targetpiao,args(egon,)) #必须加,号
p2Process(targetpiao,args(alex,))
p3Process(targetpiao,args(wupeqi,))
p4Process(targetpiao,args(yuanhao,))p1.start()
p2.start()
p3.start()
p4.start()
print(主线程)方法一 #开进程的方法二:
import time
import random
from multiprocessing import Processclass Piao(Process):def __init__(self,name):super().__init__()self.namenamedef run(self):print(%s piaoing %self.name)time.sleep(random.randrange(1,5))print(%s piao end %self.name)p1Piao(egon)
p2Piao(alex)
p3Piao(wupeiqi)
p4Piao(yuanhao)p1.start() #start会自动调用run
p2.start()
p3.start()
p4.start()
print(主线程) 练习1把所学的socket通信变成并发的形式 from socket import *
from multiprocessing import Processserversocket(AF_INET,SOCK_STREAM)
server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server.bind((127.0.0.1,8080))
server.listen(5)def talk(conn,client_addr):while True:try:msgconn.recv(1024)if not msg:breakconn.send(msg.upper())except Exception:breakif __name__ __main__: #windows下start进程一定要写到这下面while True:conn,client_addrserver.accept()pProcess(targettalk,args(conn,client_addr))p.start()server端 from socket import *clientsocket(AF_INET,SOCK_STREAM)
client.connect((127.0.0.1,8080))while True:msginput(: ).strip()if not msg:continueclient.send(msg.encode(utf-8))msgclient.recv(1024)print(msg.decode(utf-8))多个client端 每来一个客户端都在服务端开启一个进程如果并发来一个万个客户端要开启一万个进程吗你自己尝试着在你自己的机器上开启一万个10万个进程试一试。
解决方法进程池 Process对象的join方法 join主进程等等待子进程结束 from multiprocessing import Process
import time
import randomclass Piao(Process):def __init__(self,name):self.namenamesuper().__init__()def run(self):print(%s is piaoing %self.name)time.sleep(random.randrange(1,3))print(%s is piao end %self.name)pPiao(egon)
p.start()
p.join(0.0001) #等待p停止,等0.0001秒就不再等了
print(开始) 有了join程序不就是串行了吗 from multiprocessing import Process
import time
import random
def piao(name):print(%s is piaoing %name)time.sleep(random.randint(1,3))print(%s is piao end %name)p1Process(targetpiao,args(egon,))
p2Process(targetpiao,args(alex,))
p3Process(targetpiao,args(yuanhao,))
p4Process(targetpiao,args(wupeiqi,))p1.start()
p2.start()
p3.start()
p4.start()#有的同学会有疑问:既然join是等待进程结束,那么我像下面这样写,进程不就又变成串行的了吗?
#当然不是了,必须明确p.join()是让谁等
#很明显p.join()是让主线程等待p的结束卡住的是主线程而绝非进程p#详细解析如下
#进程只要start就会在开始运行了,所以p1-p4.start()时,系统中已经有四个并发的进程了
#而我们p1.join()是在等p1结束,没错p1只要不结束主线程就会一直卡在原地,这也是问题的关键
#join是让主线程等,而p1-p4仍然是并发执行的,p1.join的时候,其余p2,p3,p4仍然在运行,等#p1.join结束,可能p2,p3,p4早已经结束了,这样p2.join,p3.join.p4.join直接通过检测无需等待
# 所以4个join花费的总时间仍然是耗费时间最长的那个进程运行的时间
p1.join()
p2.join()
p3.join()
p4.join()print(主线程)#上述启动进程与join进程可以简写为
# p_l[p1,p2,p3,p4]
#
# for p in p_l:
# p.start()
#
# for p in p_l:
# p.join() Process对象的其他方法或属性了解 terminate与is_alive #进程对象的其他方法一:terminate,is_alive
from multiprocessing import Process
import time
import randomclass Piao(Process):def __init__(self,name):self.namenamesuper().__init__()def run(self):print(%s is piaoing %self.name)time.sleep(random.randrange(1,5))print(%s is piao end %self.name)p1Piao(egon1)
p1.start()p1.terminate()#关闭进程,不会立即关闭,所以is_alive立刻查看的结果可能还是存活
print(p1.is_alive()) #结果为Trueprint(开始)
print(p1.is_alive()) #结果为Falsename与pid from multiprocessing import Process
import time
import random
class Piao(Process):def __init__(self,name):# self.namename# super().__init__() #Process的__init__方法会执行self.namePiao-1,# #所以加到这里,会覆盖我们的self.namename#为我们开启的进程设置名字的做法super().__init__()self.namenamedef run(self):print(%s is piaoing %self.name)time.sleep(random.randrange(1,3))print(%s is piao end %self.name)pPiao(egon)
p.start()
print(开始)
print(p.pid) #查看pid四 守护进程 主进程创建守护进程 其一守护进程会在主进程代码执行结束后就终止 其二守护进程内无法再开启子进程,否则抛出异常AssertionError: daemonic processes are not allowed to have children 注意进程之间是互相独立的主进程代码运行结束守护进程随即终止 from multiprocessing import Process
import time
import randomclass Piao(Process):def __init__(self,name):self.namenamesuper().__init__()def run(self):print(%s is piaoing %self.name)time.sleep(random.randrange(1,3))print(%s is piao end %self.name)pPiao(egon)
p.daemonTrue #一定要在p.start()前设置,设置p为守护进程,禁止p创建子进程,并且父进程代码执行结束,p即终止运行
p.start()
print(主) #主进程代码运行完毕,守护进程就会结束
from multiprocessing import Process
from threading import Thread
import time
def foo():print(123)time.sleep(1)print(end123)def bar():print(456)time.sleep(3)print(end456)p1Process(targetfoo)
p2Process(targetbar)p1.daemonTrue
p1.start()
p2.start()
print(main-------) #打印该行则主进程代码结束,则守护进程p1应该被终止,可能会有p1任务执行的打印信息123,因为主进程打印main----时,p1也执行了,但是随即被终止迷惑人的例子 五 进程同步(锁) 进程之间数据不共享,但是共享同一套文件系统,所以访问同一个文件,或同一个打印终端,是没有问题的, 竞争带来的结果就是错乱如何控制就是加锁处理 part1多个进程共享同一打印终端 并发运行,效率高,但竞争同一打印终端,带来了打印错乱 #并发运行,效率高,但竞争同一打印终端,带来了打印错乱
from multiprocessing import Process
import os,time
def work():print(%s is running %os.getpid())time.sleep(2)print(%s is done %os.getpid())if __name__ __main__:for i in range(3):pProcess(targetwork)p.start() 加锁由并发变成了串行,牺牲了运行效率,但避免了竞争 #由并发变成了串行,牺牲了运行效率,但避免了竞争
from multiprocessing import Process,Lock
import os,time
def work(lock):lock.acquire()print(%s is running %os.getpid())time.sleep(2)print(%s is done %os.getpid())lock.release()
if __name__ __main__:lockLock()for i in range(3):pProcess(targetwork,args(lock,))p.start() part2多个进程共享同一文件 文件当数据库模拟抢票 并发运行效率高但竞争写同一文件数据写入错乱 #文件db的内容为{count:1}
#注意一定要用双引号不然json无法识别
from multiprocessing import Process,Lock
import time,json,random
def search():dicjson.load(open(db.txt))print(\033[43m剩余票数%s\033[0m %dic[count])def get():dicjson.load(open(db.txt))time.sleep(0.1) #模拟读数据的网络延迟if dic[count] 0:dic[count]-1time.sleep(0.2) #模拟写数据的网络延迟json.dump(dic,open(db.txt,w))print(\033[43m购票成功\033[0m)def task(lock):search()get()
if __name__ __main__:lockLock()for i in range(100): #模拟并发100个客户端抢票pProcess(targettask,args(lock,))p.start() 加锁购票行为由并发变成了串行牺牲了运行效率但保证了数据安全 #文件db的内容为{count:1}
#注意一定要用双引号不然json无法识别
from multiprocessing import Process,Lock
import time,json,random
def search():dicjson.load(open(db.txt))print(\033[43m剩余票数%s\033[0m %dic[count])def get():dicjson.load(open(db.txt))time.sleep(0.1) #模拟读数据的网络延迟if dic[count] 0:dic[count]-1time.sleep(0.2) #模拟写数据的网络延迟json.dump(dic,open(db.txt,w))print(\033[43m购票成功\033[0m)def task(lock):search()lock.acquire()get()lock.release()
if __name__ __main__:lockLock()for i in range(100): #模拟并发100个客户端抢票pProcess(targettask,args(lock,))p.start() 总结 加锁可以保证多个进程修改同一块数据时同一时间只能有一个任务可以进行修改即串行的修改没错速度是慢了但牺牲了速度却保证了数据安全。虽然可以用文件共享数据实现进程间通信但问题是1.效率低2.需要自己加锁处理 为此mutiprocessing模块为我们提供了基于消息的IPC通信机制队列和管道。1 队列和管道都是将数据存放于内存中2 队列又是基于管道锁实现的可以让我们从复杂的锁问题中解脱出来我们应该尽量避免使用共享数据尽可能使用消息传递和队列避免处理复杂的同步和锁问题而且在进程数目增多时往往可以获得更好的可获展性。 六 队列推荐使用 进程彼此之间互相隔离要实现进程间通信IPCmultiprocessing模块支持两种形式队列和管道这两种方式都是使用消息传递的 创建队列的类底层就是以管道和锁定的方式实现 1 Queue([maxsize]):创建共享的进程队列Queue是多进程安全的队列可以使用Queue实现多进程之间的数据传递。 参数介绍 1 maxsize是队列中允许最大项数省略则无大小限制。 方法介绍 主要方法 q.put方法用以插入数据到队列中put方法还有两个可选参数blocked和timeout。如果blocked为True默认值并且timeout为正值该方法会阻塞timeout指定的时间直到该队列有剩余的空间。如果超时会抛出Queue.Full异常。如果blocked为False但该Queue已满会立即抛出Queue.Full异常。
q.get方法可以从队列读取并且删除一个元素。同样get方法有两个可选参数blocked和timeout。如果blocked为True默认值并且timeout为正值那么在等待时间内没有取到任何元素会抛出Queue.Empty异常。如果blocked为False有两种情况存在如果Queue有一个值可用则立即返回该值否则如果队列为空则立即抛出Queue.Empty异常.q.get_nowait():同q.get(False)
q.put_nowait():同q.put(False)q.empty():调用此方法时q为空则返回True该结果不可靠比如在返回True的过程中如果队列中又加入了项目。
q.full()调用此方法时q已满则返回True该结果不可靠比如在返回True的过程中如果队列中的项目被取走。
q.qsize():返回队列中目前项目的正确数量结果也不可靠理由同q.empty()和q.full()一样 其他方法(了解) 1 q.cancel_join_thread():不会在进程退出时自动连接后台线程。可以防止join_thread()方法阻塞
2 q.close():关闭队列防止队列中加入更多数据。调用此方法后台线程将继续写入那些已经入队列但尚未写入的数据但将在此方法完成时马上关闭。如果q被垃圾收集将调用此方法。关闭队列不会在队列使用者中产生任何类型的数据结束信号或异常。例如如果某个使用者正在被阻塞在get()操作上关闭生产者中的队列不会导致get()方法返回错误。
3 q.join_thread()连接队列的后台线程。此方法用于在调用q.close()方法之后等待所有队列项被消耗。默认情况下此方法由不是q的原始创建者的所有进程调用。调用q.cancel_join_thread方法可以禁止这种行为 应用
multiprocessing模块支持进程间通信的两种主要形式:管道和队列
都是基于消息传递实现的,但是队列接口
from multiprocessing import Process,Queue
import time
qQueue(3)#put ,get ,put_nowait,get_nowait,full,empty
q.put(3)
q.put(3)
q.put(3)
print(q.full()) #满了print(q.get())
print(q.get())
print(q.get())
print(q.empty()) #空了 生产者消费者模型 在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。 为什么要使用生产者和消费者模式 在线程世界里生产者就是生产数据的线程消费者就是消费数据的线程。在多线程开发当中如果生产者处理速度很快而消费者处理速度很慢那么生产者就必须等待消费者处理完才能继续生产数据。同样的道理如果消费者的处理能力大于生产者那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。 什么是生产者消费者模式 生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯而通过阻塞队列来进行通讯所以生产者生产完数据之后不用等待消费者处理直接扔给阻塞队列消费者不找生产者要数据而是直接从阻塞队列里取阻塞队列就相当于一个缓冲区平衡了生产者和消费者的处理能力。 基于队列实现生产者消费者模型 from multiprocessing import Process,Queue
import time,random,os
def consumer(q):while True:resq.get()time.sleep(random.randint(1,3))print(\033[45m%s 吃 %s\033[0m %(os.getpid(),res))def producer(q):for i in range(10):time.sleep(random.randint(1,3))res包子%s %iq.put(res)print(\033[44m%s 生产了 %s\033[0m %(os.getpid(),res))if __name__ __main__:qQueue()#生产者们:即厨师们p1Process(targetproducer,args(q,))#消费者们:即吃货们c1Process(targetconsumer,args(q,))#开始p1.start()c1.start()print(主) 此时的问题是主进程永远不会结束原因是生产者p在生产完后就结束了但是消费者c在取空了q之后则一直处于死循环中且卡在q.get()这一步。 解决方式无非是让生产者在生产完毕后往队列中再发一个结束信号这样消费者在接收到结束信号后就可以break出死循环
生产者在生产完毕后发送结束信号None from multiprocessing import Process,Queue
import time,random,os
def consumer(q):while True:resq.get()if res is None:break #收到结束信号则结束time.sleep(random.randint(1,3))print(\033[45m%s 吃 %s\033[0m %(os.getpid(),res))def producer(q):for i in range(10):time.sleep(random.randint(1,3))res包子%s %iq.put(res)print(\033[44m%s 生产了 %s\033[0m %(os.getpid(),res))q.put(None) #发送结束信号
if __name__ __main__:qQueue()#生产者们:即厨师们p1Process(targetproducer,args(q,))#消费者们:即吃货们c1Process(targetconsumer,args(q,))#开始p1.start()c1.start()print(主)注意结束信号None不一定要由生产者发主进程里同样可以发但主进程需要等生产者结束后才应该发送该信号 主进程在生产者生产完毕后发送结束信号None from multiprocessing import Process,Queue
import time,random,os
def consumer(q):while True:resq.get()if res is None:break #收到结束信号则结束time.sleep(random.randint(1,3))print(\033[45m%s 吃 %s\033[0m %(os.getpid(),res))def producer(q):for i in range(2):time.sleep(random.randint(1,3))res包子%s %iq.put(res)print(\033[44m%s 生产了 %s\033[0m %(os.getpid(),res))if __name__ __main__:qQueue()#生产者们:即厨师们p1Process(targetproducer,args(q,))#消费者们:即吃货们c1Process(targetconsumer,args(q,))#开始p1.start()c1.start()p1.join()q.put(None) #发送结束信号print(主) 但上述解决方式在有多个生产者和多个消费者时我们则需要用一个很low的方式去解决 有几个生产者就需要发送几次结束信号相当low from multiprocessing import Process,Queue
import time,random,os
def consumer(q):while True:resq.get()if res is None:break #收到结束信号则结束time.sleep(random.randint(1,3))print(\033[45m%s 吃 %s\033[0m %(os.getpid(),res))def producer(name,q):for i in range(2):time.sleep(random.randint(1,3))res%s%s %(name,i)q.put(res)print(\033[44m%s 生产了 %s\033[0m %(os.getpid(),res))if __name__ __main__:qQueue()#生产者们:即厨师们p1Process(targetproducer,args(包子,q))p2Process(targetproducer,args(骨头,q))p3Process(targetproducer,args(泔水,q))#消费者们:即吃货们c1Process(targetconsumer,args(q,))c2Process(targetconsumer,args(q,))#开始p1.start()p2.start()p3.start()c1.start()p1.join() #必须保证生产者全部生产完毕,才应该发送结束信号p2.join()p3.join()q.put(None) #有几个生产者就应该发送几次结束信号Noneq.put(None) #发送结束信号q.put(None) #发送结束信号print(主) 其实我们的思路无非是发送结束信号而已有另外一种队列提供了这种机制 #JoinableQueue([maxsize])这就像是一个Queue对象但队列允许项目的使用者通知生成者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。#参数介绍maxsize是队列中允许最大项数省略则无大小限制。 #方法介绍JoinableQueue的实例p除了与Queue对象相同的方法之外还具有q.task_done()使用者使用此方法发出信号表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量将引发ValueError异常q.join():生产者调用此方法进行阻塞直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done方法为止 from multiprocessing import Process,JoinableQueue
import time,random,os
def consumer(q):while True:resq.get()time.sleep(random.randint(1,3))print(\033[45m%s 吃 %s\033[0m %(os.getpid(),res))q.task_done() #向q.join()发送一次信号,证明一个数据已经被取走了def producer(name,q):for i in range(10):time.sleep(random.randint(1,3))res%s%s %(name,i)q.put(res)print(\033[44m%s 生产了 %s\033[0m %(os.getpid(),res))q.join()if __name__ __main__:qJoinableQueue()#生产者们:即厨师们p1Process(targetproducer,args(包子,q))p2Process(targetproducer,args(骨头,q))p3Process(targetproducer,args(泔水,q))#消费者们:即吃货们c1Process(targetconsumer,args(q,))c2Process(targetconsumer,args(q,))c1.daemonTruec2.daemonTrue#开始p_l[p1,p2,p3,c1,c2]for p in p_l:p.start()p1.join()p2.join()p3.join()print(主) #主进程等---p1,p2,p3等----c1,c2#p1,p2,p3结束了,证明c1,c2肯定全都收完了p1,p2,p3发到队列的数据#因而c1,c2也没有存在的价值了,应该随着主进程的结束而结束,所以设置成守护进程 七 管道 进程间通信IPC方式二管道不推荐使用了解即可 #创建管道的类
Pipe([duplex]):在进程之间创建一条管道并返回元组conn1,conn2,其中conn1conn2表示管道两端的连接对象强调一点必须在产生Process对象之前产生管道
#参数介绍
dumplex:默认管道是全双工的如果将duplex射成Falseconn1只能用于接收conn2只能用于发送。
#主要方法conn1.recv():接收conn2.send(obj)发送的对象。如果没有消息可接收recv方法会一直阻塞。如果连接的另外一端已经关闭那么recv方法会抛出EOFError。conn1.send(obj):通过连接发送对象。obj是与序列化兼容的任意对象#其他方法
conn1.close():关闭连接。如果conn1被垃圾回收将自动调用此方法
conn1.fileno():返回连接使用的整数文件描述符
conn1.poll([timeout]):如果连接上的数据可用返回True。timeout指定等待的最长时限。如果省略此参数方法将立即返回结果。如果将timeout射成None操作将无限期地等待数据到达。conn1.recv_bytes([maxlength]):接收c.send_bytes()方法发送的一条完整的字节消息。maxlength指定要接收的最大字节数。如果进入的消息超过了这个最大值将引发IOError异常并且在连接上无法进行进一步读取。如果连接的另外一端已经关闭再也不存在任何数据将引发EOFError异常。
conn.send_bytes(buffer [, offset [, size]])通过连接发送字节数据缓冲区buffer是支持缓冲区接口的任意对象offset是缓冲区中的字节偏移量而size是要发送字节数。结果数据以单条消息的形式发出然后调用c.recv_bytes()函数进行接收 conn1.recv_bytes_into(buffer [, offset]):接收一条完整的字节消息并把它保存在buffer对象中该对象支持可写入的缓冲区接口即bytearray对象或类似的对象。offset指定缓冲区中放置消息处的字节位移。返回值是收到的字节数。如果消息长度大于可用的缓冲区空间将引发BufferTooShort异常。介绍
基于管道实现进程间通信与队列的方式是类似的队列就是管道加锁实现的 from multiprocessing import Process,Pipeimport time,os
def consumer(p,name):left,rightpleft.close()while True:try:baoziright.recv()print(%s 收到包子:%s %(name,baozi))except EOFError:right.close()break
def producer(seq,p):left,rightpright.close()for i in seq:left.send(i)# time.sleep(1)else:left.close()
if __name__ __main__:left,rightPipe()c1Process(targetconsumer,args((left,right),c1))c1.start()seq(i for i in range(10))producer(seq,(left,right))right.close()left.close()c1.join()print(主进程)注意生产者和消费者都没有使用管道的某个端点就应该将其关闭如在生产者中关闭管道的右端在消费者中关闭管道的左端。如果忘记执行这些步骤程序可能再消费者中的recv()操作上挂起。管道是由操作系统进行引用计数的,必须在所有进程中关闭管道后才能生产EOFError异常。因此在生产者中关闭管道不会有任何效果付费消费者中也关闭了相同的管道端点。 管道可以用于双向通信利用通常在客户端/服务器中使用的请求响应模型或远程过程调用就可以使用管道编写与进程交互的程序 from multiprocessing import Process,Pipeimport time,os
def adder(p,name):server,clientpclient.close()while True:try:x,yserver.recv()except EOFError:server.close()breakresxyserver.send(res)print(server done)
if __name__ __main__:server,clientPipe()c1Process(targetadder,args((server,client),c1))c1.start()server.close()client.send((10,20))print(client.recv())client.close()c1.join()print(主进程)
#注意send()和recv()方法使用pickle模块对对象进行序列化。八 共享数据 展望未来基于消息传递的并发编程是大势所趋 即便是使用线程推荐做法也是将程序设计为大量独立的线程集合 通过消息队列交换数据。这样极大地减少了对使用锁定和其他同步手段的需求 还可以扩展到分布式系统中 进程间通信应该尽量避免使用本节所讲的共享数据的方式 进程间数据是独立的可以借助于队列或管道实现通信二者都是基于消息传递的虽然进程间数据独立但可以通过Manager实现数据共享事实上Manager的功能远不止于此A manager object returned by Manager() controls a server process which holds Python objects and allows other processes to manipulate them using proxies.A manager returned by Manager() will support types list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value and Array. For example, #进程之间操作共享的数据 from multiprocessing import Manager,Process,Lock
import os
def work(d,lock):# with lock: #不加锁而操作共享的数据,肯定会出现数据错乱d[count]-1if __name__ __main__:lockLock()with Manager() as m:dicm.dict({count:100})p_l[]for i in range(100):pProcess(targetwork,args(dic,lock))p_l.append(p)p.start()for p in p_l:p.join()print(dic)#{count: 94}九 信号量(了解 互斥锁 同时只允许一个线程更改数据而Semaphore是同时允许一定数量的线程更改数据 比如厕所有3个坑那最多只允许3个人上厕所后面的人只能等里面有人出来了才能再进去如果指定信号量为3那么来一个人获得一把锁计数加1当计数等于3时后面的人均需要等待。一旦释放就有人可以获得一把锁信号量与进程池的概念很像但是要区分开信号量涉及到加锁的概念from multiprocessing import Process,Semaphore
import time,randomdef go_wc(sem,user):sem.acquire()print(%s 占到一个茅坑 %user)time.sleep(random.randint(0,3)) #模拟每个人拉屎速度不一样0代表有的人蹲下就起来了sem.release()if __name__ __main__:semSemaphore(5)p_l[]for i in range(13):pProcess(targetgo_wc,args(sem,user%s %i,))p.start()p_l.append(p)for i in p_l:i.join()print(》)信号量Semahpore同线程一样 十 事件(了解) python线程的事件用于主线程控制其他线程的执行事件主要提供了三个方法 set、wait、clear。事件处理的机制全局定义了一个“Flag”如果“Flag”值为 False那么当程序执行 event.wait 方法时就会阻塞如果“Flag”值为True那么event.wait 方法时便不再阻塞。clear将“Flag”设置为False
set将“Flag”设置为True#_*_coding:utf-8_*_
#!/usr/bin/env pythonfrom multiprocessing import Process,Event
import time,randomdef car(e,n):while True:if not e.is_set(): #Flaseprint(\033[31m红灯亮\033[0mcar%s等着 %n)e.wait()print(\033[32m车%s 看见绿灯亮了\033[0m %n)time.sleep(random.randint(3,6))if not e.is_set():continueprint(走你,car, n)breakdef police_car(e,n):while True:if not e.is_set():print(\033[31m红灯亮\033[0mcar%s等着 % n)e.wait(1)print(灯的是%s警车走了,car %s %(e.is_set(),n))breakdef traffic_lights(e,inverval):while True:time.sleep(inverval)if e.is_set():e.clear() #e.is_set() ----Falseelse:e.set()if __name__ __main__:eEvent()# for i in range(10):# pProcess(targetcar,args(e,i,))# p.start()for i in range(5):p Process(targetpolice_car, args(e, i,))p.start()tProcess(targettraffic_lights,args(e,10))t.start()print(》)Event同线程一样 十一 进程池 在利用Python进行系统管理的时候特别是同时操作多个文件目录或者远程控制多台主机并行操作可以节约大量的时间。多进程是实现并发的手段之一需要注意的问题是 很明显需要并发执行的任务通常要远大于核数一个操作系统不可能无限开启进程通常有几个核就开几个进程进程开启过多效率反而会下降开启进程是需要占用系统资源的而且开启多余核数目的进程也无法做到并行例如当被操作对象数目不大时可以直接利用multiprocessing中的Process动态成生多个进程十几个还好但如果是上百个上千个。。。手动的去限制进程数量却又太过繁琐此时可以发挥进程池的功效。 我们就可以通过维护一个进程池来控制进程数目比如httpd的进程模式规定最小进程数和最大进程数... ps对于远程过程调用的高级应用程序而言应该使用进程池Pool可以提供指定数量的进程供用户调用当有新的请求提交到pool中时如果池还没有满那么就会创建一个新的进程用来执行该请求但如果池中的进程数已经达到规定最大值那么该请求就会等待直到池中有进程结束就重用进程池中的进程。 创建进程池的类如果指定numprocess为3则进程池会从无到有创建三个进程然后自始至终使用这三个进程去执行所有任务不会开启其他进程 1 Pool([numprocess [,initializer [, initargs]]]):创建进程池 参数介绍 1 numprocess:要创建的进程数如果省略将默认使用cpu_count()的值
2 initializer是每个工作进程启动时要执行的可调用对象默认为None
3 initargs是要传给initializer的参数组 方法介绍 主要方法 p.apply(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。需要强调的是此操作并不会在所有池工作进程中并执行func函数。如果要通过不同参数并发地执行func函数必须从不同线程调用p.apply()函数或者使用p.apply_async()
p.apply_async(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。此方法的结果是AsyncResult类的实例callback是可调用对象接收输入参数。当func的结果变为可用时将理解传递给callback。callback禁止执行任何阻塞操作否则将接收其他异步操作中的结果。p.close():关闭进程池防止进一步操作。如果所有操作持续挂起它们将在工作进程终止前完成
P.jion():等待所有工作进程退出。此方法只能在close或teminate()之后调用 其他方法了解部分 方法apply_async()和map_async的返回值是AsyncResul的实例obj。实例具有以下方法
obj.get():返回结果如果有必要则等待结果到达。timeout是可选的。如果在指定时间内还没有到达将引发一场。如果远程操作中引发了异常它将在调用此方法时再次被引发。
obj.ready():如果调用完成返回True
obj.successful():如果调用完成且没有引发异常返回True如果在结果就绪之前调用此方法引发异常
obj.wait([timeout]):等待结果变为可用。
obj.terminate()立即终止所有工作进程同时不执行任何清理或结束任何挂起工作。如果p被垃圾回收将自动调用此函数 应用 apply同步执行阻塞式 from multiprocessing import Pool
import os,time
def work(n):print(%s run %os.getpid())time.sleep(3)return n**2if __name__ __main__:pPool(3) #进程池中从无到有创建三个进程,以后一直是这三个进程在执行任务res_l[]for i in range(10):resp.apply(work,args(i,)) #同步运行,阻塞、直到本次任务执行完毕拿到resres_l.append(res)print(res_l)apply_async异步执行非阻塞 from multiprocessing import Pool
import os,time
def work(n):print(%s run %os.getpid())time.sleep(3)return n**2if __name__ __main__:pPool(3) #进程池中从无到有创建三个进程,以后一直是这三个进程在执行任务res_l[]for i in range(10):resp.apply_async(work,args(i,)) #同步运行,阻塞、直到本次任务执行完毕拿到resres_l.append(res)#异步apply_async用法如果使用异步提交的任务主进程需要使用jion等待进程池内任务都处理完然后可以用get收集结果否则主进程结束进程池可能还没来得及执行也就跟着一起结束了p.close()p.join()for res in res_l:print(res.get()) #使用get来获取apply_aync的结果,如果是apply,则没有get方法,因为apply是同步执行,立刻获取结果,也根本无需get详解apply_async与apply #一使用进程池非阻塞,apply_async
#coding: utf-8
from multiprocessing import Process,Pool
import timedef func(msg):print( msg:, msg)time.sleep(1)return msgif __name__ __main__:pool Pool(processes 3)res_l[]for i in range(10):msg hello %d %(i)respool.apply_async(func, (msg, )) #维持执行的进程总数为processes当一个进程执行完毕后会添加新的进程进去res_l.append(res)print() #没有后面的join或get则程序整体结束进程池中的任务还没来得及全部执行完也都跟着主进程一起结束了
pool.close() #关闭进程池防止进一步操作。如果所有操作持续挂起它们将在工作进程终止前完成pool.join() #调用join之前先调用close函数否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束print(res_l) #看到的是multiprocessing.pool.ApplyResult object at 0x10357c4e0对象组成的列表,而非最终的结果,但这一步是在join后执行的,证明结果已经计算完毕,剩下的事情就是调用每个对象下的get方法去获取结果for i in res_l:print(i.get()) #使用get来获取apply_aync的结果,如果是apply,则没有get方法,因为apply是同步执行,立刻获取结果,也根本无需get#二使用进程池阻塞,apply
#coding: utf-8
from multiprocessing import Process,Pool
import timedef func(msg):print( msg:, msg)time.sleep(0.1)return msgif __name__ __main__:pool Pool(processes 3)res_l[]for i in range(10):msg hello %d %(i)respool.apply(func, (msg, )) #维持执行的进程总数为processes当一个进程执行完毕后会添加新的进程进去res_l.append(res) #同步执行即执行完一个拿到结果再去执行另外一个print()pool.close()pool.join() #调用join之前先调用close函数否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束print(res_l) #看到的就是最终的结果组成的列表for i in res_l: #apply是同步的所以直接得到结果没有get()方法print(i) 练习2使用进程池维护固定数目的进程重写练习1 server端 #Pool内的进程数默认是cpu核数假设为4查看方法os.cpu_count()
#开启6个客户端会发现2个客户端处于等待状态
#在每个进程内查看pid会发现pid使用为4个即多个客户端公用4个进程
from socket import *
from multiprocessing import Pool
import osserversocket(AF_INET,SOCK_STREAM)
server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server.bind((127.0.0.1,8080))
server.listen(5)def talk(conn,client_addr):print(进程pid: %s %os.getpid())while True:try:msgconn.recv(1024)if not msg:breakconn.send(msg.upper())except Exception:breakif __name__ __main__:pPool()while True:conn,client_addrserver.accept()p.apply_async(talk,args(conn,client_addr))# p.apply(talk,args(conn,client_addr)) #同步的话则同一时间只有一个客户端能访问客户端 from socket import *clientsocket(AF_INET,SOCK_STREAM)
client.connect((127.0.0.1,8080))while True:msginput(: ).strip()if not msg:continueclient.send(msg.encode(utf-8))msgclient.recv(1024)print(msg.decode(utf-8)) 发现并发开启多个客户端服务端同一时间只有3个不同的pid干掉一个客户端另外一个客户端才会进来被3个进程之一处理 回掉函数 需要回调函数的场景进程池中任何一个任务一旦处理完了就立即告知主进程我好了额你可以处理我的结果了。主进程则调用一个函数去处理该结果该函数即回调函数 我们可以把耗时间阻塞的任务放到进程池中然后指定回调函数主进程负责执行这样主进程在执行回调函数时就省去了I/O的过程直接拿到的是任务的结果。 from multiprocessing import Pool
import requests
import json
import osdef get_page(url):print(进程%s get %s %(os.getpid(),url))responerequests.get(url)if respone.status_code 200:return {url:url,text:respone.text}def pasrse_page(res):print(进程%s parse %s %(os.getpid(),res[url]))parse_resurl:%s size:[%s]\n %(res[url],len(res[text]))with open(db.txt,a) as f:f.write(parse_res)if __name__ __main__:urls[https://www.baidu.com,https://www.python.org,https://www.openstack.org,https://help.github.com/,http://www.sina.com.cn/]pPool(3)res_l[]for url in urls:resp.apply_async(get_page,args(url,),callbackpasrse_page)res_l.append(res)p.close()p.join()print([res.get() for res in res_l]) #拿到的是get_page的结果,其实完全没必要拿该结果,该结果已经传给回调函数处理了
打印结果:
进程3388 get https://www.baidu.com
进程3389 get https://www.python.org
进程3390 get https://www.openstack.org
进程3388 get https://help.github.com/
进程3387 parse https://www.baidu.com
进程3389 get http://www.sina.com.cn/
进程3387 parse https://www.python.org
进程3387 parse https://help.github.com/
进程3387 parse http://www.sina.com.cn/
进程3387 parse https://www.openstack.org
[{url: https://www.baidu.com, text: !DOCTYPE html\r\n...,...}]爬虫案例 from multiprocessing import Pool
import time,random
import requests
import redef get_page(url,pattern):responserequests.get(url)if response.status_code 200:return (response.text,pattern)def parse_page(info):page_content,patterninforesre.findall(pattern,page_content)for item in res:dic{index:item[0],title:item[1],actor:item[2].strip()[3:],time:item[3][5:],score:item[4]item[5]}print(dic)
if __name__ __main__:pattern1re.compile(rdd.*?board-index.*?(\d).*?title(.*?).*?star.*?(.*?).*?releasetime.*?(.*?).*?integer.*?(.*?).*?fraction.*?(.*?),re.S)url_dic{http://maoyan.com/board/7:pattern1,}pPool()res_l[]for url,pattern in url_dic.items():resp.apply_async(get_page,args(url,pattern),callbackparse_page)res_l.append(res)for i in res_l:i.get()# resrequests.get(http://maoyan.com/board/7)# print(re.findall(pattern,res.text)) 如果在主进程中等待进程池中所有任务都执行完毕后再统一处理结果则无需回调函数 from multiprocessing import Pool
import time,random,osdef work(n):time.sleep(1)return n**2
if __name__ __main__:pPool()res_l[]for i in range(10):resp.apply_async(work,args(i,))res_l.append(res)p.close()p.join() #等待进程池中所有进程执行完毕
nums[]for res in res_l:nums.append(res.get()) #拿到所有结果print(nums) #主进程拿到所有的处理结果,可以在主进程中进行统一进行处理 进程池的其他实现方式https://docs.python.org/dev/library/concurrent.futures.html 转载于:https://www.cnblogs.com/ctztake/p/7445428.html
