concurrent.futures

本文参考并转载于: 使用Python进行并发编程-PoolExecutor篇

多线程和多进程模块在使用的时候, start() 方法和 join() 方法不能省, 有时候还需要使用 Queue, 随着需求越来越复杂, 如果没有良好的设计抽象出这部分功能, 代码量会越来越多, debug 的难度也会越来越大。concurrent.futures 模块可以把这些步骤抽象, 这样我们就不需要关注这些细节。

Python 2 需要手动安装

从 Python 3.2 开始, concurrent.futures 模块就被纳入了标准库, 在 Python 2 中, 它属于第三方的 futures 库, 需要手动安装

> pip install futures

模块使用

模块主要包含下面两个类:

ThreadPoolExecutor
ProcessPoolExecutor

也就是对 threading 和 multiprocessing 进行了高级别的抽象, 暴露出统一的接口, 方便开发者使用。

import time
from concurrent.futures import ProcessPoolExecutor, as_completed


NUMBERS = range(30, 40)


def fib(n):
    if n <= 2:
        return 1
    return fib(n-1) + fib(n-2)


if __name__ == '__main__':
    start = time.time()

    with ProcessPoolExecutor(max_workers=3) as executor: # max_workers 是启动进程的个数
        for num, result in zip(NUMBERS, executor.map(fib, NUMBERS)):
            print(f'fib({num}) = {result}')

    print(f'COST: {time.time() - start}')

# 输出:
fib(30) = 832040
fib(31) = 1346269
fib(32) = 2178309
fib(33) = 3524578
fib(34) = 5702887
fib(35) = 9227465
fib(36) = 14930352
fib(37) = 24157817
fib(38) = 39088169
fib(39) = 63245986
COST: 24.688319206237793

除了用 map(), 另外一个常用的方法是 submit(), 如果要提交任务的函数是一样的, 就可以简化成 map(), 但是如果提交的函数是不一样的, 或者执行的过程中可能出现异常, 就要使用到 submit(), 因为使用 map() 在执行过程中如果出现异常会直接抛出错误, 而 submit() 则会分开处理:

from concurrent.futures import ThreadPoolExecutor, as_completed
# 使用了 ThreadPoolExecutor 线程池


NUMBERS = range(30, 35)


def fib(n):
    if n == 34:
        raise Exception('Don\' t do this')
    if n <= 2:
        return 1
    return fib(n-1) + fib(n-2)


if __name__ == '__main__':

    with ThreadPoolExecutor(max_workers=3) as executor:
        future_to_num = {executor.submit(fib, num): num for num in NUMBERS}

    for future in as_completed(future_to_num):
        num = future_to_num[future]
        try:
            result = future.result()
        except Exception as e:
            print(f'raise an exception: {e}')
        else:
            print(f'fib({num}) = {result}')

    with ThreadPoolExecutor(max_workers=3) as executor:
        for num, result in zip(NUMBERS, executor.map(fib, NUMBERS)):
            print(f'fib({num}) = {result}')

# 输出:
raise an exception: Don' t do this
fib(32) = 2178309
fib(31) = 1346269
fib(30) = 832040
fib(33) = 3524578
fib(30) = 832040
fib(31) = 1346269
fib(32) = 2178309
fib(33) = 3524578
Traceback (most recent call last):
  ......
  ......
Exception: Don' t do this

Future

Future 是很常见的一种并发设计的模式, 在很多语言中都有这种解决方案。一个 Future 对象代表了一些尚未就绪(完成)的结果, 在将来的某个时间就绪了之后就可以获取到这个结果。比如前面的例子中, 期望并发执行一些参数不同的 fib() 函数, 获取全部的结果。传统模式就是在等待 Queue.get() 返回结果, 这个是同步模式, 而在 Future 模式下, 调用方式为异步。原先等待返回的时间段，由于 Local worker thread 的存在，这个时候可以完成其他工作。

Pool 和 PoolExecutor

import time
from multiprocessing.pool import Pool


NUMBERS = range(30, 40)


def fib(n):
    if n <= 2:
        return 1
    return fib(n-1) + fib(n-2)


if __name__ == '__main__':
    start = time.time()
    pool = Pool(3)

    for num, result in zip(NUMBERS, pool.map(fib, NUMBERS)):
        print(f'fib({num}) = {result}')

    print(f'COST: {time.time() - start}')

# 输出:
fib(30) = 832040
fib(31) = 1346269
fib(32) = 2178309
fib(33) = 3524578
fib(34) = 5702887
fib(35) = 9227465
fib(36) = 14930352
fib(37) = 24157817
fib(38) = 39088169
fib(39) = 63245986
COST: 23.594091653823853

这个例子代码量更小, 而且花费的时间更短, 其实 concurrent.futures 底层还是用的 threading 和 multiprocessing 这两个模块, 相当于在这上面又封装了一层, 所以速度上会慢一点, 这个是架构和接口实现上的取舍造成的。