Python-多进程(进程,锁,通讯,进程池)

素颜马尾好姑娘i 2023-10-02 22:11 9阅读 0赞

#### 文章目录 ####

*  Python-多进程(进程,锁,通讯,进程池)
 *   *  创建多进程
     *  守护进程
     *  进程间的通信
     *   *  进程队列(Queue)通信
         *  进程管道(Pipes)通信
         *  进程的Mangers通信
         *   Value，Array资源共享
     *  进程锁
     *   *   Lock
         *  RLock
         *  Condition
         *  Semaphore
         *  Event
         *   Barrier
     *  进程池

## Python-多进程(进程,锁,通讯,进程池) ##

进程：通常一个运行着的应用程序就是一个进程，比如：我启动了一个音乐播放器，现在它就是一个进程。  
线程：线程是进程的最小执行单元，比如：我在刚启动的音乐播放器上选了一首歌曲进行播放，这就是一个线程。

在多线程一文中，我们说了因为 GIL 的原因，CPython 解释器下的多线程牺牲了并行性，为此 Python 提供了多进程模块 multiprocessing，该模块同时提供了本地和远程并发，使用子进程代替线程，可以有效的避免 GIL 带来的影响，能够充分发挥机器上的多核优势，可以实现真正的并行效果，并且它与 threading 模块的 API 基本类似，使用起来也比较方便。,所以建议先学多线程,然后在学习多进程会比较好理解

### 创建多进程 ###

Python自带的多进程库 multiprocessing 可实现多进程。  
语法: `Process(group = None,target =None,name=None, args=[ ], kwargs={ })`

参数说明：

1.  group参数未使用，值始终为None
2.  target表示调用对象，即子进程要执行的任务
3.  name为子进程的名称
4.  args表示调用对象的位置参数元组，args=(2,‘anne’)
5.  kwargs表示调用对象的字典,kwargs=\{‘name’:‘anne’,'a:18\}

> process属性&方法：

![在这里插入图片描述][e4ffaa66067b4a77a1cba762b19360e5.png]

下面我们就来创建一个进程:

import time
    import random
    from multiprocessing import Process
    
    
    def test_fun(name):
        # 随机等待1~5秒
        time.sleep(random.randrange(1, 5))
        print(f"我是{
          name}子进程！")
    
    
    # 进程一定要写在“if __name__ == '__main__':”下面
    if __name__ == '__main__':
        process_list = []  # 存放开启的进程
        for i in range(3):
            # 进程中的参数args表示调用对象的位置参数元组   注意：元组中只有一个元素时结尾要加","逗号
            p = Process(target=test_fun, args=(f"son{
          i + 1}",))
            p.start()# 启动进程
    
            process_list.append(p)
        for i in process_list:
            i.join()  # 阻塞每个子进程，主进程会等待所有子进程结束再结束主进程
        print("主进程结束！")

> 通过类继承的方法创建进程

from multiprocessing import Process  # 多进程的类
    import time
    import random
    
    
    # 创建一个进程类Pros然后继承Process类
    class Pros(Process):
        def __init__(self, name):
            super().__init__()
            self.name = name
    
        def run(self):
            print('%s 开始运行！' % self.name)
            # 随机睡眠1~5秒
            time.sleep(random.randrange(1, 5))
            print('%s 结束！' % self.name)
    
    
    if __name__ == "__main__":
        pro_list = []
        # 创建3个子进程
        for i in range(3):
            num = "p" + str(i + 1)
            num = Pros(f"子进程{
          num}")
            num.start()  # start方法会自动调用进程类中的run方法
            pro_list.append(num)
    
        # 子进程阻塞，主进程会等待所有子进程结束再结束
        for i in pro_list:
            i.join()
    
        print('主进程结束')

### 守护进程 ###

import multiprocessing
    import time
     
     
    def eat():
        for i in range(10):
            print("我吃我吃……")
            time.sleep(0.5)
     
     
    if __name__ == '__main__':
        eat_process = multiprocessing.Process(target=eat)
        # 设置进程守护(在启动进程前设置)
        eat_process.daemon = True
        eat_process.start()
     
        time.sleep(1)
        print("我吃饱了……")

### 进程间的通信 ###

为了进程安全起见，两个进程之间的数据是不能够互相访问的(默认情况下),进程与进程之间的数据是不可以互相访问的，而且每一个进程的内存是独立的。多进程的资源是独立的，不可以互相访问，如果想多个进程之间实现数据交互就必须通过中间件实现。

进程间通信方法有Queue、Pipes、Mangers和Value，Array四种：

#### 进程队列(Queue)通信 ####

Queue（\[maxsize\]）:建立一个共享的队列(其实并不是共享的，实际是克隆的，内部维护着数据的共享)，多个进程可以向队列里存/取数据。其中，参数是队列最大项数，省略则无限制。Queue的常用方法如下：

*  qsize()：返回当前队列包含的消息数量。
 *  empty()：如果队列为空，返回True；返之返回False。
 *  full()：如果队列满了，返回True；反之返回False。
 *  get(block\[,timeout\])：获取队列中的一条信息，然后将其从队列中移除，block默认值为True。如果block使用默认值，且没有设置timeout(单位秒)，消息队列为空，此时程序将被阻塞（停在读取状态），直到从消息队列读到消息为止。如果设置了timeout，则会等待timeout秒，若还没有读取任何消息，则抛出“Queue.Empty”异常。 如果block值为False，消息队列为空，则会立刻抛出“Queue.Empty”异常。
 *  Queue.get\_nowait()：相当于Queue.get(False)。
 *  Queue.put(item,\[block\[,timeout\]\])：将item消息写入队列，block默认值为True。如果block使用默认值，且没有设置timeout(单位秒)，消息队列如果已经没有空间可以写入，此时程序将被阻塞(停在写入状态)，直到从消息队列腾出空间为止，如果设置了timeout，则会等待timeout秒，若还没有空间，则抛出“Queue.Full”异常。如果block值为False，消息队列没有空间可写入，则会立刻抛出“Queue.Full”异常
 *  Queue.put\_nowait(item)：相当Queue.put(item,False)。

from multiprocessing import Queue, Process
    
    
    def fun(q, i):
        q.put([1 * i, 2 * i, 3 * i])
    
    
    if __name__ == '__main__':
        Q = Queue(5)  # 设置进程队列长度
        for i in range(2):  # 启动两个进程，想队列里put数据
            process = Process(target=fun, args=(Q, i + 1))  # 创建一个进程，将Q传入，实际上是克隆了Q
            process.start()
            process.join()
        print(Q.qsize())  # 2
        print(Q.get())    # [1, 2, 3]
        print(Q.get())    # [2, 4, 6]
        print(Q.qsize())  # 0

#### 进程管道(Pipes)通信 ####

多进程还有一种数据传递方式叫管道原理和 Queue相同。Pipe可以在进程之间创建一条管道，并返回元组（conn1,conn2）,其中conn1，conn2表示管道两端的连接对象，强调一点：必须在产生Process对象之前产生管道。

构造方法：Pipe(\[dumplex\]) dumplex:默认管道是全双工的，如果将duplex射成False，conn1只能用于接收，conn2只能用于发送。

实例方法：

*  send(obj)：通过连接发送对象。obj是与序列化兼容的任意对象
 *  recv()：接收conn2.send(obj)发送的对象。如果没有消息可接收，recv方法会一直阻塞。如果连接的另外一端已经关闭，那么recv方法会抛出EOFError。
 *  close():关闭连接。如果conn1被垃圾回收，将自动调用此方法
 *  fileno():返回连接使用的整数文件描述符
 *  poll(\[timeout\]):如果连接上的数据可用，返回True。timeout指定等待的最长时限。如果省略此参数，方法将立即返回结果。如果将timeout射成None，操作将无限期地等待数据到达。
 *  recv\_bytes(\[maxlength\]):接收c.send\_bytes()方法发送的一条完整的字节消息。maxlength指定要接收的最大字节数。如果进入的消息，超过了这个最大值，将引发IOError异常，并且在连接上无法进行进一步读取。如果连接的另外一端已经关闭，再也不存在任何数据，将引发EOFError异常。
 *  send\_bytes(buffer \[, offset \[, size\]\])：通过连接发送字节数据缓冲区，buffer是支持缓冲区接口的任意对象，offset是缓冲区中的字节偏移量，而size是要发送字节数。结果数据以单条消息的形式发出，然后调用c.recv\_bytes()函数进行接收
 *  recv\_bytes\_into(buffer \[, offset\]):接收一条完整的字节消息，并把它保存在buffer对象中，该对象支持可写入的缓冲区接口（即bytearray对象或类似的对象）。offset指定缓冲区中放置消息处的字节位移。返回值是收到的字节数。如果消息长度大于可用的缓冲区空间，将引发BufferTooShort异常。

from multiprocessing import Process, Pipe
    
    import time
    
    def f(child_conn):
        time.sleep(1)
        child_conn.send("吃了吗") # 给父进程发送消息
        print("来自父亲的问候:", child_conn.recv()) #接收父进程发送的消息
        child_conn.close()
    
    if __name__ == "__main__":
        parent_conn, child_conn = Pipe()  # 创建管道两端,必须在创建Process之前创建Pipe
        p = Process(target=f, args=(child_conn,))  # 创建子进程
        p.start()
        print("来自儿子的问候:", parent_conn.recv())# 接收子进程的消息
        parent_conn.send("嗯") # 给子进程发送消息

#### 进程的Mangers通信 ####

Manager()返回的manager对象控制了一个server进程，此进程包含的python对象可以被其他的进程通过proxies来访问。从而达到多进程间数据通信且安全。

Manager实现了多个进程间的数据共享，支持的数据类型有 list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value , Array。

Manager中 **以下类型均不是进程安全的**, Manager只是保证的数据能传递但是没有保证数据的安全所以很多时候需要我们自己通过进程锁来控制的

*  Array(self,\*args,\*\*kwds)
 *  BoundedSemaphore(self,\*args,\*\*kwds)
 *  Condition(self,\*args,\*\*kwds)
 *  Event(self,\*args,\*\*kwds)
 *  JoinableQueue(self,\*args,\*\*kwds)
 *  Lock(self,\*args,\*\*kwds)
 *  Namespace(self,\*args,\*\*kwds)
 *  Pool(self,\*args,\*\*kwds)
 *  Queue(self,\*args,\*\*kwds)
 *  RLock(self,\*args,\*\*kwds)
 *  Semaphore(self,\*args,\*\*kwds)
 *  Value(self,\*args,\*\*kwds)
 *  dict(self,\*args,\*\*kwds)
 *  list(self,\*args,\*\*kwds)

上面不安全的类型部分在进程里是有安全的可以通过`from multiprocessing import Array,Lock,Queue`\` 方式使用,没有的那么只能加锁了

import multiprocessing
    
    def f(x, arr, l, d, n):
        x.value = 3.14
        arr[0] = 5
        l.append('Hello')
        d["name"] = "xxx"
        n.a = 10
    if __name__ == '__main__':
        manager = multiprocessing.Manager()
        x = manager.Value('d', 0.0)
        arr = manager.Array('i', range(10))
        l = manager.list()
        d = manager.dict()
        n = manager.Namespace() 
        # 以上代码需要在进程创建前使用
        proc = multiprocessing.Process(target=f, args=(x, arr, l, d, n))
        proc.start()
        proc.join()
        print(x.value)
        print(arr)
        print(l)
        print(d)
        print(n.a)

以上都是Mangers提供的可传递的类型那么如果我们想自己创建一个对象也能进程之间传递那么需要怎么做?

我有一个IpConnectionPool对象需要多个进程共享,步骤如下:

1.  创建BaseManager
2.  注册IpConnectionPool对象
3.  开启一个IpConnectionPool对象进程
4.  获取进程共享IpConnectionPool对象
5.  IpConnectionPool对象传进需要使用的进程里

导包

from multiprocessing.managers import BaseManager

参考代码

# 多进程对象共享
        manager = BaseManager()
        # 一定要在start前注册，不然就注册无效
        # 注册数据库线程池对象
        manager.register('IpConnectionPool', IpConnectionPool)
        # 启动manager服务(就是开了一个新进程来管理IpConnectionPool对象的)
        manager.start()
        # 获取进程对象
        ipConnectionPool = manager.IpConnectionPool() 
        # manager必须在主线程进行创建进程否则报错
        pool = multiprocessing.Pool(int(multiprocessing.cpu_count()))  # 创建进程池
        for i in range(2,3423):
        	# 数据库线程池对象传入到进程中使用
            pool.apply_async(batchInsert,args=(i,ipConnectionPool),callback=call_back,error_callback=err_call_back)
            time.sleep(5)
        pool.close()
        pool.join()

####  Value，Array资源共享 ####

multiprocessing 中Value和Array的实现原理都是在共享内存中创建ctypes()对象来达到共享数据的目的，两者实现方法大同小异，只是选用不同的ctypes数据类型而已。

**Value**

构造方法：Value((typecode\_or\_type, args\[, lock\])

*  typecode\_or\_type：定义ctypes()对象的类型，可以传Type code或 C Type，具体对照表见下文。
 *  args：传递给typecode\_or\_type构造函数的参数
 *  lock：默认为True，创建一个互斥锁来限制对Value对象的访问，如果传入一个锁，如Lock或RLock的实例，将用于同步。如果传入False，Value的实例就不会被锁保护，它将不是进程安全的。

typecode\_or\_type支持的类型：

<table> 
 <thead> 
  <tr> 
   <th>Type code</th> 
   <th>C Type</th> 
   <th>Python Type</th> 
   <th>Minimum size in bytes</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><code>'b'</code></td> 
   <td>signed char</td> 
   <td>int</td> 
   <td>1</td> 
  </tr> 
  <tr> 
   <td><code>'B'</code></td> 
   <td>unsigned char</td> 
   <td>int</td> 
   <td>1</td> 
  </tr> 
  <tr> 
   <td><code>'u'</code></td> 
   <td>Py_UNICODE</td> 
   <td>Unicode character</td> 
   <td>2</td> 
  </tr> 
  <tr> 
   <td><code>'h'</code></td> 
   <td>signed short</td> 
   <td>int</td> 
   <td>2</td> 
  </tr> 
  <tr> 
   <td><code>'H'</code></td> 
   <td>unsigned short</td> 
   <td>int</td> 
   <td>2</td> 
  </tr> 
  <tr> 
   <td><code>'i'</code></td> 
   <td>signed int</td> 
   <td>int</td> 
   <td>2</td> 
  </tr> 
  <tr> 
   <td><code>'I'</code></td> 
   <td>unsigned int</td> 
   <td>int</td> 
   <td>2</td> 
  </tr> 
  <tr> 
   <td><code>'l'</code></td> 
   <td>signed long</td> 
   <td>int</td> 
   <td>4</td> 
  </tr> 
  <tr> 
   <td><code>'L'</code></td> 
   <td>unsigned long</td> 
   <td>int</td> 
   <td>4</td> 
  </tr> 
  <tr> 
   <td><code>'q'</code></td> 
   <td>signed long long</td> 
   <td>int</td> 
   <td>8</td> 
  </tr> 
  <tr> 
   <td><code>'Q'</code></td> 
   <td>unsigned long long</td> 
   <td>int</td> 
   <td>8</td> 
  </tr> 
  <tr> 
   <td><code>'f'</code></td> 
   <td>float</td> 
   <td>float</td> 
   <td>4</td> 
  </tr> 
  <tr> 
   <td><code>'d'</code></td> 
   <td>double</td> 
   <td>float</td> 
   <td>8</td> 
  </tr> 
 </tbody> 
</table>

**Array**

构造方法：Array(typecode\_or\_type, size\_or\_initializer, \*\*kwds\[, lock\])

*  typecode\_or\_type：同上
 *  size\_or\_initializer：如果它是一个整数，那么它确定数组的长度，并且数组将被初始化为零。否则，size\_or\_initializer是用于初始化数组的序列，其长度决定数组的长度。
 *  kwds：传递给typecode\_or\_type构造函数的参数
 *  lock：同上

import multiprocessing
    
    def f(n, a):
    
        n.value = 3.14
    
        a[0] = 5
    
    if __name__ == '__main__':
    
        num = multiprocessing.Value('d', 0.0)
    
        arr = multiprocessing.Array('i', range(10))
    
        p = multiprocessing.Process(target=f, args=(num, arr))
    
        p.start()
    
        p.join()
    
        print(num.value)
    
        print(arr[:])

### 进程锁 ###

多进程之间不共享数据，但共享同一套文件系统，像访问同一个文件、同一终端打印，如果不进行同步操作，就会出现错乱的现象。

所有在 threading 存在的同步方式，multiprocessing 中都有类似的等价物，如：锁(lock,RLock)、信号量Semaphore,Event（事件）Condition,Barrier等。不懂的可以看多线程文章里写的介绍

####  Lock ####

from multiprocessing import Process, Lock
    import time
     
     
    def fun(l, i):
        l.acquire()      # 获取锁
        print("正在运行进程： ", i)
        time.sleep(2)
        l.release()      # 使用完后释放锁
     
     
    if __name__ == '__main__':
        lock = Lock()                                # 生成锁的实例
        for i in range(5):
            p = Process(target=fun, args=(lock, i))  # 创建进程
            p.start()                                # 启动，这里没有join，进程可以并发

#### RLock ####

from multiprocessing import Process,RLock
    
    if __name__ == '__main__':
        # 创建一个rlock对象
        lock = RLock()
        # 初始化共享资源
        abce = 0
        # 本线程访问共享资源
        lock.acquire()  # 加锁
        abce = abce + 1
        # 这个线程尝试访问共享资源
        lock.acquire()  # 再次加锁
        abce = abce + 2
        lock.release()  # 释放里面的锁
        lock.release()  # 释放外面的锁
        print(abce)

#### Condition ####

from multiprocessing import Process, Condition, Manager
    import time
    
    
    class Producer(Process):
        def __init__(self, name, condtion, sheep):
            super().__init__(name=name)
            self.sheep = sheep
            self.condtion = condtion
    
        def run(self):
            while True:
                time.sleep(1)
                self.condtion.acquire()
                if len(self.sheep) < 10:
                    print(self.name + "生产了1件产品")
                    self.sheep.append('1件产品')
                    self.condtion.notify_all()  # 通知所有消费者
                else:
                    print("仓库满了，停止生产!")
                    self.condtion.wait()
    
                self.condtion.release()
    
    
    class Customer(Process):
        def __init__(self, name, condtion, sheep):
            super().__init__(name=name)
            self.sheep = sheep
            self.condtion = condtion
            pass
    
        def run(self):
    
            while True:
                time.sleep(0.5)
                self.condtion.acquire()  # 加锁
                if len(self.sheep) > 0:
                    meat = self.sheep.pop()
                    print(self.name + "购买了" + meat + "还剩多少" + str(len(self.sheep)) + "件")
                    self.condtion.notify_all()
    
                else:
                    print("买光了，等待")
                    self.condtion.wait()  # 等待生产者唤醒
                self.condtion.release()  # 解锁
    
    
    if __name__ == '__main__':
        condtion = Condition()
        manager = Manager()
        sheep = manager.list(['a件产品', 'b件产品', 'c件产品', 'd件产品', 'e件产品']) #必须这样创建不然多进程共享不了数据
        print(sheep)
        p1 = Producer("1号生产车间", condtion, sheep)
        p2 = Producer("2号生产车间", condtion, sheep)
        p3 = Producer("3号生产车间", condtion, sheep)
        p4 = Producer("4号生产车间", condtion, sheep)
        p5 = Producer("5号生产车间", condtion, sheep)
        p6 = Producer("6号生产车间", condtion, sheep)
        p1.start()
        p2.start()
        p4.start()
        p5.start()
        p6.start()
    
        c1 = Customer('小王', condtion, sheep)
        c2 = Customer('小李', condtion, sheep)
        c3 = Customer('小贾', condtion, sheep)
        c4 = Customer('小沈', condtion, sheep)
        c5 = Customer('小刘', condtion, sheep)
        c1.start()
        c2.start()
        c3.start()
        c4.start()
        c5.start()
        c1.join()
        c2.join()
        c3.join()
        c4.join()
        c5.join()
        p1.join()
        p2.join()
        p4.join()
        p5.join()
        p6.join()

#### Semaphore ####

from multiprocessing import Process
    from multiprocessing import Semaphore
    import time
    import random
    
    
    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) # 同一时间只能3个人吃饭
        for i in range(10):
            p = Process(target=fun, args=(i, sem))
            p.start()

#### Event ####

import time
    from multiprocessing import Process,Event
    
    event =  Event() # 创建事件对象
    def student(name):
        print('学生%s 正在听课' % name)
        event.wait(2)
        print('学生%s 课间活动' % name)
    
    
    def teacher(name):
        print('老师%s 正在授课' % name)
        time.sleep(7)
        event.set()
    
    
    if __name__ == '__main__':
        stu1 = Process(target=student, args=('allen',))
        stu2 = Process(target=student, args=('wxx',))
        stu3 = Process(target=student, args=('yxx',))
        t1  = Process(target=teacher, args=('winnie',))
    
        t1.start()
        stu1.start()
        stu2.start()
        stu3.start()
    
        t1.join()
        stu1.join()
        stu2.join()
        stu3.join()

####  Barrier ####

from multiprocessing import Process,Barrier
    
    def open():
        print('人数够了， 开门!')
    
    class Customer(Process):
        def __init__(self, *args, **kwargs):
            super().__init__(*args, **kwargs)
            self.n = 3 # 设置开门3次
        def run(self):
            while self.n > 0:
                self.n -= 1
                print('{0}在等着开门.'.format(self.name))
                try:
                    barrier.wait()
                except Exception as e:
                    pass
                print('开门了， go go go  ,{0} 进去了'.format(self.name))
    
    
    if __name__ == '__main__':
        barrier = Barrier(3, open) # 必须在创建进程之前
        t1 = Customer(name='A')
        t2 = Customer(name='B')
        t3 = Customer(name='C')
    
        t1.start()
        t2.start()
        t3.start()

### 进程池 ###

当程序中设置到多进程编程时，Python 提供了更好的管理多个进程的方式，就是使用进程池。

在利用 Python 进行系统管理的时候，特别是同时操作多个文件目录，或者远程控制多台主机，并行操作可以节约大量的时间。

当被操作对象数目不大时，可以直接利用 multiprocessing 中的 Process 动态生成多个进程，十几个还好，但如果是上百个，上千个目标，手动的去限制进程数量却又太过繁琐，此时可以发挥进程池的功效。

Pool可以提供指定数量的进程供用户调用，当有新的请求提交到 pool 中时，如果进程池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到规定最大值，那么该请求就会等待，直到池中有进程结束，才会创建新的进程来它。

Python concurrent.futures 模块提供了 ProcessPoolExecutor专门用来创建一个进程池和线程池的使用方法一样构建语法:

`ProcessPoolExecutor(max_workers=None, mp_context=None,initializer=None, initargs=())`\`

max\_workers: 可用于执行给定调用的最大进程数。如果指定为None或不指定，则将创建与机器拥有处理器数量相同的工作进程。

mp\_context: 启动工作者的多处理上下文。该对象应该提供SimpleQueue、Queue和Process。

initializer: 用于初始化工作进程的可调用对象。

initargs: 传递给初始化器的参数元组。

Exectuor 提供了如下常用方法：

*  submit(fn, \*args, \*\*kwargs)：将 fn 函数提交给进程池。\*args 代表传给 fn 函数的参数，\*kwargs 代表以关键字参数的形式为 fn 函数传入参数。
 *  map(func, \*iterables, timeout=None, chunksize=1)：该函数类似于全局函数 map(func, \*iterables)，只是该函数将会启动多个进程，以异步方式立即对 iterables 执行 map 处理。
 *  shutdown(wait=True)：关闭进程池。

程序将 task 函数提交（submit）给进程池后，submit 方法会返回一个 Future 对象，Future 类主要用于获取线程任务函数的返回值。由于线程任务会在新线程中以异步方式执行，因此，线程执行的函数相当于一个“将来完成”的任务，所以 Python 使用 Future 来代表。

Future 提供了如下方法：

*  cancel()：取消该 Future 代表的进程任务。如果该任务正在执行，不可取消，则该方法返回 False；否则，程序会取消该任务，并返回 True。
 *  cancelled()：返回 Future 代表的进程任务是否被成功取消。
 *  running()：如果该 Future 代表的进程任务正在执行、不可被取消，该方法返回 True。
 *  done()：如果该 Funture 代表的进程任务被成功取消或执行完成，则该方法返回 True。
 *  result(timeout=None)：获取该 Future 代表的进程任务最后返回的结果。如果 Future 代表的进程任务还未完成，该方法将会阻塞当前进程，其中 timeout 参数指定最多阻塞多少秒。
 *  exception(timeout=None)：获取该 Future 代表的进程任务所引发的异常。如果该任务成功完成，没有异常，则该方法返回 None。
 *  add\_done\_callback(fn)：为该 Future 代表的进程任务注册一个“回调函数”，当该任务成功完成时，程序会自动触发该 fn 函数。

在用完一个进程池后，应该调用该进程池的 shutdown() 方法，该方法将启动进程池的关闭序列。调用 shutdown() 方法后的进程池不再接收新任务，但会将以前所有的已提交任务执行完成。当进程池中的所有任务都执行完成后，该进程池中的所有线程都会死亡。

使用进程池来执行线程任务的步骤如下：

1.  调用 ProcessPoolExecutor类的构造器创建一个进程池。
2.  定义一个普通函数作为进程任务。
3.  调用 ProcessPoolExecutor对象的 submit() 方法来提交进程任务。
4.  当不想提交任何任务时，调用 ProcessPoolExecutor对象的 shutdown() 方法来关闭进程池。

from concurrent.futures import ProcessPoolExecutor
    import time
    import multiprocessing
    
    # 定义一个准备作为进程任务的函数
    def action(max):
        my_sum = 0
        for i in range(max):
            print("=====" + str(i))
            my_sum += i
        return my_sum
    
    
    if __name__ == '__main__':
        # 创建一个包含2条线程的进程池
        pool = ProcessPoolExecutor(max_workers=2)
        # 向线程池提交一个task, 50会作为action()函数的参数
        future1 = pool.submit(action, 50)
        # 向线程池再提交一个task, 100会作为action()函数的参数
        future2 = pool.submit(action, 100)
        # 判断future1代表的任务是否结束
        print(future1.done())
        time.sleep(3)
        # 判断future2代表的任务是否结束,如果结束返回True否则返回False
        print(future2.done())
        # 查看future1代表的任务返回的结果
        print(future1.result())
        # 查看future2代表的任务返回的结果
        print(future2.result())
        # 关闭进程池
        pool.shutdown()

**注意:** 当程序使用 Future 的 result() 方法来获取结果时，该方法会阻塞当前进程，如果没有指定 timeout 参数，当前进程一直处于阻塞状态，直到 Future 代表的任务返回。

其他操作的就不在一一演示了在可以看线程池操作方式,这里演示一个在进程里捕获异常的方式(因为进程里的异常会被吃掉,在运行的过程很难看到问题)

from concurrent.futures import ProcessPoolExecutor
    
    from src.log.Log import Log
    
    def action(max):
        print(max)
        raise Exception("异常=======")
    
    # 进程池异常回调
    def thread_pool_callback(worker):
        worker_exception = worker.exception()
        if worker_exception: # 如果进程里出现了异常,那么获取到异常写入日志中
            Log.logError("Worker return exception: {}".format(worker_exception))
    
    if __name__ == '__main__':
        # 创建一个包含2条线程的线程池
        executor = ProcessPoolExecutor(max_workers=2)
        thread_pool_exc = executor.submit(action, "======")
        thread_pool_exc.add_done_callback(thread_pool_callback) # 可以试试把这个去掉,就发现异常没有了被吞了

![在这里插入图片描述][241d60d901ab468c8532f2a2d4413093.gif]

点赞 -收藏-关注-便于以后复习和收到最新内容有其他问题在评论区讨论-或者私信我-收到会在第一时间回复在本博客学习的技术不得以任何方式直接或者间接的从事违反中华人民共和国法律,内容仅供学习、交流与参考 免责声明：本文部分素材来源于网络，版权归原创者所有，如存在文章/图片/音视频等使用不当的情况，请随时私信联系我、以迅速采取适当措施，避免给双方造成不必要的经济损失。感谢,配合,希望我的努力对你有帮助^\_^

[e4ffaa66067b4a77a1cba762b19360e5.png]: https://img-blog.csdnimg.cn/e4ffaa66067b4a77a1cba762b19360e5.png
[241d60d901ab468c8532f2a2d4413093.gif]: https://img-blog.csdnimg.cn/241d60d901ab468c8532f2a2d4413093.gif