并发编程之多进程操作

Posted 2020-11-09 xuzewen

一 multiprocessing 模块介绍

?? Python 中的多线程无法利用多核优势，如果想要充分地使用多核CPU的资源（os.cpu_count()查看），在python中大部分情况需要使用多进程。

??multiprocessing模块用来开启子进程并在子进程中执行我们定制的任务(如函数)，该模块与多线程模块threading的编程接口相似。

?? multiprocessing模块的功能众多：支持子进程、通信和共享数据、执行不同形式的同步，提供了Process、Queue、Pipe、Lock等组件。

注：与线程不同，进程没有任何共享状态，进程修改的数据，改动仅限于该进程内。

二 Process 类的介绍

Process 类用来描叙一个进程对象，Process对象可以创建子进程，但Process对象不是进程

Process 类构造方法：

multiprocessing.Process(group=None, target=None, name=None, args=(), kwargs={}) 由该类实例化得到的对象，表示子进程中的任务(尚未启动)

注意:
1、需要使用关键字的方式指定参数
2、args 指定的为传给 target 函数的位置参数，是一个元祖形式，其内参数后面必须要有逗号

参数方法

1 group 参数未使用，值为默认None
2 target 表示调用对象，即子进程要执行的任务
3 args 表示调用对象的位置参数，形式为元祖，例：args=(1,2,‘abc‘,)
4 kwargs 表示调用对象的字典， kwargs={‘name‘:‘abc‘, ‘age‘:18}
5 name 为子进程的名称，可以自定义名称

使用方法

# 实例化Process,得到一个对象
p = multiprocess.Process(target=test) 

# 方法：
1 p.start()：启动进程，并调用该子进程中的 p.run()
2 p.run()：进程启动时运行的方法，正是它去调用 tartget 指定的函数
3 p.terminate()：强制终止进程p，不会进行任务清理操作，如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况，如果p还保存了一个锁，那么也将不会被释放，进而导致死锁
4 p.is_alive()：如果p仍热运行，返回True，检测进程是否存在
5 p.join([timemout])：主进程等待p终止（强调：是主线程处于等的状态，而p是处于运行的状态）。timeout是可选的超时时间，需要强调的是，p.join只能join住start开启的进程，而不能join住run开启的进程

属性介绍

1 p.daemon：默认值为False，如果设置为True，代表p为后台运行的守护进程，当p的父进程终止时，p也随之终止，并且设定为True后，p不能创建自己的新进程，必须在p.start()之前设置
2 p.name：进程的别名
3 p.pid： 进程的PID
4 p.exitcode：进程在运行时为None，如果为-N，表示被信号N结束
5 p.authkey：进程的身份验证键，默认是由os.urandom()随机生成的32字符的字符串，这个键的用途是为涉及网络连接的底层进程间通信提供安全性，这类连接只有在具有相同的身份验证键时才能成功

三 Process 类的使用

注意：在windows中Process()必须放在 if name == ‘main‘ 下

创建并开启子进程的两种方式

"""
方式一：调用 multiprocessing 内 Process 类功能，能得到一个对象，通过对象开启多进程
"""
import time
import random
form multiprocessing import Process

def test(name):
    print(‘%s test start‘ % name)
    time.sleep(random.randrange(1,5))
    print(‘%s test stop‘ % name)

# windwos 下
if __name__ == ‘__main__‘:
    p1=Process(target=test, args=(‘abc‘,))
    p2=Process(target=test, args=(‘123‘,))

    p1.start()
    p2.start()
    print(‘主线程‘)

"""
方式二：自定义子类继承Process类
"""

class MyProcess(Process):
    def __init__(self, name):
        super().__init__()
        self.name = name

    ‘‘‘ 开启进程函数名必须要为 run ‘‘‘
    def run(self):
        print(‘%s is running‘ % self.name)
        time.sleep(3)
        print(‘%s is done‘ % self.name)

if __name__ == ‘__main__‘:
    p = MyProcess(‘abc‘)
    p.start()   # 本质就是在调用 MyProcess 类内的 run 方法

  print(‘主‘)

进程之间的内存空间时隔离的

from multiprocessing import Process
n = 100

def work():
    global n
    n = 0
  print(‘子进程内: ‘, n)

if __name__ == ‘__main__‘:
    p = Process(target=work)
    p.start()
    print(‘主进程内: ‘, n)

"""
执行结果：
主进程内:  100
子进程内:  0
"""

Process对象的join方法

join()方法表示等待子进程结束以后在继续往下运行

class Piao(Process):
    def __init__(self, name):
        super().__init__()
        self.name = name

    def run(self):
        print(‘%s is piaoing‘ % self.name)
        time.sleep(random.randrange(1, 3))
        print(‘%s is piao end‘ % self.name)

if __name__ == ‘__main__‘:
    p = Piao(‘abc‘)
    p.start()
    p.join() # 等待p进程运行完毕之后在继续向下执行
    print(‘开始‘)
"""
执行结果:
abc is piaoing
abc is piao end
开始
"""

Process对象的其他方法或属性

"""
进程对象的其他方法：
Process.terminate() 发送信号给操作系统停止进程
Process.is_alive() 判断进程是否存活
Process.name  查看进程名
Process.pid   查看进程PID
"""
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(‘abc‘)
p1.start()

print(p.name) # 查看进程用户名
print(p.pid)  # 查看进程的pid
p1.terminate() #关闭进程,不会立即关闭,所以is_alive立刻查看的结果可能还是存活
print(p1.is_alive()) # 结果为True

print(‘开始‘)
print(p1.is_alive()) # 结果为False

四 守护进程

主进程创建守护进程：

1 守护进程会在主进程执行结束之后终止
2 守护进程内无法在开启子进程，否则抛出异常

如果我们有两个任务需要并发执行，那么开一个主进程和一个子进程分别去执行就ok了
如果子进程的任务在主进程任务结束后就没有存在的必要了，那么该子进程应该在开启前就被设置成守护进程。
主进程代码运行结束，守护进程随即终止
注：进程之间是相互对立的，主进程代码运行结束，守护进程随即终止

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,3))
        print(‘%s is piao end‘ %self.name)

p=Piao(‘egon‘)
p.daemon=True #一定要在p.start()前设置,设置p为守护进程,禁止p创建子进程,并且父进程代码执行结束,p即终止运行
p.start()
print(‘主‘)

五 进程同步(互斥锁)

互斥锁：multiprocessing模块下 Lock 类

lock.acquire() #加锁
lock.release() #释放锁

注：有加锁就必须要有解锁，不然会形成死锁

进程之间数据不共享,但是共享同一套文件系统,所以访问同一个文件,或同一个打印终端,是没有问题的,

而共享带来的是竞争，竞争带来的结果就是错乱，如何控制，就是加锁处理

#并发运行,效率高,但竞争同一打印终端,带来了打印错乱
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):
        p=Process(target=work)
        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__‘:
    lock=Lock()
    for i in range(3):
        p=Process(target=work,args=(lock,))
        p.start()

互斥锁与join的区别：join是将一个任务整理变成串行，而互斥锁的好处则是可以将一个任务中的某一段代码串行

小结

# 加锁保证了多个进程修改同一块数据时，同一时间只能由一个任务可以进行修改，即串行修改，牺牲了速度却保证了数据安全
虽然可以用文件共享数据实现进程间通信，但问题是：
1.效率低（共享数据基于文件，而文件是硬盘上的数据）
2.需要自己加锁处理

#因此我们最好找寻一种解决方案能够兼顾：
1、效率高（多个进程共享一块内存的数据）
2、帮我们处理好锁问题。
这就是mutiprocessing模块为我们提供的基于消息的IPC通信机制：队列和管道。

六 队列与管道

进程彼此之间相互隔离，要实现进程间通信(IPC), multiprocessing 模块执行两种形式：队列和管道，这两种方式都是通过消息传递的

队列和管道：

1、队列和管道都是将数据存放于内存中
2、队列又是基于(管道+锁)实现的，可以从复杂的所问题中解脱出来

创建队列的类(底层就是以管道和锁定的方式实现)

multiprocess 模块内 Queue 类
multiprocess.Queue(maxsize): 创建共享的进程队列，Queue 时多进程安全的队列，可以使用 Queue 实现多进程之间的数据传递

参数介绍：
maxsize 是队列中允许的最大项数，不写则表示无限制
但要明确：
    1、队列内存放的是消息而非大数据
    2、队列占用的是内存空间，因而 maxsize 即便是无大小限制也要受限于内存大小

主要方法介绍

q=Queue(3)

1 q.put 方法是以插入数据到队列中。
    put方法还有两个可选参数：blocked 和 timeout
        如果blocked为True（默认值），并且timeout为正值，该方法会阻塞timeout指定的时间，直到该队列有剩余的空间。如果超时，会抛出Queue.Full异常。如果blocked为False，但该Queue已满，会立即抛出Queue.Full异常。
        
2 q.get 方法可以从队列读取并且删除一个元素。
    get方法也有两个可选参数：blocked 和 timeout
        如果blocked为True（默认值），并且timeout为正值，那么在等待时间内没有取到任何元素，会抛出Queue.Empty异常。如果blocked为False，有两种情况存在，如果Queue有一个值可用，则立即返回该值，否则，如果队列为空，则立即抛出Queue.Empty异常.

3 q.empty()：调用此方法时q为空则返回True，该结果不可靠，比如在返回True的过程中，如果队列中又加入了项目。

4 q.full()：调用此方法时q已满则返回True，该结果不可靠，比如在返回True的过程中，如果队列中的项目被取走。

5 q.qsize():返回队列中目前项目的正确数量，结果也不可靠，理由同q.empty()和q.full()一样

实例

from multiprocessing import Process,Queue

q=Queue(3)

#put ,get ,put_nowait,get_nowait,full,empty
q.put(1)
q.put(2)
q.put(3)
print(q.full()) #满了
# q.put(4) #再放就阻塞住了

print(q.get())
print(q.get())
print(q.get())
print(q.empty()) #空了
# print(q.get()) #再取就阻塞住了

七 生产者消费者模型

在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。

为甚麽要使用生产者消费者模型

?? 生产者指的是生产数据的任务，消费者指的是处理数据的任务，在并发编程中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。

甚麽是生产者和消费者模式

??生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

这个阻塞队列就是用来给生产者和消费者解耦的

基于队列实现生产者消费者模型

from multiprocessing import Process,Queue
import time,random,os

def consumer(q,name):
    while True:
        res=q.get()
        time.sleep(random.randint(1,3))
        print(‘\033[43m%s 吃 %s\033[0m‘ %(name,res))

def producer(q,name,food):
    for i in range(3):
        time.sleep(random.randint(1,3))
        res=‘%s%s‘ %(food,i)
        q.put(res)
        print(‘\033[45m%s 生产了 %s\033[0m‘ %(name,res))

if __name__ == ‘__main__‘:
    q=Queue()
    #生产者们:即厨师们
    p1=Process(target=producer,args=(q,‘egon‘,‘包子‘))

    #消费者们:即吃货们
    c1=Process(target=consumer,args=(q,‘alex‘))

    #开始
    p1.start()
    c1.start()
    print(‘主‘)
"""
执行结果：
主
egon 生产了 包子0
egon 生产了 包子1
alex 吃 包子0
alex 吃 包子1
egon 生产了 包子2
alex 吃 包子2
"""

此时的问题是主进程永远不会结束，原因是：生产者p在生产完后就结束了，但是消费者c在取空了q之后，则一直处于死循环中且卡在q.get()这一步。

解决方式无非是让生产者在生产完毕后，往队列中再发一个结束信号，这样消费者在接收到结束信号后就可以break出死循环

from multiprocessing import Process,Queue
import time,random,os
def consumer(q,name):
    while True:
        res=q.get()
        if res is None:break
        time.sleep(random.randint(1,3))
        print(‘\033[43m%s 吃 %s\033[0m‘ %(name,res))

def producer(q,name,food):
    for i in range(3):
        time.sleep(random.randint(1,3))
        res=‘%s%s‘ %(food,i)
        q.put(res)
        print(‘\033[45m%s 生产了 %s\033[0m‘ %(name,res))

if __name__ == ‘__main__‘:
    q=Queue()
    #生产者们:即厨师们
    p1=Process(target=producer,args=(q,‘egon‘,‘包子‘))

    #消费者们:即吃货们
    c1=Process(target=consumer,args=(q,‘alex‘))

    #开始
    p1.start()
    c1.start()

    p1.join()
    q.put(None)
    print(‘主‘)

# 注：结束信号None，不一定是由生产者发，主进程里同样可以发，但主进程需要等生产者结束后才应该发送信号

还有另外一种方式提供了发送结束信号这种机制

# JoinableQueue(maxsize)：这就像是一个Queue对象，但队列允许项目的使用者通知生成者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的

# 参数介绍
maxsize 是队列中允许的最大项数，省略则无大小限制

# 方法介绍
JoinableQueue的实例p除了与Queue对象相同的方法之外还具有：

q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发ValueError异常

q.join():生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done（）方法为止

基于JoinableQueue实现生产者消费者模型

from multiprocessing import Process,JoinableQueue
import time,random,os

def consumer(q,name):
    while True:
        res=q.get()
        time.sleep(random.randint(1,3))
        print(‘\033[43m%s 吃 %s\033[0m‘ %(name,res))
        q.task_done() #发送信号给q.join()，说明已经从队列中取走一个数据并处理完毕了

def producer(q,name,food):
    for i in range(3):
        time.sleep(random.randint(1,3))
        res=‘%s%s‘ %(food,i)
        q.put(res)
        print(‘\033[45m%s 生产了 %s\033[0m‘ %(name,res))
    q.join() #等到消费者把自己放入队列中的所有的数据都取走之后，生产者才结束

if __name__ == ‘__main__‘:
    q=JoinableQueue() #使用JoinableQueue()

    #生产者们:即厨师们
    p1=Process(target=producer,args=(q,‘egon1‘,‘包子‘))
    p2=Process(target=producer,args=(q,‘egon2‘,‘骨头‘))
    p3=Process(target=producer,args=(q,‘egon3‘,‘泔水‘))

    #消费者们:即吃货们
    c1=Process(target=consumer,args=(q,‘alex1‘))
    c2=Process(target=consumer,args=(q,‘alex2‘))
    c1.daemon=True
    c2.daemon=True

    #开始
    p1.start()
    p2.start()
    p3.start()
    c1.start()
    c2.start()

    p1.join()
    p2.join()
    p3.join()
    #1、主进程等生产者p1、p2、p3结束
    #2、而p1、p2、p3是在消费者把所有数据都取干净之后才会结束
    #3、所以一旦p1、p2、p3结束了，证明消费者也没必要存在了，应该随着主进程一块死掉，因而需要将生产者们设置成守护进程
    print(‘主‘)

总结

1、程序中由两类角色
    一类负责生产数据(生产者)
    一类负责处理数据(消费者)

2、引入生产者消费者模型而解决的问题是：
    平衡生产者和消费者之间的速度差，从而提高程序整体处理数据的速度
    程序解开耦合
    
3、如果实现：
    生产者<-->队列<--->消费者