21-Python-多进程

1、为什么需要多进程

由于GIL的存在，python中的多线程其实并不是真正的多线程，如果想要充分地使用多核CPU的资源，在python中大部分情况需要使用多进程。

multiprocessing包是Python中的多进程管理包。与threading.Thread类似，它可以利用multiprocessing.Process对象来创建一个进程。该进程可以运行在Python程序内部编写的函数。该Process对象与Thread对象的用法相同，也有start(), run(), join()的方法。此外multiprocessing包中也有Lock/Event/Semaphore/Condition类 (这些对象可以像多线程那样，通过参数传递给各个进程)，用以同步进程，其用法与threading包中的同名类一致。所以，multiprocessing的很大一部份与threading使用同一套API，只不过换到了多进程的情境。

2、多进程的创建方法

import multiprocessing
import time


def run(name):
    time.sleep(2)
    print("hello", name)


# 多进程
if __name__ == "__main__":
    for i in range(10):
        p = multiprocessing.Process(target=run, args=("druid %s" % i, ))
        p.start()
        p.join()

3、进程间通信

不同进程间内存是不共享的，要想实现两个进程间的数据交换，可以用以下方法：

3.1、Queues

from multiprocessing import Queue, Process
import os


def f(pq):
    pq.put([42, None, "druid"])
    print("子进程的父进程ID:", os.getppid())
    print("子进程的当前进程ID:", os.getpid())


if __name__ == "__main__":
    print("父进程ID:", os.getppid())
    print("当前进程ID:", os.getpid())
    q = Queue()
    p = Process(target=f, args=(q,))
    p.start()
    print(q.get())

    p.join()

3.2、Pipes

The Pipe() function returns a pair of connection objects connected by a pipe which by default is duplex (two-way). For example:

from multiprocessing import Process, Pipe


def f(conn):
    conn.send("33[31;1m msg1 from child process33[0m")
    conn.send("33[31;1m msg2 from child process33[0m")
    print("from parent:", conn.recv())
    conn.close()


if __name__ == "__main__":
    parent_conn, child_conn = Pipe()
    p = Process(target=f, args=(child_conn,))
    p.start()

    print("from child process:", parent_conn.recv())
    print("from child process:", parent_conn.recv())
    # print("from child process:", parent_conn.recv())  # 如果接收次数多余发送次数，就会卡住
    parent_conn.send("33[32;1m msg from parent process33[0m")

    p.join()

3.3、Managers

Queues和Pipes只是实现了进程间数据的传递，Managers真正实现了进程间的数据共享，即多个进程可以修改同一份数据。

A manager object returned by Manager() controls a server process which holds Python objects and allows other processes to manipulate them using proxies.

A manager returned by Manager() will support types list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value and Array. For example:

from multiprocessing import Process, Manager
import os


def f(d, l):
    d[os.getpid()] = os.getpid()
    l.append(os.getpid())
    print(l)


if __name__ == "__main__":
    with Manager() as manager:
        d = manager.dict()
        l = manager.list(range(5))
        p_list = []  # 用来存放进程
        for i in range(10):
            p = Process(target=f, args=(d, l))
            p.start()
            p_list.append(p)
        for res in p_list:
            res.join()

        print(d)
        print(l)

4、进程池

进程池内部维护一个进程序列，当使用时，则去进程池中获取一个进程，如果进程池序列中没有可供使用的进进程，那么程序就会等待，直到进程池中有可用进程为止。

进程池中有两个方法：

• • apply
  • apply_async

4.1、apply方法

from multiprocessing import Pool
import os
import time


def func(i):
    time.sleep(2)
    print("子进程ID:", os.getpid())  # 每隔两秒打印一次结果
    return i+100


if __name__ == "__main__":
    pool = Pool(processes=5)  # 允许进程池同时放入5个进程
    print("主进程ID:", os.getpid())
    for i in range(10):
        pool.apply(func=func, args=(i,))  # 串行

    print("end")
    pool.close()
    pool.join()  # 进程池中进程执行完毕后再关闭，如果不加，那么程序直接关闭

4.2、apply_async方法

from multiprocessing import Pool
import os
import time


def func(i):
    time.sleep(2)
    print("子进程ID:", os.getpid())
    return i+100


if __name__ == "__main__":
    pool = Pool(processes=5)  # 允许进程池同时放入5个进程
    print("主进程ID:", os.getpid())
    for i in range(10):
        pool.apply_async(func=func, args=(i,))  # 并行

    print("end")  # 先打印主进程ID和"end"，然后每次打印5个子进程ID
    pool.close()
    pool.join()  # 进程池中进程执行完毕后再关闭，如果不加，那么程序直接关闭

4.3、apply_async+回调函数

from multiprocessing import Pool
import os
import time


def func(i):
    time.sleep(2)
    print("子进程ID:", os.getpid())
    return i+100


def bar(arg):
    print("--->exec done: %s, SubPID:%s" % (arg, os.getpid()))


if __name__ == "__main__":
    pool = Pool(processes=5)  # 允许进程池同时放入5个进程
    print("主进程ID:", os.getpid())
    for i in range(10):
        pool.apply_async(func=func, args=(i,), callback=bar)  # callback为回调函数，前面的函数执行完了才会执行回调函数

    print("end")
    pool.close()
    pool.join()  # 进程池中进程执行完毕后再关闭，如果不加，那么程序直接关闭