python-并发编程之多进程

multiprocessing模块

创建进程的类

Process([group [, target [, name [, args [, kwargs]]]]])，由该类实例化得到的对象，表示一个子进程中的任务（尚未启动）

强调：
1. 需要使用关键字的方式来指定参数
2. args指定的为传给target函数的位置参数，是一个元组形式，必须有逗号

1 group参数未使用，值始终为None
2 
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：进程的pid
6 
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 Process
import time
def task(name):
    print('%s是好人'%name)
    time.sleep(2)
    print('你说的啥')
if __name__ == '__main__':
    p = Process(target=task,args=('tom',))
    p.start()
    print('你是坏人')
# 方法二:
from multiprocessing import Process
import time

class 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 Process


def 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 Process
import time,random
def 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_alive
from multiprocessing import Process
import time
import random

class 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()) #结果为True

print('开始')
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  ProcessPoolExecutor
import time,random,os
def piao(name,n):
    print('%s is piaoing %s' %(name,os.getpid()))
    time.sleep(1)
    return n**2
if __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 Process
import time,random,os

def piao(name,n):
    print('%s is piaoing %s' %(name,os.getpid()))
    time.sleep(1)
    return n**2
if __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中时,如果池还没有满,就会创建一个新的
进程来执行该请求,但如果池中的进程数已经达到规定最大值,那么该请求就会等待,直到翅中有进程结束,就重用进程池中的进程