multiprocessing模块

创建进程的类

Process([group [, target [, name [, args [, kwargs]]]]])，由该类实例化得到的对象，表示一个子进程中的任务（尚未启动）强调：1. 需要使用关键字的方式来指定参数2. args指定的为传给target函数的位置参数，是一个元组形式，必须有逗号1 group参数未使用，值始终为None2 3 target表示调用对象，即子进程要执行的任务4 5 args表示调用对象的位置参数元组，args=(1,2,'egon',)6 7 kwargs表示调用对象的字典,kwargs={
    'name':'egon','age':18}8 9 name为子进程的名称

方法介绍

1 p.start()：启动进程，并调用该子进程中的p.run()  2 p.run():进程启动时运行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法   3  4 p.terminate():强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放，进而导致死锁 5 p.is_alive():如果p仍然运行，返回True 6  7 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：进程的pid6 7 p.exitcode:进程在运行时为None、如果为–N，表示被信号N结束(了解即可)8 9 p.authkey:进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性，这类连接只有在具有相同的身份验证键时才能成功（了解即可）

注意:Process类,在windows中Process()必须放到#if __name__=='__main__'下

在windows中Process()必须放到#if __name__=='__main__'下

开启子进程的两种方式

开启进程的两种方式:方法一:from multiprocessing import Processimport timedef task(name):    print('%s是好人'%name)    time.sleep(2)    print('你说的啥')if __name__ == '__main__':    p = Process(target=task,args=('tom',))    p.start()    print('你是坏人')# 方法二:from multiprocessing import Processimport timeclass Myprocess(Process):    def __init__(self,name):    #定义自己的方法        super().__init__()  #继承父类的方法        self.name = name    def run(self):  #这个函数的函数名必须是run,不能是其他的,如果是其他的不会被执行该方法下面的东西        time.sleep(3)        print('%s is running'%self.name)        time.sleep(3)if __name__ == '__main__':    p = Myprocess('alex')    p.start()       #p.run()    #默认是调用的run方法    print('主')

将socket通信变成并发形式

服务端from socket import *from multiprocessing import Processdef server(ip,port):    server = socket(AF_INET, SOCK_STREAM)    server.bind((ip,port))    server.listen(5)    while True:        conn,addr = server.accept()        p = Process(target=communicate,args=(conn,))        p.start()    server.close()def communicate(conn):    while True:        try:            data = conn.recv(1024)            if not data:break            conn.send(data.upper())        except ConnectionResetError:            break    conn.close()if __name__ == '__main__':    server('127.0.0.1',8090)客户端from socket import *client = socket(AF_INET,SOCK_STREAM)client.connect(('127.0.0.1',8090))while True:    msg = input('>>')    if not msg :continue    client.send(msg.encode('utf-8'))    date = client.recv(1024)    print(date.decode('utf-8'))client.close()

虽然上面方法可以实现socket通信的并发,但是每来一个客户端,都在服务端开启一个进程,如果并发来一万个客户端,要开启一万个进程吗,你自己尝试着在你自己的机器上开一万个试试.(解决方法就是使用进程池)

process对象的join方法

join:主进程等待子进程结束

process对象的其他方法或属性

from multiprocessing import Processimport time,randomdef task():    print('孙子运行了')    time.sleep(3)def piao(name):    print('%s is piaoing' %name)    time.sleep(random.randint(1,3))    print('%s is done' %name)    p = Process(target=task)    p.start()if __name__ == '__main__':    p1 = Process(target=piao,args=('alex',),name = '***')    p1.start()    p1.join()    print(p1.name)  #-->查看进程的名称    print(p1.pid)       #查看进程的pid    # p.terminate() -->强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，用该方法需要特别小心这种情况。如果p保存了一个锁那么也将不会被释放，进而导致死锁    print('主')

#进程对象的其他方法一:terminate,is_alivefrom multiprocessing import Processimport timeimport randomclass Piao(Process):    def __init__(self,name):        self.name=name        super().__init__()    def run(self):        print('%s is piaoing' %self.name)        time.sleep(random.randrange(1,5))        print('%s is piao end' %self.name)p1=Piao('egon1')p1.start()p1.terminate()#关闭进程,不会立即关闭,所以is_alive立刻查看的结果可能还是存活print(p1.is_alive()) #结果为Trueprint('开始')print(p1.is_alive()) #结果为False

方法总结:

p.name  #-->查看进程的名称    p.pid      #查看进程的pid    p.terminate() -->强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，用该方法需要特别小心这种情况。如果p保存了一个锁那么也将不会被释放，进而导致死锁    p.join() -->主进程等子进程执行完之后再结束 - --> 等的时间就是执行时间最长的    p.is_alive()查看子进程是否还存活

进程池

提交/调用任务的方式有两种:     同步调用:提交/调用一个任务,然后就在原地等着,等到该任务执行完毕拿到结果,再执行下一行代码     异步调用:提交/调用一个任务,不在原地等着,直接执行下一行代码

这里引入进程池的概念,是为了限制并发数很多时服务器被卡死,所以就用一个池子来限制并发的数量,所有连接数都会被接收,但不会立即去执行,而是先执行进程池中的 比如进程池规定一次可以并发四个任务,也就是一次可以同时处理四个任务,当这四个任务中的任何一个被处理完后,进程池中就会立马填补一个然后再进行处理 当任务数很少时还是不使用进程池比较好

对上面描述的一个例题1

时间是3秒多一些from multiprocessing import Process,Pool该模块中的方法比较麻烦,所以使用下面的模块实现from concurrent.futures import  ProcessPoolExecutorimport time,random,osdef piao(name,n):    print('%s is piaoing %s' %(name,os.getpid()))    time.sleep(1)    return n**2if __name__ == '__main__':    p = ProcessPoolExecutor(4)  #规定每次可以并发四个任务    objs = []    start = time.time()    for i in range(10):     #这里有10个任务.        # res = p.submit(piao,'alex %s'%i,i).result() #同步调用        # print(res)        obj = p.submit(piao,'alex %s'%i,i)  #异步调用   #发出指令要开启子进程        objs.append(obj)    for obj in objs:        print(obj.result())     #result是取子进程中的结果    stop = time.time()    print(stop-start)    #关门+等    # pool.close()    # pool.join()    p.shutdown(wait=True)   #   关门+等    print('主',os.getpid())

p.submit()是发出开启子进程的指令,obj.result()是取子进程中产生的结果 p.shutdown(wait=True)   #任务数是一定的,不会再增加,也就是关门,并等待子进程结束后再执行主程序

from multiprocessing import Processimport time,random,osdef piao(name,n):    print('%s is piaoing %s' %(name,os.getpid()))    time.sleep(1)    return n**2if __name__ == '__main__':    start=time.time()    p_l=[]    for i in range(10):        p=Process(target=piao,args=('xx',1))        p_l.append(p)        p.start()    for p in p_l:        p.join()    stop=time.time()    print(stop-start)

例题1和例题2中在连接数比较少时使用2会更好,原因是2中的是所有任务一起去执行,而1中是每四个任务一起执行,要分三次,

每次使用一秒,总共要三秒多,但2中是一起执行的,是 使用了一秒多.要根据不同的环境选择不同的方法使用

多进程是实现并发的手段之一,但需要注意的问题是: 1.很明显需要并发执行的任务通常要远大于核数 2.一个操作系统不可能无限开启进程,通常有几个核就开几个进程 3.进程开启过多,效率反而会下降(开启进程是需要占用操作系统资源的,而且开启多余核数目的进程也无法做到并行) 当被操作对象数目不大时,可以直接利用multiprocessing中的Process动态动态生成多个进程,十几个还好,如果是上百个,上千个,手动的去限制进程数量却又太过繁琐,此时使用程序池是最好的 对于远程过程调用高级应用程序而言,应该使用进程池,pool可以提供指定数量的进程,供用户调用,当有新的请求提交到pool中时,如果池还没有满,就会创建一个新的 进程来执行该请求,但如果池中的进程数已经达到规定最大值,那么该请求就会等待,直到翅中有进程结束,就重用进程池中的进程