43. 使用多进程

秒速五厘米 2023-07-04 04:54 14阅读 0赞

由于Python中全局解释器锁（GIL）的存在，在任意时刻只允许一个线程在解释器中运行，因此Python的多线程不适合处理CPU密集型的任务。

要求：想要处理CPU密集型的任务，可以使用多进程模型。

解决方案：

使用标准库中`multiprocessing.Process`类，它可以确定子进程执行任务。操作接口、进程间通信、进程加同步等都与`threading.Thread`类类似。

--------------------

*  对于`multiprocessing.Process`类：

在multiprocessing中，通过创建一个Process对象然后调用它的`start()`方法来生成进程。 Process和`threading.Thread`的API 相同，`start()`、`join()`和`run()`方法作用一致。

进程和线程的根本区别：进程是操作系统资源分配的基本单位，而线程是任务调度和执行的基本单位。

进程是线程的容器，不存在没有线程的进程。如果一个进程内有多个线程，则执行过程不是一条线的，而是多条线（线程）共同完成的；线程是进程的一部分，所以线程也被称为轻权进程或者轻量级进程。

进程有自己独立的地址空间，每启动一个进程，系统都会为其分配地址空间，建立数据表来维护代码段、堆栈段和数据段，线程没有独立的地址空间，它使用相同的地址空间共享数据。

*  对于进程间通信：

队列：

线程间通信使用的是标准库中的`queue.Queue`类，它是一个线程安全的队列。而进程间通信使用的是`multiprocessing.Queue`类，它是一个近似`queue.Queue`的克隆，也是线程和进程安全的。

管道：

`Pipe()`函数返回一个由管道连接的连接对象，默认情况下是双工（双向）。返回的两个连接对象`Pipe()`表示管道的两端。每个连接对象都有`send()`和`recv()`方法（相互之间的）。请注意，如果两个进程（或线程）同时尝试读取或写入管道的同一端，则管道中的数据可能会损坏。当然，同时使用管道的不同端的进程不存在损坏的风险。

>>> from multiprocessing import Process, Pipe
    
    >>> c1, c2 = Pipe()
    
    >>> def f(c):
    ...     print('in child')
    ...     data = c.recv()
    ...     print(data)
    ...     c.send(data * 2)
    ... 
    
    >>> Process(target=f, args=(c2,)).start()
    in child
    
    >>> c1.send(100)
    100
    
    >>> c1.recv()
    200

`send()`方法表示输入数据到管道；`recv()`方法表示从管道中接收数据。

--------------------

*  方案示例：

import time
    from threading import Thread
    from multiprocessing import Process
    from queue import Queue as Thread_Queue
    from multiprocessing import Queue as Process_Queue
    
    def is_armstrong(n):                #判断是否是水仙花数
        a, t = [], n
        while t :
            a.append(t % 10)
            t //= 10
        k = len(a)
        return sum(x**k for x in a) == n
    
    def find_armstrong(a, b, q=None):               #在(a, b)内找出水仙花数
        res = [x for x in range(a, b) if is_armstrong(x)]
        if q:
            q.put(res)
        return res
    
    def find_by_thread(*ranges):                #通过线程寻找
        q = Thread_Queue()
        workers = []
        for r in ranges:
            a, b = r
            t = Thread(target=find_armstrong, args=(a, b, q))
            t.start()
            workers.append(t)
        
        res = []
        for _ in range(len(ranges)):
            res.append(q.get())
        return res
    
    def find_by_process(*ranges):               #通过进程寻找
        q = Process_Queue()
        workers = []
        for r in ranges:
            a, b = r
            t = Process(target=find_armstrong, args=(a, b, q))
            t.start()
            workers.append(t)
        
        res = []
        for _ in range(len(ranges)):
            res.extend(q.get())
        return res
    
    if __name__ == '__main__':
        t0 = time.time()
        res = find_by_thread([10000000, 15000000], [15000000, 20000000], [20000000, 25000000], [25000000, 30000000])
        # res = find_by_process([10000000, 15000000], [15000000, 20000000], [20000000, 25000000], [25000000, 30000000])
        print(res)
        print(time.time() - t0)

[[24678050, 24678051], [], [], []]              #find_by_thread 结果
    98.16692352294922
    
    [24678050, 24678051]                #find_by_process 结果
    55.84143948554993

从运行结果来看，对于CPU密集型的任务，通过多进程来处理比多线程更好。

--------------------