devbox
2023面试高手
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Python 多进程：multiprocessing、aiomultiprocess(异步多进程)

作者：2023面试高手 | 2024-02-22 16:17:44

赞

踩

Python 多进程：multiprocessing、aiomultiprocess(异步多进程)

multiprocessing 文档：https://docs.python.org/zh-cn/3.10/library/multiprocessing.html

Process、Lock、Semaphore、Queue、Pipe、Pool：https://cuiqingcai.com/3335.html

1、进程、线程、并发、并行

进程、线程

  • 进程：可以看作是线程的集合，由一个或多个线程构成。同时进程也是系统进行资源分配和调度的最小独立单位，所以各进程之间的数据是无法共享的，如多个进程无法共享一个全局变量，进程之间的数据共享需要由单独的机制来实现。
  • 线程：是操作系统进行运算调度的最小单位，是进程中的最小运行单元。也是CPU调度的最小单元

并发、并行

在计算机中运行一个程序，底层是通过处理器运行一条条指今来实现的。处理器同一时刻只能执行一条指令，

  • 并发(concurrency)：是指多个线程对应的多条指令被快速轮换地执行。例如：一个处理器，它先执行线程A的指令一段时间，再执行线程B的指令一段时间，然后再切回线程A执行一段时间。处理器执行指令的速度和切换线程的速度都非常快，人完全感知不到计算机在这个过程中还切换了多个线程的上下文，这使得多个线程从宏观上看起来是同时在运行。从微观上看，处理器连续不断地在多个线程之间切换和执行，每个线程的执行都一定会占用这个处理器的一个时间片段，因此同一时刻其实只有一个线程被执行。
  • 并行(parallel)：指同一时刻有多条指令在多个处理器上同时执行，这意着并行必须依赖多个处理器。不论是从宏观还是微观上看，多个线程都是在同一时刻一起执行的。并行只能存在于多处理器系统中，因此如果计算机处理器只有一个核，就不可能实现并行。而并发在单处理器和多处理器系统中都可以存在，因为仅靠一个核，就可以实现并发。例如。系统处理器需要同时运行多个线程。如果系统处理器只有一个核，那它只能通过并发的方式来运行这些线程。而如果系统处理器有多个核，那么在一个核执行一个线程的同时，另一个核可以执行另一个线程，这样这两个线程就实现了并行执行。当然，其他线程也可能和另外的线程在同一个核上执行，它们之间就是并发执行。具体的执行方式，取决于操作系统如何调度。

Python 并发执行

Python 标准库：https://docs.python.org/zh-cn/3.12/library/index.html

2、Python "多线程、多进程"

GIL 对 Python 多线程 影响

GIL (全称为 GlobalInterpreter Lock 意思是全局解释器锁) 设计之初是对数据安全的考虑。在 Python多线程，每个线程的执行方式分如下三步：" 获取 GIL; 执行对应线程的代码; 释放 GIL; "，可见，某个线程要想执行，必须先拿到 GIL，可以把 GIL看作通行证，并且在一个Python进程中，GIL 只有一个。线程要是拿不到通行证，就不允许执行。这样会导致即使在多核条件下，一个Python进程中的多个线程在同一时刻也只能执行一个。

总结：

  • 在 多线程 场景中，Python因为 GIL 的限制，不论是单核还是多核下，同一时刻都只能运行一个线程，这使得 Python 多线程无法发挥多核并行的优势。
  • 在 多进程 场景中 "每个进程都有属于自己的GIL" ，所以在多核处理器下，多进程的运行是不会受GL影响的。所以多进程能够更好地发挥多核优势。但是使用多进程时，要注意 进程间共享数据、进程同步问题。

Python 单线程

就是 Python 代码按照顺序依次执行。执行过程中，不会中断去执行其他代码

  1. import time
  2. import datetime
  3.  
  4.  
  5. def func_music(argv):
  6.     for i in range(2):
  7.         print(f"听音乐  {argv}. {datetime.datetime.now()}")
  8.         time.sleep(1)
  9.  
  10.  
  11. def func_movie(argv):
  12.     for i in range(2):
  13.         print(f"看电影  {argv}. {datetime.datetime.now()}")
  14.         time.sleep(5)
  15.  
  16.  
  17. if __name__ == '__main__':
  18.     func_music('trouble is a friend')
  19.     func_movie('变形金刚')
  20.     print(f"结束: {datetime.datetime.now()}")

Python 多线程

Python 多线程的用法：https://setup.scrape.center/python-threading

Python 中使用线程有两种方式：

  • 通过 类继承 threading.Thread 创建一个线程对象。
  • 通过使用 start_new_thread 函数。

方法 1：类继承 创建线程

通过继承 threading.Thread 并重写 run 方法，来实现多线程。threading.Thread 方法说明

  • t.start()       激活线程，   
  • t.getName()     获取线程的名称   
  • t.setName()     设置线程的名称   
  • t.name          获取或设置线程的名称   
  • t.is_alive()    判断线程是否为激活状态   
  • t.isAlive()     判断线程是否为激活状态   
  • t.setDaemon()   设置为后台线程或前台线程（默认：False）必须在执行start()方法之后;
  • t.isDaemon()    判断是否为守护线程   
  • t.ident         获取线程的标识符。线程标识符是一个非零整数，只有在调用了start()方法之后该属性才有效，否则它只返回None。   
  • t.join()        逐个执行每个线程，执行完毕后继续往下执行，该方法使得多线程变得无意义   
  • t.run()         线程被cpu调度后自动执行线程对象的run方法

示例代码：

  1. import time
  2. import threading
  3. from datetime import datetime
  4.  
  5. exitFlag = 0
  6.  
  7.  
  8. def print_time(thread_name, delay, counter):
  9.     while counter:
  10.         if exitFlag:
  11.             thread_name.exit()
  12.         time.sleep(delay)
  13.         print(f"{thread_name}: {datetime.now()}")
  14.         counter -= 1
  15.  
  16.  
  17. class MyThread(threading.Thread):
  18.     def __init__(self, thread_id, name, delay):
  19.         super().__init__()
  20.         self.thread_id = thread_id
  21.         self.name = name
  22.         self.delay = delay
  23.  
  24.     def run(self):
  25.         print("开始线程：" + self.name)
  26.         print_time(self.name, self.delay, 5)
  27.         print("退出线程：" + self.name)
  28.  
  29.  
  30. # 创建新线程
  31. thread1 = MyThread(1, "Thread-1", 1)
  32. thread2 = MyThread(2, "Thread-2", 2)
  33.  
  34. # 开启新线程
  35. thread1.start()
  36. thread2.start()
  37. thread1.join()
  38. thread2.join()
  39. print("退出主线程")

_thread 和 threading 模块（ 强烈建议直接使用 threading ）。python 提供了两个模块来实现多线程。 _thread 有一些缺点，在 threading 得到了弥补。

方法 2：函数式 创建线程

函数式：调用 _thread 模块中的 start_new_thread() 函数来产生新线程

语法如下：

  • _thread.start_new_thread ( function, args[, kwargs] )

参数说明:

  • function - 线程函数。
  • args - 传递给线程函数的参数，必须是个tuple类型。
  • kwargs - 可选参数。

使用示例：

  1. import time
  2. import _thread
  3.  
  4.  
  5. # 为线程定义一个函数
  6. def print_time(thread_name, delay):
  7.     count = 0
  8.     while count < 5:
  9.         time.sleep(delay)
  10.         count += 1
  11.         print("%s: %s" % (thread_name, time.ctime(time.time())))
  12.  
  13.  
  14. if __name__ == '__main__':
  15.     try:
  16.         # 创建两个线程
  17.         _thread.start_new_thread(print_time, ("Thread-1", 2,))
  18.         _thread.start_new_thread(print_time, ("Thread-2", 4,))
  19.     except BaseException as be:
  20.         print(f"Error: 无法启动线程 ---> {be}")
  21.  
  22.     while True:
  23.         pass

守护线程 ( 精灵线程 )

通过 setDaemon(True) 将线程声明为守护线程，必须在 start() 方法调用之前设置，如果不设置为守护线程程序会被无限挂起。子线程启动后，父线程也继续执行下去，当父线程执行完最后一条语句 print("父进程退出") 后，没有等待子线程，直接就退出了，同时子线程也一同结束。

  1. # -*- coding: utf-8 -*-
  2.  
  3. import time
  4. import threading
  5. import datetime
  6.  
  7.  
  8. def music(music_name):
  9.     for i in range(2):
  10.         print(f"[{datetime.datetime.now().replace(microsecond=0)}] 听音乐 {music_name}")
  11.         time.sleep(1)
  12.  
  13.  
  14. def movie(movie_name):
  15.     for i in range(2):
  16.         print(f"[{datetime.datetime.now().replace(microsecond=0)}] 看电影 {movie_name}")
  17.         time.sleep(5)
  18.  
  19.  
  20. tid_list = [
  21.     threading.Thread(target=music, args=('trouble is a friend',)),
  22.     threading.Thread(target=movie, args=('变形金刚',))
  23. ]
  24.  
  25.  
  26. if __name__ == '__main__':
  27.     for tid in tid_list:
  28.         tid.setDaemon(True)
  29.         tid.start()
  30.     print("父进程退出")
  31.     pass

join() 等待线程终止

join() 的作用：在子线程完成运行之前，这个子线程的父线程将一直被阻塞。注意：上面程序中 join() 方法的位置是在 for 循环外的，也就是说必须等待for循环里的两个进程都结束后，才去执行主进程。

  1. # -*- coding: utf-8 -*-
  2.  
  3. import time
  4. import threading
  5. import datetime
  6.  
  7.  
  8. def music(music_name):
  9.     for i in range(2):
  10.         print(f"[{datetime.datetime.now().replace(microsecond=0)}] 听音乐 {music_name}")
  11.         time.sleep(1)
  12.  
  13.  
  14. def movie(movie_name):
  15.     for i in range(2):
  16.         print(f"[{datetime.datetime.now().replace(microsecond=0)}] 看电影 {movie_name}")
  17.         time.sleep(5)
  18.  
  19.  
  20. tid_list = [
  21.     threading.Thread(target=music, args=('trouble is a friend',)),
  22.     threading.Thread(target=movie, args=('变形金刚',))
  23. ]
  24.  
  25.  
  26. if __name__ == '__main__':
  27.     for tid in tid_list:
  28.         tid.start()
  29.     for tid in tid_list:
  30.         tid.join()
  31.     print("父进程退出")

线程 事件

Python 线程的事件用于主线程控制其他线程的执行。

threading.Event 事件主要提供了三个方法

  • set 。将 “Flag” 设置为 False
  • wait 。将 “Flag” 设置为 True
  • clear 。判断标识位是否为Ture。

事件处理的机制：全局定义了一个 “Flag”，如果 “Flag” 值为 False，那么当程序执行 event.wait 方法时就会阻塞，如果 “Flag” 值为True，那么 event.wait 方法时便不再阻塞。

  1. import threading
  2.  
  3. def do(event):
  4.     print('start')
  5.     event.wait()
  6.     print('execute')
  7.  
  8. event_obj = threading.Event()
  9. for i in range(10):
  10.     t = threading.Thread(target=do, args=(event_obj,))
  11.     t.start()
  12.  
  13. event_obj.clear()
  14. # inp = input('input:')
  15. inp = raw_input('input:')
  16. if inp == 'true':
  17.     event_obj.set()

当线程执行的时候，如果 flag 为False，则线程会阻塞，当 flag 为True 的时候，线程不会阻塞。它提供了 本地 和 远程 的并发性。

线程同步：线程锁

如果多个线程共同对某个数据修改，则可能出现不可预料的结果，为了保证数据的正确性，需要对多个线程进行同步。

多线程的优势在于可以同时运行多个任务（至少感觉起来是这样）。但是当线程需要共享数据时，可能存在数据不同步的问题。考虑这样一种情况：一个列表里所有元素都是 0，线程 "set" 从后向前把所有元素改成1，而线程 "print" 负责从前往后读取列表并打印。那么，可能线程 "set" 开始改的时候，线程 "print" 便来打印列表了，输出就成了一半 0 一半 1，这就是 数据的不同步。
为了避免这种情况，引入了 锁 的概念。锁有两种状态：锁定 和 未锁定。
每当一个线程比如 "set" 要访问共享数据时，必须先获得锁定；如果已经有别的线程比如 "print" 获得锁定了，那么就让线程 "set" 暂停，也就是同步阻塞；等到线程 "print" 访问完毕，释放锁以后，再让线程 "set" 继续。经过这样的处理，打印列表时要么全部输出0，要么全部输出1，不会再出现一半0一半1的尴尬场面。

线程锁

threading.RLock 和 threading.Lock 可以实现简单的线程同步，这两个对象都有 acquire方法 和 release方法。对于那些需要每次只允许一个线程操作的数据，可以将其操作放到 acquire 和release 方法之间。

threading.RLock 和 threading.Lock 的区别：

  • RLock 允许在同一线程中被多次 acquire。而 Lock 却不允许这种情况。
  • RLock 时 acquire 和 release 必须成对出现，即调用了 n 次 acquire，必须调用 n 次的release 才能真正释放所占用的琐。

示例代码

  1. # -*- coding: utf-8 -*-
  2.  
  3. import threading
  4.  
  5. lock = threading.Lock()    # Lock对象
  6. rLock = threading.RLock()  # RLock对象 
  7.  
  8.  
  9. def main_1():
  10.     lock.acquire()
  11.     lock.acquire()  # 产生了死琐。
  12.     lock.release()
  13.     lock.release()
  14.  
  15.  
  16. def main_2():    
  17.     rLock.acquire()
  18.     rLock.acquire()  # 在同一线程内，程序不会堵塞。 
  19.     rLock.release()
  20.     rLock.release()

示例：

  1. # -*- coding: utf-8 -*-
  2.  
  3. import time
  4. import datetime
  5. import threading
  6.  
  7. # 定义一个 "线程锁"
  8. threadLock = threading.Lock()
  9.  
  10. count = 20
  11.  
  12.  
  13. class MyThread(threading.Thread):
  14.     def __init__(self, name, delay_second):
  15.         threading.Thread.__init__(self)
  16.         self.name = name
  17.         self.delay_second = delay_second
  18.  
  19.     def run(self):
  20.         print("Starting " + self.name)
  21.         result = print_time()
  22.         while result > 0:
  23.             print(f"[{datetime.datetime.now().replace(microsecond=0)}] {self.name} ---> {result}")
  24.             result = print_time()
  25.             time.sleep(self.delay_second)
  26.  
  27.  
  28. def print_time():
  29.     # 获得锁，成功获得锁定后返回 True
  30.     # 可选的 timeout 参数不填时将一直阻塞直到获得锁定
  31.     # 否则超时后将返回 False
  32.     threadLock.acquire()
  33.     global count
  34.     count -= 1
  35.     # 释放锁
  36.     threadLock.release()
  37.     return count
  38.  
  39.  
  40. if __name__ == '__main__':
  41.     tid_list = [
  42.         MyThread("thread_1", 1),
  43.         MyThread("thread_2", 1)
  44.     ]
  45.  
  46.     for tid in tid_list:
  47.         tid.start()
  48.     for tid in tid_list:
  49.         tid.join()
  50.     print("主线程退出")
  51.     pass

示例：

  1. import threading
  2. import time
  3. globals_num = 0
  4. lock = threading.RLock()
  5.  
  6.  
  7. def func():
  8.     lock.acquire()  # 获得锁
  9.     global globals_num
  10.     globals_num += 1
  11.     time.sleep(1)
  12.     print(globals_num)
  13.     lock.release()  # 释放锁
  14.  
  15. for i in range(10):
  16.     t = threading.Thread(target=func)
  17.     t.start()
  18.     pass

线程同步：条件变量

一个  threading.Condition 变量总是与某些类型的锁相联系，当几个condition变量必须共享和同一个锁的时候是很有用的。锁 是 conditon 对象的一部分：没有必要分别跟踪。

Condition 类实现了一个 conditon 变量。这个 conditiaon 变量允许一个或多个线程等待，直到他们被另一个线程通知。

  • 如果 lock 参数非空，那么他必须是一个 lock 或者 Rlock 对象，它用来做底层锁。
  • 如果 lock 参数为空，则会创建一个新的 Rlock 对象，用来做底层锁。

condition 变量服从上下文管理协议：with 语句块封闭之前可以获取与锁的联系。
acquire() 和 release() 会调用与锁相关联的相应的方法。
其他和锁关联的方法必须被调用，wait()方法会释放锁，
当另外一个线程使用 notify() or notify_all()唤醒它之前会一直阻塞。一旦被唤醒，wait()会重新获得锁并返回，
wait(timeout=None) ：等待通知，或者等到设定的超时时间。
当调用这wait()方法时，如果调用它的线程没有得到锁，那么会抛出一个RuntimeError异常。
wati()释放锁以后，在被调用相同条件的另一个进程用notify() or notify_all() 叫醒之前会一直阻塞。
wait()还可以指定一个超时时间。 如果有等待的线程，notify()方法会唤醒一个在等待conditon变量的线程。notify_all() 则会唤醒所有在等待conditon变量的线程。

注意： notify()和notify_all()不会释放锁，也就是说，线程被唤醒后不会立刻返回他们的wait() 调用。
除非线程调用notify()和notify_all()之后放弃了锁的所有权。
在典型的设计风格里，利用condition变量用锁去通许访问一些共享状态，线程在获取到它想得到的状态前，会反复调用wait()。
修改状态的线程在他们状态改变时调用 notify() or notify_all()，用这种方式，线程会尽可能的获取到想要的一个等待者状态。

例子：生产者-消费者模型

  1. import threading
  2. import time
  3.  
  4.  
  5. def consumer(cond):
  6.     with cond:
  7.         print("consumer before wait")
  8.         cond.wait()
  9.         print("consumer after wait")
  10.  
  11.  
  12. def producer(cond):
  13.     with cond:
  14.         print("producer before notifyAll")
  15.         cond.notifyAll()
  16.         print("producer after notifyAll")
  17.  
  18.  
  19. condition = threading.Condition()
  20. consumer_1 = threading.Thread(name="c1", target=consumer, args=(condition,))
  21. consumer_2 = threading.Thread(name="c2", target=consumer, args=(condition,))
  22. producer = threading.Thread(name="p", target=producer, args=(condition,))
  23.  
  24. consumer_1.start()
  25. time.sleep(2)
  26. consumer_2.start()
  27. time.sleep(2)
  28. producer.start()

线程同步：线程安全的 queue

使用 "线程锁" 可以实现线程同步，

同样，使用 "线程安全的 queue (队列)" 也可以达到线程同步。

Python 的 queue ( 线程安全 )

：https://docs.python.org/zh-cn/3.10/search.html?q=queue+&check_keywords=yes&area=default

Python 的 queue 模块中提供了同步的、线程安全的队列类。包括

  • FIFO（先入先出) 队列
  • LIFO（后入先出）队列
  • 优先级队列 PriorityQueue

这些队列都实现了 锁原语，能够在多线程中直接使用。可以使用队列来实现线程间的同步。

 Queue 模块中的常用方法:

  1. Queue.qsize()    返回队列的大小
  2. Queue.empty()    如果队列为空，返回True,反之False
  3. Queue.full()     如果队列满了，返回True,反之False
  4. Queue.full 与 maxsize 大小对应
  5. Queue.get([block[, timeout]]) 获取队列，timeout是等待时间
  6. Queue.get_nowait()            相当Queue.get(False)
  7. Queue.put(item)               写入队列，timeout是等待时间
  8. Queue.put_nowait(item)        相当Queue.put(item, False)
  9. Queue.task_done()             在完成一项工作之后，Queue.task_done()函数向任务已经完成的队列发送一个信号
  10. Queue.join()                  实际上意味着等到队列为空，再执行别的操作

示例：

  1. # -*- coding: utf-8 -*-
  2.  
  3. import time
  4. import queue
  5. import threading
  6.  
  7. task_queue = queue.Queue()
  8.  
  9.  
  10. def produce():
  11.     while True:
  12.         for num in range(100):
  13.             task_queue.put(num)
  14.             time.sleep(0.1)
  15.  
  16.  
  17. def consume():
  18.     while True:
  19.         if task_queue.empty():
  20.             print('队列为空')
  21.             continue
  22.         num = task_queue.get()
  23.         print(num)
  24.         time.sleep(1)
  25.  
  26.  
  27. if __name__ == '__main__':
  28.     thread_list = []
  29.     t1 = threading.Thread(target=produce)
  30.     thread_list.append(t1)
  31.     for i in range(3):
  32.         t_id = threading.Thread(target=consume)
  33.         thread_list.append(t_id)
  34.     for index in thread_list:
  35.         index.start()
  36.     for index in thread_list:
  37.         index.join()

Python 多进程

Python 多进程的用法：https://setup.scrape.center/python-multiprocessingo

把一个 "多线程" 改成 "多进程"，主要有下面几种方法：

  • subprocess
  • signal
  • multiprocessing
  • concurrent.futures 标准库。

concurrent.futures 标准库

"concurrent.futures 标准库" 提供了 ThreadPoolExecutor 和 ProcessPoolExecutor 两个类，实现了对 threading 和 multiprocessing 的更高级的抽象，对编写 " 线程池 / 进程池 "  提供了直接的支持。

使用示例代码：

  1. import redis
  2. from redis import WatchError
  3. from concurrent.futures import ProcessPoolExecutor
  4.  
  5. r = redis.Redis(host='127.0.0.1', port=6379)
  6.  
  7. # 减库存函数, 循环直到减库存完成
  8. # 库存充足, 减库存成功, 返回True
  9. # 库存不足, 减库存失败, 返回False
  10.  
  11.  
  12. def reduce_stock():
  13.  
  14.     # python中redis事务是通过pipeline的封装实现的
  15.     with r.pipeline() as pipe:
  16.         while True:
  17.             try:
  18.                 # watch库存键, multi后如果该key被其他客户端改变, 
  19.                 # 事务操作会抛出WatchError异常
  20.                 pipe.watch('stock:count')
  21.                 count = int(pipe.get('stock:count'))
  22.                 if count > 0:  # 有库存
  23.                     # 事务开始
  24.                     pipe.multi()
  25.                     pipe.decr('stock:count')
  26.                     # 把命令推送过去
  27.                     # execute返回命令执行结果列表, 这里只有一个decr返回当前值
  28.                     print(pipe.execute()[0])
  29.                     return True
  30.                 else:
  31.                     return False
  32.             except WatchError as ex:
  33.                 # 打印WatchError异常, 观察被watch锁住的情况
  34.                 print(ex)
  35.                 pipe.unwatch()
  36.  
  37.  
  38. def worker():
  39.     while True:
  40.         # 没有库存就退出
  41.         if not reduce_stock():
  42.             break
  43.  
  44.  
  45. def main():
  46.     # 设置库存为100
  47.     r.set("stock:count", 100)
  48.     # 多进程模拟多个客户端提交
  49.     with ProcessPoolExecutor() as pool:
  50.         for _ in range(10):
  51.             pool.submit(worker)
  52.  
  53.  
  54. if __name__ == "__main__":
  55.     main()
  56.     pass


 

3、multiprocessing  模块

文档：https://docs.python.org/zh-cn/3.13/library/multiprocessing.html

概述

由于 Python(这里指 CPython) 限制(GLI) 最多只能用满1个CPU核心。但是 multiprocessing 使用 "子进程 而非线程" 有效地绕过了 全局解释器锁，从而可以使用多核处理器。

multiprocessing 包中也有 Lock / Event / Semaphore / Condition类，用来同步进程，其用法也与 threading 包中的同名类一样。multiprocessing 的很大一部份与 threading 使用同一套 API，只不过换到了多进程的场景。但在使用这些共享API的时候，我们要注意以下几点：

  • 在 UNIX 平台上，当某个进程终结之后，该进程需要被其父进程调用 wait，否则进程成为僵尸进程(Zombie)。所以，有必要对每个Process对象调用join()方法 (实际上等同于wait)。对于多线程来说，由于只有一个进程，所以不存在此必要性。
  • multiprocessing 提供了 threading 包中没有的 IPC ( 比如：Pipe 和 Queue )，效率上更高。应优先考虑 Pipe 和 Queue，避免使用 Lock / Event / Semaphore / Condition 等同步方式 (因为它们占据的不是用户进程的资源 )。
  • 多进程应该避免共享资源。在多线程中，我们可以比较容易地共享资源，比如 使用全局变量或者传递参数。在多进程情况下，由于每个进程有自己独立的内存空间，以上方法并不合适。此时我们可以通过 "共享内存" 和 "Manager 的方法" 来共享资源。但这样做提高了程序的复杂度，并因为同步的需要而降低了程序的效率。
  • Process.PID中保存有PID，如果进程还没有start()，则PID为None。

multiprocessing 的 __init__.py，通过 __all__ 可以看到所有可用的 对象或者属性

Process

Process 类的构造方法：

参数说明：
    group：进程所属组。基本不用
    target：表示调用对象。
    args：表示调用对象的位置参数元组。
    name：别名
    kwargs：表示调用对象的字典。

利用 multiprocessing.Process 可以创建一个 Process 对象，该对象与 Thread对象 的用法相同，拥有如下方法：

  • is_alive()
  • join([timeout])
  • run()
  • start()
  • terminate()
  • ........

属性有：

  • authkey
  • daemon（要通过start()设置）、
  • exitcode(进程在运行时为None、如果为–N，表示被信号N结束）、
  • name
  • pid
  • ........

直接 创建 进程

  1. # -*- coding: utf-8 -*-
  2.  
  3. from multiprocessing import Process, current_process
  4.  
  5.  
  6. def worker_func(num):
  7.     print(f'worker_func [进程号:{current_process().ident}] ---> {num}')
  8.  
  9.  
  10. def main():
  11.     pid_list = []
  12.     for i in range(5):
  13.         pid = Process(target=worker_func, args=(i,))
  14.         pid_list.append(pid)
  15.     for pid in pid_list:
  16.         pid.start()
  17.     for pid in pid_list:
  18.         pid.join()
  19.  
  20.  
  21. if __name__ == '__main__':
  22.     main()

继承 multiprocessing.Process 创建 进程

利用 class 来创建进程，定制子类

  1. from multiprocessing import Process, current_process
  2.  
  3.  
  4. class Worker(Process):
  5.  
  6.     def run(self):
  7.         print(f'进程号:{current_process().ident} ---> 进程名:{self.name}')
  8.         return
  9.  
  10.  
  11. if __name__ == '__main__':
  12.     jobs = []
  13.     for i in range(5):
  14.         p = Worker()
  15.         jobs.append(p)
  16.         p.start()
  17.     for j in jobs:
  18.         j.join()

守护进程

守护进程就是主程序退出后，子进程不退出，仍然继续执行。

mutilprocess.setDaemon(True) 设置 "精灵进程"

也可以等待守护进程退出，要加上 join，join 可以传入浮点数值，等待n久就不等了

  1. import multiprocessing
  2. import time
  3. import sys
  4.  
  5.  
  6. def daemon():
  7.     name = multiprocessing.current_process().name
  8.     print('Starting:', name)
  9.     time.sleep(2)
  10.     print('Exiting :', name)
  11.  
  12.  
  13. def non_daemon():
  14.     name = multiprocessing.current_process().name
  15.     print('Starting:', name)
  16.     print('Exiting :', name)
  17.  
  18.  
  19. if __name__ == '__main__':
  20.     pid_1 = multiprocessing.Process(name='daemon', target=daemon)
  21.     pid_1.daemon = True
  22.  
  23.     pid_2 = multiprocessing.Process(name='non-daemon', target=non_daemon)
  24.     pid_2.daemon = False
  25.  
  26.     pid_1.start()
  27.     pid_2.start()
  28.  
  29.     pid_1.join(1)
  30.     print(f'pid_2.is_alive() ---> {pid_2.is_alive()}')
  31.     pid_2.join()

终止进程

最好使用 poison pill，强制的使用 terminate()。注意 terminate 之后要 join，使其可以更新状态

  1. import multiprocessing
  2. import time
  3.  
  4.  
  5. def slow_worker():
  6.     print('Starting worker')
  7.     time.sleep(0.1)
  8.     print('Finished worker')
  9.  
  10.  
  11. if __name__ == '__main__':
  12.     p = multiprocessing.Process(target=slow_worker)
  13.     print('BEFORE:', p, p.is_alive())
  14.  
  15.     p.start()
  16.     print('DURING:', p, p.is_alive())
  17.  
  18.     p.terminate()
  19.     print('TERMINATED:', p, p.is_alive())
  20.  
  21.     p.join()
  22.     print('JOINED:', p, p.is_alive())

进程的退出状态

  •  == 0     未生成任何错误
  •  0           进程有一个错误，并以该错误码退出
  •  < 0       进程由一个-1 * exitcode信号结束
  1. import multiprocessing
  2. import sys
  3. import time
  4.  
  5.  
  6. def exit_error():
  7.     sys.exit(1)
  8.  
  9.  
  10. def exit_ok():
  11.     return
  12.  
  13.  
  14. def return_value():
  15.     return 1
  16.  
  17.  
  18. def raises():
  19.     raise RuntimeError('There was an error!')
  20.  
  21.  
  22. def terminated():
  23.     time.sleep(3)
  24.  
  25.  
  26. if __name__ == '__main__':
  27.     jobs = []
  28.     for f in [exit_error, exit_ok, return_value, raises, terminated]:
  29.         print('Starting process for', f.func_name)
  30.         j = multiprocessing.Process(target=f, name=f.func_name)
  31.         jobs.append(j)
  32.         j.start()
  33.  
  34.     jobs[-1].terminate()
  35.  
  36.     for j in jobs:
  37.         j.join()
  38.         print('%15s.exitcode = %s' % (j.name, j.exitcode))

异常

进程间通信 (IPC)：管道、队列

multiprocessing 提供了 threading 包中没有的 IPC ( 比如：Pipe 和 Queue )，效率上更高。应优先考虑 Pipe 和 Queue，避免使用 Lock / Event / Semaphore / Condition 等同步方式 (因为它们占据的不是用户进程的资源 )。Pipe 和 Queue 可以用来传送常见的对象。

管道

Pipe 可以是单向(half-duplex)，也可以是双向(duplex)。

通过mutiprocessing.Pipe(duplex=False) 创建单向管道 (默认为双向)。一个进程从 PIPE 一端输入对象，然后被 PIPE 另一端的进程接收，单向管道只允许管道一端的进程输入，而双向管道则允许从两端输入。下面的程序展示了 Pipe 的使用：（ 这里的 Pipe 是双向的。 ）

  1. import multiprocessing as mul
  2.  
  3.  
  4. def proc1(pipe=None):
  5.     pipe.send('hello')
  6.     print('proc1 rec:', pipe.recv())
  7.  
  8.  
  9. def proc2(pipe=None):
  10.     print('proc2 rec:', pipe.recv())
  11.     pipe.send('hello, too')
  12.  
  13.  
  14. # Build a pipe
  15. pipe = mul.Pipe()
  16.  
  17. # Pass an end of the pipe to process 1
  18. p1 = mul.Process(target=proc1, args=(pipe[0],))
  19. # Pass the other end of the pipe to process 2
  20. p2 = mul.Process(target=proc2, args=(pipe[1],))
  21. p1.start()
  22. p2.start()
  23. p1.join()
  24. p2.join()

Pipe 对象建立的时候，返回一个含有两个元素的表，每个元素代表 Pipe 的一端(Connection对象)。对 Pipe 的某一端调用 send() 方法来传送对象，在另一端使用 recv() 来接收。

队列 Queue

mutiprocessing.Queue 与 Pipe 相类似，都是先进先出的结构。但 mutiprocessing.Queue允许多个进程放入，多个进程从队列取出对象。下面的程序展示了 Queue 的使用:

  1. import os
  2. import multiprocessing
  3. import time
  4.  
  5.  
  6. # input worker
  7. def input_queue(queue=None):
  8.     info = str(os.getpid()) + '(put):' + str(time.time())
  9.     queue.put(info)
  10.  
  11.  
  12. # output worker
  13. def output_queue(queue=None, lock):
  14.     info = queue.get()
  15.     lock.acquire()
  16.     print(str(os.getpid()) + '(get):' + info)
  17.     lock.release()
  18.  
  19.  
  20. # ===================
  21. # Main
  22. record1 = []  # store input processes
  23. record2 = []  # store output processes
  24. lock = multiprocessing.Lock()  # To prevent messy print
  25. queue = multiprocessing.Queue(3)
  26.  
  27. # input processes
  28. for i in range(10):
  29.     process = multiprocessing.Process(target=input_queue, args=(queue,))
  30.     process.start()
  31.     record1.append(process)
  32.  
  33. # output processes
  34. for i in range(10):
  35.     process = multiprocessing.Process(target=output_queue, args=(queue, lock))
  36.     process.start()
  37.     record2.append(process)
  38.  
  39. for p in record1:
  40.     p.join()
  41.  
  42. queue.close()  # No more object will come, close the queue
  43.  
  44. for p in record2:
  45.     p.join()

一些进程使用 put() 在 Queue 中放入字符串，这个字符串中包含 PID 和时间。另一些进程从Queue 中取出，并打印自己的 PID 以及 get() 的字符串

简单队列 SimpleQueue

JoinableQueue

示例：

 Python 进程间传递消息，一般的情况是 Queue 来传递。

  • multiprocessing.Queue "进程安全的队列" 。multiprocessing.Queue 在 "多线程环境中并不适用"，因为它无法提供线程安全的特性。
  • queue.Queue（位于标准库的 queue 模块中）提供了 "线程安全的队列"，并且可供多个线程同时使用。
  1. import multiprocessing
  2.  
  3.  
  4. class MyFancyClass(object):
  5.  
  6.     def __init__(self, name):
  7.         self.name = name
  8.  
  9.     def do_something(self):
  10.         proc_name = multiprocessing.current_process().name
  11.         print(f"{proc_name} ---> {self.name}")
  12.  
  13.  
  14. def worker(q):
  15.     obj = q.get()
  16.     obj.do_something()
  17.  
  18.  
  19. if __name__ == '__main__':
  20.     queue = multiprocessing.Queue()
  21.     p = multiprocessing.Process(target=worker, args=(queue,))
  22.     p.start()
  23.     queue.put(MyFancyClass('Fancy Dan'))
  24.     # Wait for the worker to finish
  25.     queue.close()
  26.     queue.join_thread()
  27.     p.join()

示例：

  1. import time
  2. import multiprocessing
  3.  
  4.  
  5. class Consumer(multiprocessing.Process):
  6.  
  7.     def __init__(self, task_queue, result_queue):
  8.         super(Consumer, self).__init__()
  9.         self.task_queue = task_queue
  10.         self.result_queue = result_queue
  11.  
  12.     def run(self):
  13.         proc_name = self.name
  14.         while True:
  15.             next_task = self.task_queue.get()
  16.             if next_task is None:
  17.                 # Poison pill means shutdown
  18.                 print(f"退出进程 ---> {proc_name}")
  19.                 self.task_queue.task_done()
  20.                 break
  21.             print(f'{proc_name} 下一个任务 ---> {next_task}')
  22.             answer = next_task()
  23.             self.task_queue.task_done()
  24.             self.result_queue.put(answer)
  25.         return
  26.  
  27.  
  28. class Task(object):
  29.     def __init__(self, a, b):
  30.         self.a = a
  31.         self.b = b
  32.  
  33.     def __call__(self):
  34.         time.sleep(0.1)  # pretend to take some time to do the work
  35.         return f'{self.a} * {self.b} = {self.a * self.b}'
  36.  
  37.     def __str__(self):
  38.         return str(self.a * self.a)
  39.  
  40.  
  41. if __name__ == '__main__':
  42.     # Establish communication queues
  43.     tasks = multiprocessing.JoinableQueue()
  44.     results = multiprocessing.Queue()
  45.  
  46.     # Start consumers
  47.     num_consumers = multiprocessing.cpu_count() * 2
  48.     print(f"创建 {num_consumers} 消费者")
  49.     consumers = [Consumer(tasks, results) for i in range(num_consumers)]
  50.     for w in consumers:
  51.         w.start()
  52.  
  53.     # Enqueue jobs
  54.     num_jobs = 10
  55.     for i in range(num_jobs):
  56.         tasks.put(Task(i, i))
  57.  
  58.     # Add a poison pill for each consumer
  59.     for i in range(num_consumers):
  60.         tasks.put(None)
  61.  
  62.     # Wait for all tasks to finish
  63.     tasks.join()
  64.  
  65.     # Start printing results
  66.     while num_jobs:
  67.         result = results.get()
  68.         print(f"结果 ---> {result}")
  69.         num_jobs -= 1

示例：队列

  1. from multiprocessing import Process, Queue
  2.  
  3. def f(q):
  4.     q.put([42, None, 'hello'])
  5.  
  6. if __name__ == '__main__':
  7.     q = Queue()
  8.     p = Process(target=f, args=(q,))
  9.     p.start()
  10.     print(q.get())    # prints "[42, None, 'hello']"
  11.     p.join()

示例：管道

  1. from multiprocessing import Process, Pipe
  2.  
  3. def f(conn):
  4.     conn.send([42, None, 'hello'])
  5.     conn.close()
  6.  
  7. if __name__ == '__main__':
  8.     parent_conn, child_conn = Pipe()
  9.     p = Process(target=f, args=(child_conn,))
  10.     p.start()
  11.     print(parent_conn.recv())   # prints "[42, None, 'hello']"
  12.     p.join()

杂项

连接对象（Connection）

进程  同步

同步原语

通常来说同步原语在多进程环境中并不像它们在多线程环境中那么必要。参考 threading 模块的文档。multiprocessing 包含来自 threading 的所有同步原语的等价物。也可以使用管理器对象创建同步原语，参考 管理器 。

例如，可以使用锁来确保一次只有一个进程打印到标准输出:

  1. from multiprocessing import Process, Lock
  2.  
  3. def f(l, i):
  4.     l.acquire()
  5.     try:
  6.         print('hello world', i)
  7.     finally:
  8.         l.release()
  9.  
  10. if __name__ == '__main__':
  11.     lock = Lock()
  12.  
  13.     for num in range(10):
  14.         Process(target=f, args=(lock, num)).start()

不使用锁的情况下，来自于多进程的输出很容易产生混淆。

我们可以从下面的程序中看到 Thread 对象和 Process对象 在使用上的相似性与结果上的不同。各个线程和进程都做一件事：打印PID。但问题是，所有的任务在打印的时候都会向同一个标准输出(stdout)输出。这样输出的字符会混合在一起，无法阅读。使用 Lock 同步，在一个任务输出完成之后，再允许另一个任务输出，可以避免多个任务同时向终端输出。

  1. import os
  2. import threading
  3. import multiprocessing
  4.  
  5.  
  6. # worker function
  7. def worker(sign=None, t_lock=None):
  8.     t_lock.acquire()
  9.     print(sign, os.getpid())
  10.     t_lock.release()
  11.  
  12.  
  13. # Main
  14. print('Main:', os.getpid())
  15.  
  16. # Multi-thread
  17. record = []
  18. threading_lock = threading.Lock()
  19. for i in range(5):
  20.     thread = threading.Thread(target=worker, args=('thread', threading_lock))
  21.     thread.start()
  22.     record.append(thread)
  23.  
  24. for thread in record:
  25.     thread.join()
  26.  
  27. # Multi-process
  28. record = []
  29. process_lock = multiprocessing.Lock()
  30. for i in range(5):
  31.     process = multiprocessing.Process(target=worker, args=('process', process_lock))
  32.     process.start()
  33.     record.append(process)
  34.  
  35. for process in record:
  36.     process.join()

所有 Thread 的 PID 都与主程序相同，而每个 Process 都有一个不同的 PID。

Event 提供一种简单的方法，可以在进程间传递状态信息。事件可以切换设置和未设置状态。通过使用一个可选的超时值，时间对象的用户可以等待其状态从未设置变为设置。

  1. import multiprocessing
  2. import time
  3.  
  4.  
  5. def wait_for_event(e):
  6.     """Wait for the event to be set before doing anything"""
  7.     print('wait_for_event: starting')
  8.     e.wait()
  9.     print('wait_for_event: e.is_set()->', e.is_set())
  10.  
  11.  
  12. def wait_for_event_timeout(e, t):
  13.     """Wait t seconds and then timeout"""
  14.     print('wait_for_event_timeout: starting')
  15.     e.wait(t)
  16.     print('wait_for_event_timeout: e.is_set()->', e.is_set())
  17.  
  18.  
  19. if __name__ == '__main__':
  20.     e = multiprocessing.Event()
  21.     w1 = multiprocessing.Process(name='block', target=wait_for_event, args=(e,))
  22.     w1.start()
  23.  
  24.     w2 = multiprocessing.Process(name='nonblock', target=wait_for_event_timeout, args=(e, 2))
  25.     w2.start()
  26.  
  27.     print('main: waiting before calling Event.set()')
  28.     time.sleep(3)
  29.     e.set()
  30.     print('main: event is set')

共享内存：共享 ctypes 对象

应该尽量避免多进程共享资源。多进程共享资源必然会带来进程间相互竞争。而这种竞争又会造成race condition，我们的结果有可能被竞争的不确定性所影响。但如果需要，依然可以通过 共享内存 和 Manager对象 这么做。

在共享内存上创建可被子进程继承的共享对象时是可行的。

可以使用 Value 或 Array 将数据存储在共享内存映射中。例如，以下代码：

  1. from multiprocessing import Process, Value, Array
  2.  
  3.  
  4. def f(n, a):
  5.     n.value = 3.1415927
  6.     for i in range(len(a)):
  7.         a[i] = -a[i]
  8.  
  9.  
  10. if __name__ == '__main__':
  11.     num = Value('d', 0.0)
  12.     arr = Array('i', range(10))
  13.  
  14.     p = Process(target=f, args=(num, arr))
  15.     p.start()
  16.     p.join()
  17.  
  18.     print(num.value)
  19.     print(arr[:])

将打印
3.1415927
[0, -1, -2, -3, -4, -5, -6, -7, -8, -9]
创建 num 和 arr 时使用的 'd' 和 'i' 参数是 array 模块使用的类型的 typecode ： 'd' 表示双精度浮点数， 'i' 表示有符号整数。这些共享对象将是进程和线程安全的。

为了更灵活地使用共享内存，可以使用 multiprocessing.sharedctypes 模块，该模块支持创建从共享内存分配的任意 ctypes 对象。

例子:

  1. import multiprocessing
  2.  
  3.  
  4. def f(n, a):
  5.     n.value = 3.14
  6.     a[0] = 5
  7.  
  8.  
  9. num = multiprocessing.Value('d', 0.0)
  10. arr = multiprocessing.Array('i', range(10))
  11.  
  12. p = multiprocessing.Process(target=f, args=(num, arr))
  13. p.start()
  14. p.join()
  15.  
  16. print(num.value)
  17. print(arr[:])

这里我们实际上只有主进程和Process对象代表的进程。我们在主进程的内存空间中创建共享的内存，也就是Value和Array两个对象。对象Value被设置成为双精度数(d), 并初始化为0.0。而Array则类似于C中的数组，有固定的类型(i, 也就是整数)。在Process进程中，我们修改了Value和Array对象。回到主程序，打印出结果，主程序也看到了两个对象的改变，说明资源确实在两个进程之间共享。

管理器 Manager()

管理器 Manager() 提供了一种创建共享数据的方法，从而可以在不同进程中共享，甚至可以通过网络跨机器共享数据。管理器维护一个用于管理 共享对象 的服务。其他进程可以通过代理访问这些共享对象。

由 Manager() 返回的管理器对象控制一个服务进程，该进程保存Python对象并允许其他进程使用代理操作它们。Manager()返回的管理器支持类型： list 、 dict 、 Namespace 、 Lock 、 RLock 、 Semaphore 、 BoundedSemaphore 、 Condition 、 Event 、 Barrier 、 Queue 、 Value 和 Array 。

管理器的特点：

服务器进程管理器比使用共享内存对象更灵活，它们支持二进制对象类型。

同时，一个单独的manager可以被网络上的不同计算机的进程共享。

缺点是比使用shared memory慢。

代理对象

共享资源 --- Manager ( 分布式 进程 )

Manager 对象类似于 服务器 与 客户 之间的通信 (server-client)。

  • 用一个进程作为服务器，建立 Manager 来真正存放资源。
  • 其它的进程可以通过 "网络传递参数" 的方式访问 Manager，建立连接后，操作服务器上的资源。即 Manager 可以运用于多计算机，实现分布式。

Python 中 multiprocessing 的子模块 managers 支持把多进程分布到多台机器上，一个服务进程可以作为调度者来将任务分布到其它的多个进程当中，并依靠网络进行互相通信。由于managers的模块封装好了，所以在Python中我们调用它时可以不需要了解网络通信的底层细节，就可以直接进行分布式多进程程序的编写。

Manager() 返回一个 manager 对象。它控制一个服务器进程，这个进程会管理Python对象并允许其他进程通过代理的方式来操作这些对象。

Manager 的使用类似于 shared memory，但可以共享更丰富的对象类型。

  1. import multiprocessing
  2.  
  3.  
  4. def f(x, arr, l):
  5.     x.value = 3.14
  6.     arr[0] = 5
  7.     l.append('Hello')
  8.  
  9.  
  10. server = multiprocessing.Manager()
  11. x = server.Value('d', 0.0)
  12. arr = server.Array('i', range(10))
  13. l = server.list()
  14.  
  15. proc = multiprocessing.Process(target=f, args=(x, arr, l))
  16. proc.start()
  17. proc.join()
  18.  
  19. print(x.value)
  20. print(arr)
  21. print(l)

Manager 利用 list() 方法提供了表的共享方式。实际上你可以利用 dict() 来共享词典，Lock() 来共享 threading.Lock ( 注意，我们共享的是 threading.Lock，而不是进程的 mutiprocessing.Lock。后者本身已经实现了进程共享) 等。 这样 Manager 就允许我们共享更多样的对象。

多台主机 分布式 进程

服务端

3.8版本增加了freeze_support()函数。主要是为了支持windows可执行文件。毕竟multiprocessing可用于分布式进程。所以必须引入freeze_support：

  1. import random, time, queue
  2. from multiprocessing.managers import BaseManager
  3. from multiprocessing import freeze_support
  4.  
  5. # 建立2个队列，一个发送，一个接收
  6. task_queue = queue.Queue()
  7. result_queue = queue.Queue()
  8.  
  9.  
  10. def get_task():
  11.     return task_queue
  12.  
  13.  
  14. def get_result():
  15.     return result_queue
  16.  
  17.  
  18. class QueueManager(BaseManager):
  19.     pass
  20.  
  21.  
  22. # 服务器的管理器上注册2个共享队列
  23. QueueManager.register('get_task', callable=get_task)
  24. QueueManager.register('get_result', callable=get_result)
  25. # 设置端口，地址默认为空。验证码authkey需要设定。
  26. manager = QueueManager(address=('127.0.0.1', 5000), authkey=b'abc')
  27.  
  28.  
  29. def manager_run():
  30.     manager.start()
  31.     # 通过管理器访问共享队列。
  32.     task = manager.get_task()
  33.     result = manager.get_result()
  34.  
  35.     # 对队列进行操作, 往task队列放进任务。
  36.     for value in range(10):
  37.         n = random.randint(0, 100)
  38.         print('Put task %d' % n)
  39.         task.put(n)
  40.     # 从result队列取出结果
  41.     print('Try get result...')
  42.     try:
  43.         for value in range(10):
  44.             r = result.get(timeout=10)
  45.             print('Result: %s' % r)
  46.     except queue.Empty:
  47.         print('result is empty')
  48.     # 关闭管理器。
  49.     manager.shutdown()
  50.     print('master exit.')
  51.  
  52.  
  53. if __name__ == '__main__':
  54.     freeze_support()
  55.     manager_run()

客户端

  1. import time, sys, queue
  2. from multiprocessing.managers import BaseManager
  3.  
  4.  
  5. class QueueManager(BaseManager):
  6.     pass
  7.  
  8.  
  9. # 从网络上的服务器上获取Queue，所以注册时只提供服务器上管理器注册的队列的名字:
  10. QueueManager.register('get_task')
  11. QueueManager.register('get_result')
  12.  
  13. server_addr = '127.0.0.1'
  14. print('Connect to server %s...' % server_addr)
  15. # b'abc'相当于'abc'.encode('ascii'),类型是bytes
  16. m = QueueManager(address=(server_addr, 5000), authkey=b'abc')
  17. # 连接服务器
  18. m.connect()
  19. # 获得服务器上的队列对象
  20. task = m.get_task()
  21. result = m.get_result()
  22.  
  23. for value in range(10):
  24.     try:
  25.         n = task.get(timeout=1)
  26.         print('run task %d * %d...' % (n, n))
  27.         r = '%d * %d = %d' % (n, n, n * n)
  28.         time.sleep(1)
  29.         result.put(r)
  30.     except queue.Empty:
  31.         print('task queue is empty')
  32.  
  33. print('worker exit.')

进程池

直接使用 Process类动态的生成多个进程，十几个还好，但是如果上百个甚至更多，那手动去限制进程数量就显得特别的繁琐，此时 进程池 就派上用场了。进程池 (Process Pool) 可以创建多个进程。这些进程就像是随时待命的士兵，准备执行任务(程序)。一个进程池中可以容纳多个待命的士兵。

Pool 类可以提供指定数量的进程供用户调用，当有新的请求提交到Pool中时，如果池还没有满，就会创建一个新的进程来执行请求。如果池满，请求就会告知先等待，直到池中有进程结束，才会创建新的进程来执行这些请求。

Pool 类描述了一个工作进程池，他有几种不同的方法让任务卸载工作进程。 进程池内部维护一个进程序列，当使用时，则去进程池中获取一个进程，如果进程池序列中没有可供使用的进程，那么程序就会等待，直到进程池中有可用进程为止。我们可以用 Pool 类创建一个进程池，展开提交的任务给进程池。

一个进程池对象可以控制工作进程池的哪些工作可以被提交，它支持 超时 和 回调的异步结果，有一个类似 map 的实现。

参数

  • processes ：进程的数量，如果 processes 是 None 那么使用 os.cpu_count() 返回的数量。
  • initializer：如果是 None，那么每一个工作进程在开始的时候会调 initializer(*initargs)。 
  • maxtasksperchild：工作进程退出之前可以完成的任务数，完成后用一个新的工作进程来替代原进程，来让闲置的资源被释放。maxtasksperchild 默认是 None，意味着只要 Pool 存在工作进程就会一直存活。 
  • context：用在制定工作进程启动时的上下文，一般使用 multiprocessing.Pool() 或者 一个context 对象的 Pool() 方法来创建一个池，两种方法都适当的设置了context

进程池的方法 

  • apply(func[, args[, kwds]]) ：调用 func 函数并传递 args 和 kwds，不建议使用，并且 3.x 以后不在出现。结果返回前主进程会被阻塞直到函数执行结束，由于这个原因，apply_async() 更适合并发执行。
  • apply_async(func[, args[, kwds[, callback[, error_callback]]]]) ： apply() 方法的一个变体，会返回一个结果对象。 如果 callback 被指定，那么 callback 可以接收一个参数然后被调用，当结果准备好回调时会调用 callback，调用失败时，则用 error_callback 替换 callback。 Callbacks 应被立即完成，否则处理结果的线程会被阻塞。   非阻塞 版本
  • close() ：阻止更多的任务提交到 pool，待任务完成后，工作进程会退出。
  • terminate() ：不管任务是否完成，立即停止工作进程。在对 pool 对象进程垃圾回收的时候，会立即调用 terminate()。
  • join()：wait 工作线程的退出，在调用 join() 前，必须调用 close() 或者 terminate()。这样是因为被终止的进程需要被父进程调用 wait（join等价与wait），否则进程会成为僵尸进程。( 主进程阻塞，等待子进程的退出， join 方法要在 close 或 terminate 之后使用。 )
  • map(func, iterable[, chunksize])   
  • map_async(func, iterable[, chunksize[, callback[, error_callback]]])
  • imap(func, iterable[, chunksize])
  • imap_unordered(func, iterable[, chunksize])   
  • starmap(func, iterable[, chunksize])
  • starmap_async(func, iterable[, chunksize[, callback[, error_back]]])

Pool 类表示一个工作进程池。它具有允许以几种不同方式将任务分配到工作进程的方法。

  1. from multiprocessing import Pool, TimeoutError
  2. import time
  3. import os
  4.  
  5.  
  6. def f(x):
  7.     return x * x
  8.  
  9.  
  10. if __name__ == '__main__':
  11.     # start 4 worker processes
  12.     with Pool(processes=4) as pool:
  13.  
  14.         # print "[0, 1, 4,..., 81]"
  15.         print(pool.map(f, range(10)))
  16.  
  17.         # print same numbers in arbitrary order
  18.         for i in pool.imap_unordered(f, range(10)):
  19.             print(i)
  20.  
  21.         # evaluate "f(20)" asynchronously
  22.         res = pool.apply_async(f, (20,))  # runs in *only* one process
  23.         print(res.get(timeout=1))  # prints "400"
  24.  
  25.         # evaluate "os.getpid()" asynchronously
  26.         res = pool.apply_async(os.getpid, ())  # runs in *only* one process
  27.         print(res.get(timeout=1))  # prints the PID of that process
  28.  
  29.         # launching multiple evaluations asynchronously *may* use more processes
  30.         multiple_results = [pool.apply_async(os.getpid, ()) for i in range(4)]
  31.         print([res.get(timeout=1) for res in multiple_results])
  32.  
  33.         # make a single worker sleep for 10 secs
  34.         res = pool.apply_async(time.sleep, (10,))
  35.         try:
  36.             print(res.get(timeout=1))
  37.         except TimeoutError:
  38.             print("We lacked patience and got a multiprocessing.TimeoutError")
  39.  
  40.         print("For the moment, the pool remains available for more work")
  41.  
  42.     # exiting the 'with'-block has stopped the pool
  43.     print("Now the pool is closed and no longer available")

非阻塞 进程池 --- apply_async

注意：是 "进程池"，不是 "线程池"。它可以让你跑满多核CPU，而且使用方法非常简单。

注意要用 apply_async，如果省略了 async，就变成阻塞版本了。

使用 进程池（ 非阻塞 版本）

  1. # -*- coding: utf-8 -*-
  2.  
  3. import time
  4. import multiprocessing
  5.  
  6.  
  7. def worker_func(process_name=None, arg=None):
  8.     for i in range(3):
  9.         print(f"{process_name} ---> {arg}")
  10.         time.sleep(1)
  11.  
  12.  
  13. if __name__ == "__main__":
  14.  
  15.     cpu_count = multiprocessing.cpu_count()
  16.     pool = multiprocessing.Pool(processes=cpu_count)
  17.     for index in range(10):
  18.         data = f"hello {index}"
  19.         pool.apply_async(worker_func, (f"process_name_{index}", data))
  20.     pool.close()
  21.     pool.join()
  22.     print("主线程退出")

示例：

  1. # -*- coding: utf-8 -*-
  2.  
  3. import os
  4. import time
  5. import random
  6. import multiprocessing
  7.  
  8.  
  9. def func_1():
  10.     # os.getpid()获取当前的进程的ID
  11.     print(f"func_1 ---> {os.getpid()}")
  12.     start = time.time()
  13.     time.sleep(random.random() * 10)
  14.     end = time.time()
  15.     print(f'func_1 ---> 运行 {end - start}秒')
  16.  
  17.  
  18. def func_2():
  19.     print(f"func_2 ---> {os.getpid()}")
  20.  
  21.     start = time.time()
  22.     time.sleep(random.random() * 20)
  23.     end = time.time()
  24.     print(f'func_2 ---> 运行 {end - start}秒')
  25.  
  26.  
  27. def func_3():
  28.     print(f"func_3 ---> {os.getpid()}")
  29.     start = time.time()
  30.     time.sleep(random.random() * 30)
  31.     end = time.time()
  32.     print(f'func_3 ---> 运行 {end - start}秒')
  33.  
  34.  
  35. if __name__ == '__main__':
  36.     func_list = [func_1, func_2, func_3]
  37.     print(f"父进程 ---> {os.getpid()}")
  38.  
  39.     pool = multiprocessing.Pool(4)
  40.     for func in func_list:
  41.         # Pool执行函数，当有一个进程执行完毕后，会添加一个新的进程到pool中
  42.         pool.apply_async(func)
  43.  
  44.     print('等待字进程...')
  45.     pool.close()
  46.  
  47.     # 调用join之前，一定要先调用close() 函数，否则会出错
  48.     # close()执行后不会有新的进程加入到 pool, join 函数等待素有子进程结束
  49.     pool.join()
  50.     print('所有进程都完成')
  51.     pass

阻塞 进程池 --- apply

  1. # coding: utf-8
  2. import time
  3. import multiprocessing
  4.  
  5.  
  6. def worker_func(arg):
  7.     print(f"worker_func ---> {arg}")
  8.     time.sleep(3)
  9.  
  10.  
  11. if __name__ == "__main__":
  12.     pool = multiprocessing.Pool(processes=3)
  13.     for i in range(4):
  14.         data = f"hello {i}"
  15.         # 维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去
  16.         pool.apply(worker_func, (data,))
  17.     pool.close()
  18.     pool.join()

map()

函数原型：map(func, iterable[, chunksize=None])

Pool 类中的 map 方法，与内置的 map 函数用法行为基本一致，它会使进程阻塞直到返回结果。
注意，虽然第二个参数是一个迭代器，但在实际使用中，必须在整个队列都就绪后，程序才会运行子进程。

import time
import multiprocessing


def worker_func(arg=None):
    time.sleep(1)
    print('arg * arg')
    return arg * arg


if __name__ == "__main__":
    temp_list = [1, 2, 3, 4, 5, 6,7,8,9]

    start_time = time.time()
    for item in temp_list:
        worker_func(item)
    end_time = time.time()
    print("顺序执行时间：", int(end_time - start_time))

    pool = multiprocessing.Pool(5)  # 创建拥有5个进程数量的进程池
    start_time = time.time()
    result = pool.map(worker_func, temp_list)  # 使进程阻塞直到返回结果
    pool.close()  # 关闭进程池，不再接受新的进程
    pool.join()   # 主进程阻塞等待子进程的退出
    end_time = time.time()
    print("并行执行时间：", int(end_time - start_time))
    print(f'map 的所有子进程返回的结果列表: {result}')

上例是一个创建多个进程并发处理与顺序执行处理同一数据，所用时间的差别。从结果可以看出，并发执行的时间明显比顺序执行要快很多，但是进程是要耗资源的，所以平时工作中，进程数也不能开太大。程序中的 result 表示全部进程执行结束后全部的返回结果集，run 函数有返回值，所以一个进程对应一个返回结果，这个结果存在一个列表中。

对 Pool对象调用 join() 方法会等待所有子进程执行完毕，调用 join() 之前必须先调用 close()，让其不再接受新的 Process。

close()

关闭进程池（pool），使其不在接受新的任务。

terminate()

结束工作进程，不在处理未处理的任务。

join()

主进程阻塞等待子进程的退出，join 方法必须在 close 或 terminate 之后使用。

使用 Pool，并需要关注结果

更多的时候，我们不仅需要多进程执行，还需要关注每个进程的执行结果，如下：

  1. # -*- coding: utf-8 -*-
  2.  
  3. import multiprocessing
  4. import time
  5.  
  6.  
  7. def func(msg):
  8.     for i in range(3):
  9.         print(msg)
  10.         time.sleep(1)
  11.     return "done " + msg
  12.  
  13.  
  14. if __name__ == "__main__":
  15.     pool = multiprocessing.Pool(processes=4)
  16.     result = []
  17.     for index in range(10):
  18.         msg = f"hello {index}"
  19.         result.append(pool.apply_async(func, (msg,)))
  20.     pool.close()
  21.     pool.join()
  22.     for res in result:
  23.         print(res.get())
  24.     print("Sub-process(es) done.")

示例 

  1. import multiprocessing
  2.  
  3.  
  4. def do_calculation(data):
  5.     return data * 2
  6.  
  7.  
  8. def start_process():
  9.     print('Starting', multiprocessing.current_process().name)
  10.  
  11.  
  12. if __name__ == '__main__':
  13.     inputs = list(range(10))
  14.     print('Inputs  :', inputs)
  15.  
  16.     builtin_output = list(map(do_calculation, inputs))
  17.     print('Build-In :', builtin_output)
  18.  
  19.     pool_size = multiprocessing.cpu_count() * 2
  20.     pool = multiprocessing.Pool(processes=pool_size, initializer=start_process, )
  21.     # 默认情况下，Pool会创建固定数目的工作进程，并向这些工作进程传递作业，直到再没有更多作业为止。
  22.     # maxtasksperchild 参数为每个进程执行 task 的最大数目，
  23.     # 设置 maxtasksperchild参数可以告诉池在完成一定数量任务之后重新启动一个工作进程，
  24.     # 来避免运行时间很长的工作进程消耗太多的系统资源。
  25.     # pool = multiprocessing.Pool(processes=pool_size, initializer=start_process, maxtasksperchild=2)
  26.     print('-' * 20)
  27.     pool_outputs = pool.map(do_calculation, inputs)
  28.     pool.close()
  29.     pool.join()
  30.  
  31.     print('Pool  :', pool_outputs)

监听器及客户端

认证密码

日志记录

线程池

multiprocessing.dummy 模块

编程指导

使用 multiprocessing 时，应遵循一些指导原则和习惯用法。

下面这些适用于所有start方法。

避免共享状态

应该尽可能避免在进程间传递大量数据，越少越好。

最好坚持使用队列或者管道进行进程间通信，而不是底层的同步原语。

可序列化

保证所代理的方法的参数是可以序列化的。

代理的线程安全性

不要在多线程中同时使用一个代理对象，除非你用锁保护它。

（而在不同进程中使用 相同 的代理对象却没有问题。）

使用 Join 避免僵尸进程

在 POSIX 上当一个进程结束但没有被合并则它将变成僵尸进程。 这样的进程应该不会很多因为每次启动新进程（或 active_children() 被调用）时所有尚未被合并的已完成进程都将被合并。 而且调用一个已结束进程的 Process.is_alive 也会合并这个进程。 虽然如此但显式地合并你所启动的所有进程仍然是个好习惯。

继承优于序列化、反序列化

当使用 spawn 或者 forkserver 的启动方式时，multiprocessing 中的许多类型都必须是可序列化的，这样子进程才能使用它们。但是通常我们都应该避免使用管道和队列发送共享对象到另外一个进程，而是重新组织代码，对于其他进程创建出来的共享对象，让那些需要访问这些对象的子进程可以直接将这些对象从父进程继承过来。

避免杀死进程

通过 Process.terminate 停止一个进程很容易导致这个进程正在使用的共享资源（如锁、信号量、管道和队列）损坏或者变得不可用，无法在其他进程中继续使用。

所以，最好只对那些从来不使用共享资源的进程调用 Process.terminate 。

Join 使用队列的进程

记住，往队列放入数据的进程会一直等待直到队列中所有项被"feeder" 线程传给底层管道。（子进程可以调用队列的 Queue.cancel_join_thread 方法禁止这种行为）

这意味着，任何使用队列的时候，你都要确保在进程join之前，所有存放到队列中的项将会被其他进程、线程完全消费。否则不能保证这个写过队列的进程可以正常终止。记住非精灵进程会自动 join 。

死锁的例子：交换最后两行可以修复这个问题（或者直接删掉 p.join()）。

  1. from multiprocessing import Process, Queue
  2.  
  3. def f(q):
  4.     q.put('X' * 1000000)
  5.  
  6. if __name__ == '__main__':
  7.     queue = Queue()
  8.     p = Process(target=f, args=(queue,))
  9.     p.start()
  10.     p.join()                    # this deadlocks
  11.     obj = queue.get()

显式传递资源给子进程

在 POSIX 上使用 fork 启动方法，子进程将能够访问使用全局资源在父进程中创建的共享资源。 但是，更好的做法是将对象作为子进程构造器的参数来传入。

除了（部分原因）让代码兼容 Windows 以及其他的进程启动方式外，这种形式还保证了在子进程生命期这个对象是不会被父进程垃圾回收的。如果父进程中的某些对象被垃圾回收会导致资源释放，这就变得很重要。

示例：

  1. from multiprocessing import Process, Lock
  2.  
  3. def f(l):
  4.     ... do something using "l" ...
  5.  
  6. if __name__ == '__main__':
  7.     lock = Lock()
  8.     for i in range(10):
  9.         Process(target=f, args=(lock,)).start()

spawn 和 forkserver 启动方式相对于 fork 启动方式，有一些额外的限制。

更依赖序列化

Process.__init__() 的所有参数都必须可序列化。同样的，当你继承 Process 时，需要保证当调用 Process.start 方法时，实例可以被序列化。

全局变量

记住，如果子进程中的代码尝试访问一个全局变量，它所看到的值（如果有）可能和父进程中执行 Process.start 那一刻的值不一样。

当全局变量只是模块级别的常量时，是不会有问题的。

安全导入主模块

确保新的 Python 解释器可以安全地导入主模块，而不会导致意想不到的副作用（如启动新进程）。

例如，使用 spawn 或 forkserver 启动方式执行下面的模块，会引发 RuntimeError 异常而失败。

  1. from multiprocessing import Process
  2.  
  3. def foo():
  4.     print('hello')
  5.  
  6. p = Process(target=foo)
  7. p.start()

应该通过下面的方法使用 if __name__ == '__main__': ，从而保护程序"入口点":

  1. from multiprocessing import Process, freeze_support, set_start_method
  2.  
  3. def foo():
  4.     print('hello')
  5.  
  6. if __name__ == '__main__':
  7.     freeze_support()
  8.     set_start_method('spawn')
  9.     p = Process(target=foo)
  10.     p.start()

（如果程序将正常运行而不是冻结，则可以省略 freeze_support() 行）

这允许新启动的 Python 解释器安全导入模块然后运行模块中的 foo() 函数。

如果主模块中创建了进程池或者管理器，这个规则也适用。

multiprocessing 示例

示例：创建和使用自定义管理器、代理

  1. from multiprocessing import freeze_support
  2. from multiprocessing.managers import BaseManager, BaseProxy
  3. import operator
  4.  
  5.  
  6. ##
  7.  
  8. class Foo:
  9.     def f(self):
  10.         print('you called Foo.f()')
  11.  
  12.     def g(self):
  13.         print('you called Foo.g()')
  14.  
  15.     def _h(self):
  16.         print('you called Foo._h()')
  17.  
  18.  
  19. # A simple generator function
  20. def baz():
  21.     for i in range(10):
  22.         yield i * i
  23.  
  24.  
  25. # Proxy type for generator objects
  26. class GeneratorProxy(BaseProxy):
  27.     _exposed_ = ['__next__']
  28.  
  29.     def __iter__(self):
  30.         return self
  31.  
  32.     def __next__(self):
  33.         return self._callmethod('__next__')
  34.  
  35.  
  36. # Function to return the operator module
  37. def get_operator_module():
  38.     return operator
  39.  
  40.  
  41. ##
  42.  
  43. class MyManager(BaseManager):
  44.     pass
  45.  
  46.  
  47. # register the Foo class; make `f()` and `g()` accessible via proxy
  48. MyManager.register('Foo1', Foo)
  49.  
  50. # register the Foo class; make `g()` and `_h()` accessible via proxy
  51. MyManager.register('Foo2', Foo, exposed=('g', '_h'))
  52.  
  53. # register the generator function baz; use `GeneratorProxy` to make proxies
  54. MyManager.register('baz', baz, proxytype=GeneratorProxy)
  55.  
  56. # register get_operator_module(); make public functions accessible via proxy
  57. MyManager.register('operator', get_operator_module)
  58.  
  59.  
  60. ##
  61.  
  62. def test():
  63.     manager = MyManager()
  64.     manager.start()
  65.  
  66.     print('-' * 20)
  67.  
  68.     f1 = manager.Foo1()
  69.     f1.f()
  70.     f1.g()
  71.     assert not hasattr(f1, '_h')
  72.     assert sorted(f1._exposed_) == sorted(['f', 'g'])
  73.  
  74.     print('-' * 20)
  75.  
  76.     f2 = manager.Foo2()
  77.     f2.g()
  78.     f2._h()
  79.     assert not hasattr(f2, 'f')
  80.     assert sorted(f2._exposed_) == sorted(['g', '_h'])
  81.  
  82.     print('-' * 20)
  83.  
  84.     it = manager.baz()
  85.     for i in it:
  86.         print('<%d>' % i, end=' ')
  87.     print()
  88.  
  89.     print('-' * 20)
  90.  
  91.     op = manager.operator()
  92.     print('op.add(23, 45) =', op.add(23, 45))
  93.     print('op.pow(2, 94) =', op.pow(2, 94))
  94.     print('op._exposed_ =', op._exposed_)
  95.  
  96.  
  97. ##
  98.  
  99. if __name__ == '__main__':
  100.     freeze_support()
  101.     test()

示例：使用 Pool

  1. import multiprocessing
  2. import time
  3. import random
  4. import sys
  5.  
  6.  
  7. #
  8. # Functions used by test code
  9. #
  10.  
  11. def calculate(func, args):
  12.     result = func(*args)
  13.     return '%s says that %s%s = %s' % (
  14.         multiprocessing.current_process().name,
  15.         func.__name__, args, result
  16.     )
  17.  
  18.  
  19. def calculatestar(args):
  20.     return calculate(*args)
  21.  
  22.  
  23. def mul(a, b):
  24.     time.sleep(0.5 * random.random())
  25.     return a * b
  26.  
  27.  
  28. def plus(a, b):
  29.     time.sleep(0.5 * random.random())
  30.     return a + b
  31.  
  32.  
  33. def f(x):
  34.     return 1.0 / (x - 5.0)
  35.  
  36.  
  37. def pow3(x):
  38.     return x ** 3
  39.  
  40.  
  41. def noop(x):
  42.     pass
  43.  
  44.  
  45. #
  46. # Test code
  47. #
  48.  
  49. def test():
  50.     PROCESSES = 4
  51.     print('Creating pool with %d processes\n' % PROCESSES)
  52.  
  53.     with multiprocessing.Pool(PROCESSES) as pool:
  54.         #
  55.         # Tests
  56.         #
  57.  
  58.         TASKS = [(mul, (i, 7)) for i in range(10)] + \
  59.                 [(plus, (i, 8)) for i in range(10)]
  60.  
  61.         results = [pool.apply_async(calculate, t) for t in TASKS]
  62.         imap_it = pool.imap(calculatestar, TASKS)
  63.         imap_unordered_it = pool.imap_unordered(calculatestar, TASKS)
  64.  
  65.         print('Ordered results using pool.apply_async():')
  66.         for r in results:
  67.             print('\t', r.get())
  68.         print()
  69.  
  70.         print('Ordered results using pool.imap():')
  71.         for x in imap_it:
  72.             print('\t', x)
  73.         print()
  74.  
  75.         print('Unordered results using pool.imap_unordered():')
  76.         for x in imap_unordered_it:
  77.             print('\t', x)
  78.         print()
  79.  
  80.         print('Ordered results using pool.map() --- will block till complete:')
  81.         for x in pool.map(calculatestar, TASKS):
  82.             print('\t', x)
  83.         print()
  84.  
  85.         #
  86.         # Test error handling
  87.         #
  88.  
  89.         print('Testing error handling:')
  90.  
  91.         try:
  92.             print(pool.apply(f, (5,)))
  93.         except ZeroDivisionError:
  94.             print('\tGot ZeroDivisionError as expected from pool.apply()')
  95.         else:
  96.             raise AssertionError('expected ZeroDivisionError')
  97.  
  98.         try:
  99.             print(pool.map(f, list(range(10))))
  100.         except ZeroDivisionError:
  101.             print('\tGot ZeroDivisionError as expected from pool.map()')
  102.         else:
  103.             raise AssertionError('expected ZeroDivisionError')
  104.  
  105.         try:
  106.             print(list(pool.imap(f, list(range(10)))))
  107.         except ZeroDivisionError:
  108.             print('\tGot ZeroDivisionError as expected from list(pool.imap())')
  109.         else:
  110.             raise AssertionError('expected ZeroDivisionError')
  111.  
  112.         it = pool.imap(f, list(range(10)))
  113.         for i in range(10):
  114.             try:
  115.                 x = next(it)
  116.             except ZeroDivisionError:
  117.                 if i == 5:
  118.                     pass
  119.             except StopIteration:
  120.                 break
  121.             else:
  122.                 if i == 5:
  123.                     raise AssertionError('expected ZeroDivisionError')
  124.  
  125.         assert i == 9
  126.         print('\tGot ZeroDivisionError as expected from IMapIterator.next()')
  127.         print()
  128.  
  129.         #
  130.         # Testing timeouts
  131.         #
  132.  
  133.         print('Testing ApplyResult.get() with timeout:', end=' ')
  134.         res = pool.apply_async(calculate, TASKS[0])
  135.         while 1:
  136.             sys.stdout.flush()
  137.             try:
  138.                 sys.stdout.write('\n\t%s' % res.get(0.02))
  139.                 break
  140.             except multiprocessing.TimeoutError:
  141.                 sys.stdout.write('.')
  142.         print()
  143.         print()
  144.  
  145.         print('Testing IMapIterator.next() with timeout:', end=' ')
  146.         it = pool.imap(calculatestar, TASKS)
  147.         while 1:
  148.             sys.stdout.flush()
  149.             try:
  150.                 sys.stdout.write('\n\t%s' % it.next(0.02))
  151.             except StopIteration:
  152.                 break
  153.             except multiprocessing.TimeoutError:
  154.                 sys.stdout.write('.')
  155.         print()
  156.         print()
  157.  
  158.  
  159. if __name__ == '__main__':
  160.     multiprocessing.freeze_support()
  161.     test()

示例：使用队列来向一组工作进程提供任务并收集结果

  1. import time
  2. import random
  3.  
  4. from multiprocessing import Process, Queue, current_process, freeze_support
  5.  
  6. #
  7. # Function run by worker processes
  8. #
  9.  
  10. def worker(input, output):
  11.     for func, args in iter(input.get, 'STOP'):
  12.         result = calculate(func, args)
  13.         output.put(result)
  14.  
  15. #
  16. # Function used to calculate result
  17. #
  18.  
  19. def calculate(func, args):
  20.     result = func(*args)
  21.     return '%s says that %s%s = %s' % \
  22.         (current_process().name, func.__name__, args, result)
  23.  
  24. #
  25. # Functions referenced by tasks
  26. #
  27.  
  28. def mul(a, b):
  29.     time.sleep(0.5*random.random())
  30.     return a * b
  31.  
  32. def plus(a, b):
  33.     time.sleep(0.5*random.random())
  34.     return a + b
  35.  
  36. #
  37. #
  38. #
  39.  
  40. def test():
  41.     NUMBER_OF_PROCESSES = 4
  42.     TASKS1 = [(mul, (i, 7)) for i in range(20)]
  43.     TASKS2 = [(plus, (i, 8)) for i in range(10)]
  44.  
  45.     # Create queues
  46.     task_queue = Queue()
  47.     done_queue = Queue()
  48.  
  49.     # Submit tasks
  50.     for task in TASKS1:
  51.         task_queue.put(task)
  52.  
  53.     # Start worker processes
  54.     for i in range(NUMBER_OF_PROCESSES):
  55.         Process(target=worker, args=(task_queue, done_queue)).start()
  56.  
  57.     # Get and print results
  58.     print('Unordered results:')
  59.     for i in range(len(TASKS1)):
  60.         print('\t', done_queue.get())
  61.  
  62.     # Add more tasks using `put()`
  63.     for task in TASKS2:
  64.         task_queue.put(task)
  65.  
  66.     # Get and print some more results
  67.     for i in range(len(TASKS2)):
  68.         print('\t', done_queue.get())
  69.  
  70.     # Tell child processes to stop
  71.     for i in range(NUMBER_OF_PROCESSES):
  72.         task_queue.put('STOP')
  73.  
  74.  
  75. if __name__ == '__main__':
  76.     freeze_support()
  77.     test()

4、Python 协程

协程 英文名 Coroutine。 关于协程，可以参考 greenlet、stackless、gevent、eventlet 等的实现。

并发编程目前有四种方式：多进程，多线程，异步，和协程。

多线程编程 Python 中有 Thread 和 threading，在 linux 下所谓的线程，实际上是 LWP 轻量级进程，其在内核中具有和进程相同的调度方式，有关 LWP，COW（写时拷贝），fork，vfork，clone等的资料较多，不再赘述。异步在 linux 下主要有三种实现 select，poll，epoll 。

协程的好处：

  • 无需线程上下文切换的开销，协程的切换是由程序自身控制，因此，没有线程切换的开销，和多线程比，线程数量越多，协程的性能优势就越明显。
  • 不需要多线程的锁机制。因为只有一个线程，也不存在同时写变量冲突，
  • 方便切换控制流，简化编程模型
  • 高并发+高扩展性+低成本：一个CPU支持上万的协程都不是问题。所以很适合用于高并发处理。

如何利用多核资源：

  • 协程的本质是个单线程，它不能同时利用 CPU 多核资源。
  • 协程需要和进程配合才能运行在多CPU上。"异步多进程"  可以解决利用多核资源。多进程 + 协程，既充分利用多核，又充分发挥协程的高效率，可获得极高的性能。
  • 进行阻塞（Blocking）操作（如IO时）会阻塞掉整个程序

传统的 "生产者 --- 消费者" 模型是一个线程写消息，一个线程取消息，通过锁机制控制队列和等待，但一不小心就可能死锁。如果改用协程，生产者生产消息后，直接通过 yield 跳转到消费者开始执行，待消费者执行完毕后，切换回生产者继续生产，效率极高：

  1. import time
  2.  
  3. def consumer():
  4.     r = ''
  5.     while True:
  6.         n = yield r
  7.         if not n:
  8.             return
  9.         print('[CONSUMER] Consuming %s...' % n)
  10.         time.sleep(1)
  11.         r = '200 OK'
  12.  
  13. def produce(c):
  14.     c.next()
  15.     n = 0
  16.     while n < 5:
  17.         n = n + 1
  18.         print('[PRODUCER] Producing %s...' % n)
  19.         r = c.send(n)
  20.         print('[PRODUCER] Consumer return: %s' % r)
  21.     c.close()
  22.  
  23. if __name__=='__main__':
  24.     c = consumer()
  25.     produce(c)

执行结果：

[PRODUCER] Producing 1...
[CONSUMER] Consuming 1...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] Producing 2...
[CONSUMER] Consuming 2...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] Producing 3...
[CONSUMER] Consuming 3...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] Producing 4...
[CONSUMER] Consuming 4...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] Producing 5...
[CONSUMER] Consuming 5...
[PRODUCER] Consumer return: 200 OK

注意到consumer函数是一个generator（生成器），把一个consumer传入produce后：
        1. 首先调用c.next()启动生成器；
        2. 然后，一旦生产了东西，通过c.send(n)切换到consumer执行；
        3. consumer通过yield拿到消息，处理，又通过yield把结果传回；
        4. produce拿到consumer处理的结果，继续生产下一条消息；
        5. produce决定不生产了，通过c.close()关闭consumer，整个过程结束。

整个流程无锁，由一个线程执行，produce和consumer协作完成任务，所以称为“协程”，而非线程的抢占式多任务。最后套用Donald Knuth的一句话总结协程的特点：“子程序就是协程的一种特例”

线程、进程 的操作是由程序触发系统接口，最后的执行者是系统；

协程 的操作则是程序员。

协程示例：

import asyncio


async def cor1():
    print("COR1 start")
    await cor2()
    print("COR1 end")


async def cor2():
    print("COR2")


asyncio.run(cor1())

greenlet、Eventlet，gevent ( 推荐 )、

  • greenlet ：是使用生成器实现的协程，调度起来很麻烦，而且不是正在意义上的协程，只是实现的代码执行过程中的挂起，唤醒操作。Greenlet没有自己的调度过程，所以一般不会直接使用。greenlet：http://greenlet.readthedocs.org/en/latest/
  • eventlet：是在 greenlet 的基础上实现了自己的 GreenThread，实际上就是 greenlet 类的扩展封装，而与Greenlet的不同是，Eventlet实现了自己调度器称为Hub，Hub类似于Tornado的IOLoop，是单实例的。在Hub中有一个event loop，根据不同的事件来切换到对应的GreenThread。同时 Eventlet 还实现了一系列的补丁来使 Python 标准库中的 socket 等等module 来支持 GreenThread 的切换。Eventlet 的 Hub 可以被定制来实现自己调度过程。eventlet 目前支持 CPython 2.7 和 3.4+ 并将在未来删除，仅保留 CPython 3.5+ 支持。
  • gevent：基于 libev 与 Greenlet 实现。不同于 Eventlet 的用 python 实现的 hub 调度，Gevent 通过 Cython 调用 libev 来实现一个高效的 event loop 调度循环。同时类似于 Event，Gevent 也有自己的 monkey_patch，在打了补丁后，完全可以使用 python 线程的方式来无感知的使用协程，减少了开发成本。

在 Python 的世界里由于 GIL 的存在，线程一直都不是很好用，所以就有了各种协程的 hack。gevnet 是当前使用起来最方便的协程，但是由于依赖于 libev 所以不能在 pypy上 跑，如果需要在pypy上使用协程，Eventlet 是最好的选择。

gevent 属于第三方模块需要下载安装包
        pip3 install --upgrade pip3
        pip3 install gevent

示例：

  1. import gevent
  2. def fun1():
  3.     print("www.baidu.com")  # 第一步
  4.     gevent.sleep(0)
  5.     print("end the baidu.com")  # 第三步
  6.  
  7. def fun2():
  8.     print("www.zhihu.com")  # 第二步
  9.     gevent.sleep(0)
  10.     print("end th zhihu.com")  # 第四步
  11.  
  12. gevent.joinall([
  13.     gevent.spawn(fun1),
  14.     gevent.spawn(fun2),
  15. ])

示例：遇到 IO 操作自动切换：

  1. import gevent
  2. from gevent import monkey
  3. from datetime import datetime
  4. monkey.patch_all()
  5. import requests
  6.  
  7.  
  8. def func(url):
  9.     print(f"[{datetime.now().replace(microsecond=0)}] 开始请求 {url}")
  10.     gevent.sleep(0)
  11.     proxies = {
  12.         "http": "http://172.17.18.80:8080",
  13.         "https": "http://172.17.18.80:8080",
  14.     }
  15.     proxies = None
  16.     resp = requests.get(url, proxies=proxies)
  17.     print(f"[{datetime.now().replace(microsecond=0)}] {{resp.url}} ---> {len(resp.content)}")
  18.  
  19.  
  20. gevent.joinall([
  21.     gevent.spawn(func, 'https://www.baidu.com/'),
  22.     gevent.spawn(func, 'https://www.sina.com.cn/'),
  23.     gevent.spawn(func, 'https://www.qq.com/'),
  24. ])

多进程 + 协程：

  1. import gevent
  2. from gevent import monkey
  3. monkey.patch_all()
  4.  
  5. import sys
  6. import requests
  7. from multiprocessing import Process
  8. from importlib import reload
  9.  
  10.  
  11. def fetch(url):
  12.     try:
  13.         s = requests.Session()
  14.         r = s.get(url, timeout=1)  # 在这里抓取页面
  15.     except BaseException as be:
  16.         print(be)
  17.     return None
  18.  
  19.  
  20. def setup_process(url_list):
  21.     task_list = []
  22.     for url in url_list:
  23.         task_list.append(gevent.spawn(fetch, url))
  24.     gevent.joinall(task_list)  # 使用协程来执行
  25.  
  26.  
  27. def main(filepath, per_task=100000):  # 每10W条url启动一个进程
  28.  
  29.     with open(filepath, 'r') as f:  # 从给定的文件中读取url
  30.         url_list = f.readlines()
  31.  
  32.     url_list = [temp.strip() for temp in url_list]
  33.     url_len = len(url_list)
  34.     process_count = int(url_len / per_task) if url_len % per_task == 0 else int(url_len / per_task) + 1
  35.     for index in range(process_count):
  36.         task_list = url_list[index * per_task: (index + 1) * per_task]
  37.         p = Process(target=setup_process, args=(url_list,))
  38.         p.start()
  39.  
  40.  
  41. if __name__ == '__main__':
  42.     main('./test_data.txt')  # 读取指定文件

例子中隐藏了一个问题：进程的数量会随着 url 数量的增加而不断增加，我们在这里不使用进程池multiprocessing.Pool 来控制进程数量的原因是 multiprocessing.Pool 和 gevent 有冲突不能同时使用，但是有兴趣的同学可以研究一下 gevent.pool 这个协程池。

  1. """
  2. 对于有些人来说Gevent和multiprocessing组合在一起使用算是个又高大上又奇葩的工作模式.
  3. Python的多线程受制于GIL全局锁的特性，Gevent身为协程也是线程的一种，只是io调度上自己说了算而已。
  4. 那么如何使用多个cpu核心？
  5. 可以利用多进程 mutliprocessing 来进行多核并行工作，
  6. 在多进程里面使用gevent协程框架可以更好的做io调度,相比线程来说减少了无谓的上下文切换.
  7. """
  8.  
  9. import datetime
  10. from multiprocessing import Process, cpu_count, Queue, JoinableQueue
  11. import gevent
  12. from gevent import monkey
  13. monkey.patch_all()
  14.  
  15.  
  16. class Consumer(object):
  17.     def __init__(self, task_queue, task_list, consumer_name):
  18.         self.task_queue = task_queue
  19.         self.task_list = task_list
  20.         self.consumer_name = consumer_name
  21.         self.__run_gevent()
  22.  
  23.     def __run_gevent(self):
  24.         jobs = [gevent.spawn(self.__print_value) for x in range(self.task_list)]
  25.         gevent.joinall(jobs)
  26.  
  27.     def __print_value(self):
  28.         while True:
  29.             value = self.task_queue.get()
  30.             if value is None:
  31.                 self.task_queue.task_done()
  32.                 break
  33.             else:
  34.                 print(f"[{datetime.datetime.now()}] {self.consumer_name} ---> value: {value}")
  35.         return
  36.  
  37.  
  38. class Producer(object):
  39.     def __init__(self, task_queue, task_list, producer_name, consumers_tasks):
  40.         self.task_queue = task_queue
  41.         self.task_list = task_list
  42.         self.producer_name = producer_name
  43.         self.consumer_tasks = consumers_tasks
  44.         self.__run_gevent()
  45.  
  46.     def __run_gevent(self):
  47.         jobs = [gevent.spawn(self.produce) for x in range(self.task_list)]
  48.         gevent.joinall(jobs)
  49.         for x in range(self.consumer_tasks):
  50.             self.task_queue.put_nowait(None)
  51.         self.task_queue.close()
  52.  
  53.     def produce(self):
  54.         for no in range(10000):
  55.             print(no)
  56.             self.task_queue.put(no, block=False)
  57.         return
  58.  
  59.  
  60. def main():
  61.     worker_count = cpu_count() * 2
  62.     task_queue = JoinableQueue()
  63.     producer_gevent = 10
  64.     consumer_gevent = 7
  65.     pid_list = []
  66.     for index in range(worker_count):
  67.         if not index % 2:
  68.             pid = Process(
  69.                 target=Producer,
  70.                 args=(task_queue, 10, f"producer_{index}", consumer_gevent)
  71.             )
  72.             pid.start()
  73.             pid_list.append(pid)
  74.         else:
  75.             pid = Process(
  76.                 target=Consumer,
  77.                 args=(task_queue, consumer_gevent, f"consumer_{index}")
  78.             )
  79.             pid.start()
  80.             pid_list.append(pid)
  81.  
  82.     for pid in pid_list:
  83.         pid.join()
  84.  
  85.  
  86. if __name__ == '__main__':
  87.     main()

5、异步多进程 aiomultiprocess 

使用格式

  1. import asyncio
  2. from aiomultiprocess import Pool
  3.  
  4.  
  5. async def async_worker(data):
  6.     # 异步任务的逻辑
  7.     await do_something_async(data)
  8.  
  9.  
  10. def do_something_async(data):
  11.     # 在此处编写异步任务的逻辑
  12.     ...
  13.  
  14.  
  15. def process_worker(data):
  16.     # 同步任务的逻辑
  17.     do_something(data)
  18.  
  19.  
  20. def do_something(data):
  21.     # 在此处编写同步任务的逻辑
  22.     ...
  23.  
  24.  
  25. async def main():
  26.     # 创建进程池
  27.     pool = await Pool()
  28.  
  29.     # 准备数据
  30.     data_list = [1, 2, 3, 4, 5]
  31.  
  32.     # 异步任务
  33.     async_tasks = [async_worker(data) for data in data_list]
  34.     await asyncio.gather(*async_tasks)
  35.  
  36.     # 多进程同步任务
  37.     await pool.map(process_worker, data_list)
  38.     pool.close()
  39.     await pool.join()
  40.  
  41.  
  42. # 运行主程序
  43. if __name__ == '__main__':
  44.     asyncio.run(main())

示例

  1. import json
  2. import asyncio
  3. import aiomultiprocess
  4. from loguru import logger
  5. from ichrome import AsyncChromeDaemon
  6. from ichrome.async_utils import Chrome
  7.  
  8.  
  9. async def startup_chrome(dp_port=None):
  10.     """
  11.         设置 chrome 参数，然后启动 chrome
  12.     :param dp_port: 自定义 debug port
  13.     :return:
  14.     """
  15.     logger.info(f'dp_port ---> {dp_port}')
  16.     timeout = 5
  17.     # 也可以给 Chrome 添加代理
  18.     proxy = '127.0.0.1:8080'
  19.     udd= f'c:/chrome_user_data_dir_{dp_port}'
  20.     async with AsyncChromeDaemon(port=dp_port, proxy=proxy, user_data_dir=udd) as cd:
  21.         async with cd.connect_tab(index=0) as tab:
  22.             url = 'https://space.bilibili.com/1904149/'
  23.             await tab.set_url(url, timeout=timeout)
  24.             await asyncio.sleep(5)
  25.             cookie = await tab.get_cookies(url, timeout=timeout)
  26.             cookie_string = json.dumps(cookie, ensure_ascii=False)
  27.             logger.info(f'cookie_string ---> {cookie_string}')
  28.  
  29.  
  30. async def main():
  31.     db_list = [9301 + offset for offset in range(5)]
  32.     async with aiomultiprocess.Pool() as aio_pool:
  33.         await aio_pool.map(startup_chrome, db_list)
  34.     await aio_pool.join()
  35.  
  36.  
  37. if __name__ == "__main__":
  38.     asyncio.run(main())
  39.     pass


 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/2023面试高手/article/detail/130803
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号