Python并发编程系列之多进程（multiprocessing）

1 引言

　　本篇博文主要对Python中并发编程中的多进程相关内容展开详细介绍，Python进程主要在multiprocessing模块中，本博文以multiprocessing种Process类为中心，通过实例代码对多进程设计到的进程间的同步机制、通信机制、数据共享机制进程池进行介绍。

2 创建进程

　　创建进程有两种方式，分别是通过定义函数的方式和通过定义类的方式。两种方式创建进程都必须通过实例化Process类。

　　Process类参数如下：

　　1） group：这一参数值始终为None，尚未启用，是为以后Python版本准备的

　　2） target：表示调用对象，即子进程要执行的任务

　　3） args：表示调用对象的位置参数元组，即target的位置参数，必须是元组，如args=(0,1,[1,2,3])

　　4） kwargs:表示调用对象的字典参数,kwargs={'name':'egon','age':18}

　　5） name：为子进程的名称

　　另外，无论用那种方式创建进程都必须有“if __name__ == '__main__':”这一行代码作为程序入口，否则会报错。

2.1 通过定义函数的方式创建进程

　　通过函数方式创建进程这一方法需在实例化进程实例时将函数名作为参数传递进去，函数的参数用一个tuple传递给进程实例：

from multiprocessing import Process

import time

def func (n):

    print("子进程开始运行:{}……".format(n))

    time.sleep(1)

    print("子进程结束运行:{}……".format(n))

if __name__ == '__main__':#创建进程执行一定要这一行代码

    print("主进程开始运行……")

    p = Process(target=func,args=(10,))  # 注册

    p.start()# 启动一个子进程

    time.sleep(1)

print("主进程结束运行……" )

2.2通过定义类的方式创建进程

 　　用类的方式创建进程时，自定义的类必须必须继承Process类，且必须覆写run方法，所有功能代码都放在run方法中：

from multiprocessing import Process

import os

class MyProcess(Process):

    def run(self):#必须覆写run方法

        print('子进程:' , self.pid)

 

if __name__=="__main__":

    p1 = MyProcess()

    p1.start()

    p2 = MyProcess()

    p2.start()

print('主进程：', os.getpid())

　　上面代码没有传递传输，如果要传递参数该怎么做呢？那就要自定义构造方法了，但是在构造方法中，一定要先调用Process类的构造方法：

from multiprocessing import Process

import os

class MyProcess(Process):

    def __init__(self , arg1 , arg2):

        super().__init__()

        self.arg1 = arg1

        self.arg2 = arg2

    def run(self):

        print('子进程:' , self.pid , self.arg1 , self.arg2)

 if __name__=="__main__":

    p1 = MyProcess('我是arg1' , '我是arg2')

    p1.start()

    p2 = MyProcess('我是arg1' , '我是arg2')

    p2.start()

print('主进程：', os.getpid())

3 Process的常用属性和方法

 　　Process类常用属性如下：

　　1）daemon：默认值为False，如果设为True，则设为守护进程。

　　2）name:进程的名称

　　3）pid：进程的pid

　　Process类常用方法如下：

　　1）start()：启动进程，并调用该子进程中的p.run() ；

　　2 ）run():进程启动时运行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法；

　　3 ）terminate():强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放，进而导致死锁；

　　4）is_alive():如果p仍然运行，返回True；

　　5）join([timeout]):主线程等待子进程终止

3.1 守护进程：daemon

　　若要将一个进程设置为守护进程，在进程start之前，将其daemon属性设置为True即可。但什么是守护进程呢？我们通过如下代码来说明，我们要通过代码实现如下效果：主进程创建p1、p2进程，之后立马沉睡5秒，p1进程设置为守护进程，进程p1每隔1秒打印一条语句，进程p2打印一条语句后立马沉睡10秒。代码如下：

import time

import os

from multiprocessing import Process

def func():

    i = 1

    while True:

        time.sleep(1)

        print('{}--子进程p1正在执行,pid:{}'.format(i , os.getpid()))

        i+=1

def func2():

    print('子进程p2开始执行,pid:{}'.format(os.getpid()))

    time.sleep(10)

    print('子进程p2结束执行,pid:{}'.format(os.getpid()))

if __name__=='__main__':

    print('主进程代码开始运行,pid:{}'.format(os.getpid()))

    p = Process(target=func)

    p.daemon = True # 设置为守护进程

    p.start()

    p2 = Process(target=func2)

    p2.start()

    time.sleep(5)

print('主进程代码运行完了,pid:{}'.format(os.getpid()))

　　运行结果：

　　主进程代码开始运行,pid:11608

　　子进程p2开始执行,pid:7260

　　1--子进程p1正在执行,pid:1060

　　2--子进程p1正在执行,pid:1060

　　3--子进程p1正在执行,pid:1060

　　4--子进程p1正在执行,pid:1060

　　主进程代码运行完了,pid:11608

　　子进程p2结束执行,pid:7260

　　从运行结果字面上似乎看不出什么，因为区别在于输出时间上。在主进程运行的那5秒时间（输出“主进程代码运行完了,pid:11608”之前），p1进程确实可以每隔1秒输出一条语句，但是主进程结束那5秒后，p1不在输出，且在任务管理器中也可以查看到，p1进程也已经死亡，主进程代码虽然运行完了，但依然存活，这时候p2进程依然还在沉睡，10秒后，p2进程打印“子进程p2结束执行,pid:7260”，然后主进程和p2进程一起死亡。

　　可以得出结果，守护进程依附于主进程代码，只要主进程代码运行完了，那么无论守护进程代码是否运行完，守护进程都会结束。另外，守护进程不能创建自己的子进程。

3.2 进程终结于存活检查：terminate()与is_alive()

　　terminate()与is_alive()都是由进程实例调用，分别用来终结一个进程、检查一个进程是否依然存活：

import time

import os

from multiprocessing import Process

 

def func():

    i = 1

    while True:

        time.sleep(1)

        print('{}--子进程p1正在执行,pid:{}'.format(i , os.getpid()))

        i+=1


if __name__=='__main__':

    p = Process(target=func)

    p.start()

    time.sleep(3)

    p.terminate() # 终结进程p

    print(p.is_alive()) # 检查p是否依然存活

    time.sleep(1)

    print(p.is_alive())

　　输出结果：

　　主进程代码开始运行,pid:13164

　　1--子进程p1正在执行,pid:8896

　　2--子进程p1正在执行,pid:8896

　　True

　　False

　　主进程代码运行完了,pid:13164

　　为什么结束之后第一次调用is_alive()方法输出的是True呢？因为terminate()方法终结一个进程时操作系统需要一定的响应时间，所以可能会有延迟。

3.3 join()方法

　　join方法功能是阻塞当前所在进程（例如下面的主进程），等到被join的进程（下面的进程p1）结束之后，回到当前进程继续执行。

from multiprocessing import Process

import time

def func1 ():

    print("进程1开始运行……")

    for i in range(3):

        time.sleep(1)

        print("进程1运行了{}秒……".format(i+1))

    print("进程1结束运行……")

 

def func2 ():

    print("进程2开始运行……")

    for i in range(6):

        time.sleep(1)

        print("进程2运行了{}秒……".format(i+1))

    print("进程2结束运行……")

 

if __name__ == '__main__':

    print("主进程开始运行……")

    p1 = Process(target=func1)

    p2 = Process(target=func2)

    p1.start()

    p2.start()

    time.sleep(1)

    p1.join()

    # p2.join()

    print("主进程结束运行……" )

　　上述代码不进行join、分别是对进程1、进程2进行join的运行结果，发现，主进程会等待被join的进程运行完才会继续执行join下面的代码。

4 进程间的同步控制

4.1 进程锁：Lock

　　当多个进程对同一资源进行IO操作时，需要对资源“上锁”，否则会出现意外结果。上锁之后，同一件就只能有一个进程运行上锁的代码块。例如有一个txt文件，里面内容是一个数字10，我们要用多进程去读取这个文件的值，然后每读一次，让txt中的这个数字减1，不加锁时代码如下：

from multiprocessing import Process

from multiprocessing import Lock

import time

import os

def func ():

    if os.path.exists('num.txt'):

        with open('num.txt' , 'r') as f:

            num = int(f.read())

            num -= 1

        time.sleep(1)

        with open('num.txt' , 'w') as f:

            f.write(str(num))

    else:

        with open('num.txt' , 'w') as f:

            f.write('10')

if __name__ == '__main__':

    print("主进程开始运行……")

    p_list = []

    for i in range(10):

        p = Process(target=func)

        p_list.append(p)

        p.start()

    for p in p_list:

        p.join()

    with open('num.txt', 'r') as f:

        num = int(f.read())

    print('最后结果为：{}'.format(num))

    print("主进程结束运行……" )

　　输出结果：

　　主进程开始运行……

　　最后结果为：8

　　主进程结束运行……

　　虽然我们用了10个进程读取并修改txt文件，但最后的值却不是1。这正是多进程共同访问资源造成混乱造成的。要达到预期结果，就要给资源上锁：

from multiprocessing import Process

from multiprocessing import Lock

import time

import os

def func (lock):

    if os.path.exists('num.txt'):

        lock.acquire()

        with open('num.txt' , 'r') as f:

            num = int(f.read())

            num -= 1

        time.sleep(1)

        with open('num.txt' , 'w') as f:

            f.write(str(num))

        lock.release()

    else:

        with open('num.txt' , 'w') as f:

            f.write('10')

if __name__ == '__main__':

    print("主进程开始运行……")

    lock = Lock()

    p_list = []

    for i in range(10):

        p = Process(target=func , args=(lock,))

        p_list.append(p)

        p.start()

    for p in p_list:

        p.join()

    with open('num.txt', 'r') as f:

        num = int(f.read())

    print('最后结果为：{}'.format(num))

    print("主进程结束运行……" )

　　输出结果：

　　主进程开始运行……

　　最后结果为：0

　　主进程结束运行……

　　果然，用了进程锁之后获得了预料中的结果。但是，如果你运行了上面两块代码你就会发现，加了锁之后，程序明显变慢了很多，因为程序成了串行的了，当然好处是数据安全有保证。

4.2 信号量： Semaphore

 　　锁同时只允许一个线程更改数据，而信号量是同时允许一定数量的进程更改数据 。假如有一下应用场景：有100个人吃饭，但只有一张餐桌，只允许做3个人，没上桌的人不允许吃饭，已上桌吃完饭离座之后，下面的人才能抢占桌子继续吃饭，如果不用信号量，肯定是100人一窝蜂一起吃饭：

from multiprocessing import Process

import time

import random


def fun(i):

    print('{}号顾客上座，开始吃饭'.format(i))

    time.sleep(random.randint(3, 8))

    print('{}号顾客吃完饭了，离座'.format(i))


if __name__=='__main__':

    for i in range(100):

        p = Process(target=fun, args=(i,))

        p.start()

　　输出结果：

　　1号顾客上座，开始吃饭

　　2号顾客上座，开始吃饭

　　0号顾客上座，开始吃饭

　　3号顾客上座，开始吃饭

　　4号顾客上座，开始吃饭

　　5号顾客上座，开始吃饭

　　6号顾客上座，开始吃饭

　　7号顾客上座，开始吃饭

　　……

 　　用了信号量，实现了轮流吃饭，每次只有3个人吃饭：

from multiprocessing import Process

import time

import random

from multiprocessing import Semaphore

def fun(i , sem):

    sem.acquire()

    print('{}号顾客上座，开始吃饭'.format(i))

    time.sleep(random.randint(3, 8))

    print('{}号顾客吃完饭了，离座'.format(i))

    sem.release()


if __name__=='__main__':

    sem = Semaphore(3)

    for i in range(10):

        p = Process(target=fun, args=(i,sem))

        p.start()

         输出结果：

　　1号顾客上座，开始吃饭

　　0号顾客上座，开始吃饭

　　2号顾客上座，开始吃饭

　　1号顾客吃完饭了，离座

　　3号顾客上座，开始吃饭

　　2号顾客吃完饭了，离座

　　4号顾客上座，开始吃饭

　　0号顾客吃完饭了，离座

　　5号顾客上座，开始吃饭

　　……

　　上面只是输出结果的一部分，不过已经看出，在同一时刻，只有3位顾客在吃饭（3个进程占用资源），且只有在一位顾客离座之后才会有下一个顾客入座（一个进程结束对资源的占用，下一个进程才能访问资源）。事实上，Semaphore的作用也类似于锁，只不过在锁机制上添加了一个计数器，允许多个人拥有“钥匙”。

4.3 事件：Event

　　python进程的事件用于主进程控制其他子进程的执行，Event类有如下几个主要方法：

　　1）wait()    插入在进程中插入一个标记（flag）默认为 False，当 flag为False时，程序会停止运行进入阻塞状态；

　　2）set()     使flag为True，程序会进入非阻塞状态

　　3）clear()      使flag为False，程序会停止运行，进入阻塞状态 

　　4）is_set()   判断flag  是否为True，是的话返回True，不是则返回False

　　有如下需求：获取当前时间的秒数的个位数，如果小于5，则设置子进程阻塞，如果大于5则设置子进程非阻塞。代码如下：

from multiprocessing import Event, Process

import time

from datetime import datetime

def func(e):

    print('子进程：开始运行……')

    while True:

        print('子进程：现在事件秒数是{}'.format(datetime.now().second))

        e.wait()  # 阻塞等待信号  这里插入了一个flag  默认为 False

        time.sleep(1)

 

if __name__ == '__main__':

    e = Event()

    p = Process(target=func, args=(e,))

    p.start()

    for i in range(10):

        s = int(str(datetime.now().second)[-1])#获取当前秒数的个位数

        if s < 5:

            print('子进程进入阻塞状态')

            e.clear()  # 使插入的flag为False 进程进入阻塞状态

        else:

            print('子进程取消阻塞状态')

            e.set()  # 使插入的flag为True，进程进入非阻塞状态

        time.sleep(1)

    e.set()

    time.sleep(3)

    p.terminate()

    print("主进程运行结束……")

　　输出结果：

　　子进程取消阻塞状态

　　子进程：开始运行……

　　子进程：现在事件秒数是58

　　子进程取消阻塞状态

　　子进程：现在事件秒数是59

　　子进程取消阻塞状态

　　子进程：现在事件秒数是0

　　子进程进入阻塞状态

　　子进程：现在事件秒数是1

　　子进程进入阻塞状态

　　子进程进入阻塞状态

　　子进程进入阻塞状态

　　子进程进入阻塞状态

　　子进程取消阻塞状态

　　子进程取消阻塞状态

　　子进程：现在事件秒数是6

　　子进程：现在事件秒数是7

　　子进程：现在事件秒数是8

　　子进程：现在事件秒数是9

　　主进程运行结束……

5 进程间通信

5.1进程队列：Queue

　　常用方法：　　

　　get( [ block [ ,timeout ] ] ) ：返回q中的一个项目。如果q为空，此方法将阻塞，直到队列中有项目可用为止。block用于控制阻塞行为，默认为True. 如果设置为False，将引发Queue.Empty异常（定义在Queue模块中）。timeout是可选超时时间，用在阻塞模式中。如果在制定的时间间隔内没有项目变为可用，将引发Queue.Empty异常。

　　get_nowait( ) ：同get(False)方法。

　　put(item [, block [,timeout ] ] ) ：将item放入队列。如果队列已满，此方法将阻塞至有空间可用为止。block控制阻塞行为，默认为True。如果设置为False，将引发Queue.Empty异常（定义在Queue库模块中）。timeout指定在阻塞模式中等待可用空间的时间长短。超时后将引发Queue.Full异常。

　　qsize() ：返回队列中目前项目的正确数量。此函数的结果并不可靠，因为在返回结果和在稍后程序中使用结果之间，队列中可能添加或删除了项目。在某些系统上，此方法可能引发NotImplementedError异常。

　　empty() ：如果调用此方法时队列为空，返回True。如果其他进程或线程正在往队列中添加项目，结果是不可靠的。也就是说，在返回和使用结果之间，队列中可能已经加入新的项目。

　　full() ：如果q已满，返回为True. 由于线程的存在，结果也可能是不可靠的。

　　close() ：关闭队列，防止队列中加入更多数据。调用此方法时，后台线程将继续写入那些已入队列但尚未写入的数据，但将在此方法完成时马上关闭。如果队列被垃圾收集，将自动调用此方法。关闭队列不会在队列使用者中生成任何类型的数据结束信号或异常。例如，如果某个使用者正被阻塞在get（）操作上，关闭生产者中的队列不会导致get（）方法返回错误。

　　cancel_join_thread()：不会再进程退出时自动连接后台线程。这可以防止join_thread()方法阻塞。

　　join_thread()：连接队列的后台线程。此方法用于在调用close()方法后，等待所有队列项被消耗。默认情况下，此方法由不是队列的原始创建者的所有进程调用。调用cancel_join_thread()方法可以禁止这种行为。

from multiprocessing import Process

from multiprocessing import Queue

import random

import os

# 向queue中输入数据的函数

def inputQ(queue ):

    info = random.randint(1,100)

    queue.put(info)

    print('进程{}往队列中存了一个数据：{}'.format(os.getpid() , info))

# 向queue中输出数据的函数

def outputQ(queue):

    info = queue.get()

    print ('进程{}从队列中取出一个数据：{}'.format(os.getpid() , info))

if __name__ == '__main__':

    queue = Queue(5)

    lst_1  = []

    lst_2 = []

    for i in range(3):

        process = Process(target=inputQ,args=(queue,))

        process.start()

        lst_1.append(process)

    # 输出进程

    for i in range(2):

        process = Process(target=outputQ ,args=(queue,))

        process.start()

        lst_1.append(process)

    for p in lst_2:

        p.join()

    for p in lst_2:

        p.join()

　　另外，在使用进程池Pool时，使用Queue会出错，需要使用Manager.Queue。

5.2 管道：Pipe

　　创建管道方法：

　　Pipe([duplex]):在进程之间创建一条管道，并返回元组（conn1,conn2）,其中conn1，conn2表示管道两端的连接对象，强调一点：必须在产生Process对象之前产生管道。dumplex:默认管道是全双工的，如果将duplex射成False，conn1只能用于接收，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, Pipe

def f(conn):

    conn.send('主进程，你好呀！')  # 发送数据给主进程

    print('子进程收到主进程发来的数据：{}'.format(conn.recv()))

    conn.close()  # 关闭

if __name__ == '__main__':

    parent_conn, child_conn = Pipe()  #Pipe是一个函数，返回的是一个元组

    p = Process(target=f, args=(child_conn,))  # 创建一个子进程

    p.start()

    print("主进程收到子进程发来的数据：{}".format(parent_conn.recv()))

    parent_conn.send('子进程，你好啊！')  # 发送数据给子进程

p.join()

6 进程间的数据共享

6.1 进程之间的数据隔离

　　进程间的变量是无法共享的，就算是全局变量也不行：

from multiprocessing import Process

import os

def func():

    global n

    n = 10

    print('子进程pid:{},n:{}'.format(os.getppid() , n))

if __name__=='__main__':

    n=100

    print('主进程pid:{},n:{}'.format(os.getppid(), n))

    p = Process(target=func)

    p.start()

    p.join()

    print('主进程中输出n：{}'.format(n))

　　但是，如果建成间要进行数据共享，要怎么做呢？

6.2 Manager

　　进程间的数据共享还可以通过Manager

import os

from multiprocessing import Manager,Process

# 定义了一个foo函数,接收一个字典和一个列表

def foo(d, l):

    # 字典和列表都放进程ID

    d[os.getpid()] = os.getpid()

    l.append(os.getpid())

if __name__ == '__main__':

    # 生成Manager对象

    manager = Manager()

    d = manager.dict()

    l = manager.list(range(3))

    # 10个进程分别join

    p_list = []

    for i in range(10):

        p = Process(target=foo, args=(d, l))

        p.start()

        p_list.append(p)

    for res in p_list:

        res.join()
 
    # 打印字典和列表

    print(d)

    print(l)

7 进程池

　　为什么要用进程池呢？如果我们有几百上千个任务需要自行，那么按照之前的做法，我们就要创建几百上千个进程，每一个进程都要占用一定的内存空间，进程间的切换也费时，系统开销很大，而且，难道这成千上百个进程能同时并发执行的有几个呢？其实也就那么几个子，所以，根本没必要创建那么多进程。那么怎么办呢？那就创建进程池。进程池里有固定数量的进程，每次执行任务时都从进程池中取出一个空闲进程来执行，如果任务数量超过进程池中进程数量，那么就等待已经在执行的任务结束之后，有进程空闲之后再执行，也就是说，同一时间，只有固定数量的进程在执行，这样对操作系统得压力也不会太大，效率也得到保证。

import os,time,random

from multiprocessing import Pool

def func1(n):

    print('任务{}开始执行，进程为：{}'.format(n,os.getpid()))

    time.sleep(random.randint(1,4))

    print('任务{}结束执行，进程为：{}'.format(n,os.getpid()))

if __name__ == '__main__':

    p=Pool(3) #c创建一个进程池，里面有三个进程

    for i in range(10):

        res=p.apply(func1,args=(i,))

　　输出结果：

　　任务0开始执行，进程为：14380

　　任务0结束执行，进程为：14380

　　任务1开始执行，进程为：14772

　　任务1结束执行，进程为：14772

　　任务2开始执行，进程为：10972

　　任务2结束执行，进程为：10972

　　任务3开始执行，进程为：14380

　　任务3结束执行，进程为：14380

　　任务4开始执行，进程为：14772

　　任务4结束执行，进程为：14772

　　可以看出，自始至终都只有3个进程在执行任务，但这些任务都是被同步执行的，如果要异步执行呢：

import os,time,random
from multiprocessing import Pool


def func1(n):
    print('任务{}开始执行，进程为：{}'.format(n,os.getpid()))

    time.sleep(random.randint(1,4))

    print('任务{}结束执行，进程为：{}'.format(n,os.getpid()))

if __name__ == '__main__':

    p=Pool(3) #c创建一个进程池，里面有三个进程

    for i in range(5):

        res=p.apply_async(func1,args=(i,))

    p.close()#一定要关闭

    p.join()#一定要使用join，不然进程池里的进程没来得及执行，主进程结束了，子进程也都跟着结束。

　　每个任务其实也都可以有返回值：

from multiprocessing import Pool

def func1(n):

    return n**2

if __name__ == '__main__':

    p=Pool(3)

    result_lst = []

    for i in range(5):

        result_lst.append(p.apply_async(func1,args=(i,)))#异步

    p.close()

    p.join()

    for result in result_lst:

        print(result.get())#如果是同步，就不用get了，直接用result获取

8 总结

　　至此，Python并发编程中多进程部分就总结完了，花了几天时间，参考了很多资料。

　　参考资料：

　　http://www.cnblogs.com/Eva-J/articles/8253549.html

　　https://www.jianshu.com/p/8195c023915e