本文首发于公众号:Hunter后端

原文链接:Python笔记六之多进程

在 Python 里,我们使用 multiprocessing 这个模块来进行多进程的操作。

multiprocessing 模块通过创建子进程的方式来运行多进程,因此绕过了 Python 里 GIL 的限制,可以充分利用机器上的多个处理器。

1、多进程使用示例

多进程的使用方式和多线程的方式类似,这里使用到的是 multiprocessing.Process 类,下面是一个简单的示例:

from multiprocessing import Process
import time

def f(x):
    if x % 2 == 1:
        time.sleep(x+1)
    print(x * x)
    return x * x


def test_multi_process():
    processes = []

    for i in range(5):
        processes.append(Process(target=f, args=(i,)))

    for p in processes:
        p.start()

    for p in processes:
        p.join(0.5)

    for p in processes:
        print(p, p.is_alive(), p.pid, p._parent_pid)


if __name__ == "__main__":
    test_multi_process()

在上面的示例中,test_multi_process() 函数里使用多进程的方式调用 f 函数,和多线程的调用方式一致,通过 start() 方法启动进程活动,使用 join() 方法阻塞调用其的进程。

接下来介绍一下 multiprocessing.Process 的一些方法和属性。

1. run()

表示进程活动的方法,可以在子类中重载此方法,比如多线程笔记的操作里重写 run() 对函数执行报错进行了处理,并返回了执行结果

2. start()

启动进程活动,将对象的 run() 方法在一个单独的进程中调用

3. join()

阻塞调用 join() 方法的进程,在上面的示例中也就是父进程,默认值为 None,也就表示阻塞操作。

如果设置为其他正数值,那么则最多会阻塞多少秒,比如上面的示例为 0.5 秒,如果超时,那么父进程则会继续往后执行。

比如上面的示例输出结果如下:

0
4
16
<Process name='Process-1' pid=6600 parent=24248 stopped exitcode=0> False 6600 24248
<Process name='Process-2' pid=4368 parent=24248 started> True 4368 24248
<Process name='Process-3' pid=13024 parent=24248 stopped exitcode=0> False 13024 24248
<Process name='Process-4' pid=3288 parent=24248 started> True 3288 24248
<Process name='Process-5' pid=16880 parent=24248 stopped exitcode=0> False 16880 24248
1
9

在打印每个进程的信息时,f() 函数内部进行 sleep 的进程还没有执行结束,但是进程已经超时了,所以不再阻塞父进程向下执行。

4. is_alive()

上面有打印出信息,返回布尔值,表示该进程是否还活着。

5. pid 和 parent_pid

上面使用 .pid 和 ._parent_pid 属性打印出了每个进程的 id 和其父进程的 id。

2、进程池

进程使用的对象是 multiprocessing.pool.Pool()。

接受 processes 参数为进程数,表示要使用的工作进程数目,如果不传入,则默认使用 cpu 的核数,根据 os.cpu_count() 获取。

接下来分别使用示例介绍 multiprocessing.pool 下的几个调用方法,进程池的使用可以使用 map() 和 starmap() 两个函数。

1. map()

map() 接受两个参数,func 表示多进程要执行的函数,iterable 表示要执行的 func 函数输入的参数的迭代对象。

这里需要注意一下,map() 函数使用的 func 函数只能接受一个参数,比如我们前面定义的 f 函数,下面是其使用示例:

def f(x):
    return x * x


def test_pool_map():
    with Pool(processes=4) as pool:
        results = pool.map(func=f, iterable=range(10))

    print(results)

2. starmap()

starmap() 函数与 map() 使用方法类似,但是 iterable 迭代参数的元素是 func 函数的多个参数,比如我们想要对下面的 add() 函数使用多进程:

def f_add(x, y):
    return x + y

它的调用方式如下:

def test_pool_starmap():
    with Pool(processes=4) as pool:
        results = pool.starmap(func=f_add, iterable=zip(range(6), range(6, 12)))
    
    print(results)

这里返回的 results 是一个列表,元素是每个进程执行的函数的返回结果。

3、进程间交换对象

前面介绍了,多进程的运行方式是通过建立子进程的形式来操作,而不同进程间数据是不共享的,这一点不同于多线程。

因为多线程的操作是在同一个进程内实现的,所以线程间数据是共享数据资源的。

接下来介绍一下如何在进程间进行对象的交换,其实进程间进行对象的交换是一个子命题,更高层级的概括是在进程间进行通信,在官方的文档中对其进行了细分,所以这里也对其进行分类别的介绍。

在进程间进行对象交换的方式有两种,一种是队列,一种是管道。

1. 队列

1) 队列的代码示例

这里的模块的引入是 multiprocess.Queue,这个类近似于是 queue.Queue 的克隆,以下是官方文档的一个示例,内容是在父进程中创建一个队列,然后在子进程中写入数据,然后再在父进程中读取:


from multiprocessing import Process, Queue

def f(q):
    q.put([42, None, 'hello'])

if __name__ == '__main__':
    q = Queue()
    p = Process(target=f, args=(q,))
    p.start()
    print(q.get())    # prints "[42, None, 'hello']"
    p.join()

队列的写入使用 put(),读取使用 get()。

get() 还可以加上两个参数,block 和 timeout,block 表示是否阻塞,timeout 表示获取的超时时间。

接下来我们再实现一个功能,两个子进程写入数据,一个子进程读取数据,代码示例如下:

from multiprocessing import Queue, Process


def f_write(q, n, name):
    for i in range(n):
        q.put(f"{name}_{i}")
        time.sleep(0.1)


def f_read(q):
    while q.qsize() > 0:
        print(q.get(block=False, timeout=1))
        time.sleep(0.5)


def test_queue():
    # 三个进程,一个写进程,两个读进程
    q = Queue()
    q.put("origin_value")
    q.put("b")

    # p1 = Process(target=f_queue, args=(q, "c"))
    # p2 = Process(target=f_queue, args=(q, ))
    p1 = Process(target=f_write, args=(q, 5, "a"))
    p2 = Process(target=f_write, args=(q, 8, "b"))
    p3 = Process(target=f_read, args=(q,))

    p1.start()
    p2.start()
    p3.start()

    p1.join()
    p2.join()
    p3.join()

    print("total: ", q.qsize())

if __name__ == "__main__":
    test_queue()

2) 队列的相关方法

关于队列的相关函数,除了前面介绍的几种,还有比如判断队列的长度,是否为空等。

a) Queue()

在定义一个队列的时候,我们前面是直接定义 q=Queue(),不为其设置元素长度,而如果我们想要为其设置一个最大的长度,可以加上 maxsize 参数:

q = Queue(maxsize=3)

那么队列里最多只能有三个元素,而如果队列满了还往其中 put() 加入操作,则会阻塞,直到其他进程对其读取其中的数据。

b) put()

put() 函数表示的是往队列里添加元素,元素的类型不限,添加数字,字符串,字典,列表都可以:

q = Queue()
q.put(1)
q.put({"a": 4})
q.put([1,3,4])

前面介绍了,如果队列满了,还往队列里进行 put() 操作,则会进入阻塞操作,可以通过添加 block 或者 timeout 来进行避免。

block 表示是否阻塞,为 True 的话则会进入阻塞等待状态。False 的话则会引发异常。

timeout 表示超时,尝试往队列里添加数据,超出等待时间同样已发队列已满的异常。

c) get()

get() 函数表示从队列中读取元素,队列的写入和读取的原则是先入先出,最先进去的最先出来。

而为了避免队列为空的情况下进行 get() 进入阻塞状态,get() 可以使用两个参数,一个是 block,表示是否阻塞,一个是 timeout,表示超时时间。

如果队列为空还进行 get() 操作,使用上面这两个操作则会 raise 一个 Empty 的 error。

d) qsize()

返回队列的长度,但由于多进程或多线程的上下文,这个数字是不可靠的。

e) empty()

如果队列是空的,则返回 True,否则返回 False,由于多进程或多线程的环境,该状态是不可靠的。

f) full()

如果队列设置了 maxsize 参数,那么如果队列满了,则返回 True,否则返回 False,由于多进程或多线程的环境,该状态是不可靠的。

g) close()

关闭队列,如果执行了 q.close(),再往里面添加元素执行 q.put() 操作,则会引发报错。

2. 管道

1) 管道的相关函数

管道的引入方式如下:

from multiprocessing import Pipe

管道的定义可以直接实例化 Pipe,返回管道的两端:

conn1, conn2 = Pipe()

默认情况下,Pipe() 的参数 duplex 值为 True,表示管道是双工的,也就是可以双向通信的,比如 conn1 可以写入,也可以读出,conn2 可以写入也可以读出数据。

而如果手动设置 duplex 为 False,那么管道则是单向的,conn1 只能用于接收消息,conn2 只能发送消息。

管道用于发送和接收的函数分别如下:

发送信息

conn.send(obj)

发送的对象可以是字符串,也可以是其他对象,比如列表,字典等。

接收信息

conn.recv()

关闭连接对象

我们可以使用 close() 来关闭连接对象,当连接对象被垃圾回收时会自动调用:

conn.close()

判断连接对象中是否有可以读取的数据

如果我们直接使用 conn.recv() 的时候,如果管道内没有可接收的对象,会进入阻塞状态,直到管道内传入数据。

我们可以使用 poll() 函数判断管道内是否有可以读取的数据,返回的是一个布尔型数据,表示是否有数据:

has_data = conn.poll()

但是如果不设置超时时间,同样会进入等待状态,所以可以设置一个最大阻塞秒数:

has_data = conn.pool(timeout=3)  # 等待 3 秒

2) 管道的代码示例

接下来我们用下面的代码来进行管道的双工测试,即从管道的两端分别写入和读取数据。

from multiprocessing import Process, Pipe


def send_info(conn, info):
    conn.send(info)
    conn.close()


def read_info(conn):
    while conn.poll(timeout=2):
        info = conn.recv()
        print(info)


def test_pipe():
    # 两个 conn 分别都往里面读和写
    parent_conn, child_conn = Pipe()

    # p1 向 child 管道写入
    print("id out of func: ", id(child_conn))
    p1 = Process(target=send_info, args=(child_conn, "send_info_from_child"))
    p1.start()
    p1.join()

    # p2 从 parent 管道读取
    p2 = Process(target=read_info, args=(parent_conn,))
    p2.start()
    p2.join()

    # p3 向 parent 管道写入
    p3 = Process(target=send_info, args=(parent_conn, "send_info_from_parent"))
    p3.start()
    p3.join()

    # p4 从 child 管道读取
    p4 = Process(target=read_info, args=(child_conn,))
    p4.start()
    p4.join()


if __name__ == "__main__":
    test_pipe()

注意 :如果两个进程(或线程)同时尝试读取或写入管道的 同一 端,则管道中的数据可能会损坏。当然,在不同进程中同时使用管道的不同端的情况下不存在损坏的风险。

4、进程间同步

与多线程一样,多进程也可以使用锁来确保一次只有一个进程来执行一个操作,比如有一个打印到标准输出的操作,我们需要确保其打印的日志不紊乱,就可以使用下面的操作:

from multiprocessing import Process, Lock

def f(l, i):
    l.acquire()
    try:
        print("hello ", i)
    finally:
        l.release()

if __name__ == "__main__":
    lock = Lock()
    for num in range(10):
        Process(target=f, args=(lock, num)).start()

而如果不使用锁,我们重写 f 函数如下:

def f(l, i):
    print("hello ", i)

多执行几次,我们可以看到控制台的输出会出现错乱的情况,这样就可能对输出信息不能直观查看,比如:

hello  2
hello  0
hello  4
hello hello  3
 1
hello  5
hello  6
hello  8
hello  9
hello  7

5、进程间共享状态

在并发编程的时候,应当尽量避免使用共享状态,尤其是多进程操作时,但如果真的有这个需求,需要共享一些数据,multiprocessing 提供了两种方法,一种是共享内存,一种是服务进程。

1. 共享内存

我们可以使用 Value 或者 Array 将数据存储在共享内存映射中。

Value 是存储的单个变量,Array 存储的是数组,注意下,这里的 Value 和 Array 在定义的时候都需要指定元素类型。

其引入及代码示例如下:

from multiprocessing import Process, Value, Array


def f(n, a):
    n.value = 5
    a[0] = 100


if __name__ == "__main__":
    num = Value('d', 1)
    arr = Array('i', range(5))
    print(num.value)
    print(arr[:])

    p = Process(target=f, args=(num, arr))
    p.start()
    p.join()

    print(num.value)
    print(arr[:])

其中,引入的方式可以直接从 multiprocessing 中引入,在定义 Value 和 Array 的时候,第一个参数是 ‘d’ 和 ‘i’,分别表示类型是双精度浮点数和有符号整数。

这些共享对象将是进程和线程安全的。

更多的关于共享内存的信息,可以使用 multiprocessing.sharedctypes 模块。

2. 服务进程

我们可以使用 Manager() 返回的管理对象控制一个服务进程,这个进程还可以保存 Python 对象并允许其他进程使用代理操作它们。

这个操作的意思就是使用 Manager() 会跟多进程的操作方式一样,创建一个子进程,然后将一些需要共享的数据都放到这个子进程里,其他子进程可以操作这个子进程的数据来达到数据共享的目的。

Manager() 支持的数据类型有:list,dict,Namespace,Lock,Value,Array 等,下面介绍一下代码示例:

from multiprocessing import Process, Manager


def f(d, l):
    d["a"] = 1
    d["b"] = 2
    l[0] = 100


if __name__ == "__main__":
    with Manager() as manager:
        d = manager.dict()
        l = manager.list(range(5))

        p = Process(target=f, args=(d, l))
        p.start()
        p.join()

        print(d)
        print(l)

使用服务进程的管理器比使用共享内存对象更灵活,因为它们可以支持任意对象类型。

此外,单个管理器可以通过网络由不同计算机上的进程共享。但是,它们比使用共享内存慢。

如果想获取更多后端相关文章,可扫码关注阅读: