concurrent.futures

Python 在执行时，通常是采用同步的任务处理模式 ( 一个处理完成后才会接下去处理第二个 )，然而 Python 的标准函数“concurrent.futures”，提供了平行任务处理 ( 异步 ) 的功能，能够同时处理多个任务，这篇教学会介绍 concurrent.futures 的用法。

本篇使用的 Python 版本为 3.7.12，所有范例可使用 Google Colab 实作，不用安装任何软件 ( 参考：使用 Google Colab )

同步与异步

同步和异步的常见说法是：“同步模式下，每个任务必须按照顺序执行，后面的任务必须等待前面的任务执行完成，在异步模式下，后面的任务不用等前面的，各自执行各自的任务”，也可以想像成“同一个步道 vs 不同步道”，透过步道的方式，会更容易明白同步和异步。( 因为有时会将同步与异步的中文字面意思，想成“一起走”或“不要一起走”，很容易搞错 )

• 同步：“同一个步道”，只能依序排队前进。
• 异步：“不 ( 非 ) 同步道”，可以各走各的。

Thread 和 Process

concurrent.futures 提供了 ThreadPoolExecutor 和 ProcessPoolExecutor 两种可以平行处理任务的实作方法，ThreadPoolExecutor 是针对 Thread ( 执行绪 )，ProcessPoolExecutor 是针对 Process ( 程序 )，下方是 Thread 和 Process 的简单差异说明：

import concurrent.futures

要使用 concurrent.futures 必须先 import concurrent.futures 模组，或使用 from 的方式，单独 import 特定的类型。

更多信息可以参考 Python 官方文件：concurrent.futures 启动平行任务

import concurrent.futures
from concurrent.futures import ThreadPoolExecutor

ThreadPoolExecutor

ThreadPoolExecutor 会透过 Thread 的方式建立多个 Executors ( 执行器 ) ，执行并处理多个任务 ( tasks )，ThreadPoolExecutor 有四个参数，最常用的为 max_workers：

使用 ThreadPoolExecutor 后，就能使用 Executors 的相关方法：

举例来说，下方的程序码执行后，会按照顺序 ( 同步 ) 显示出数字，前一个任务尚未处理完，就不会执行后续的工作。

import time
def test(n):
    for i in range(n):
        print(i, end=' ')
        time.sleep(0.2)

test(2)
test(3)
test(4)

# 0 1 0 1 2 0 1 2 3

如果改成 ThreadPoolExecutor 的方式，就会发现三个函数就会一起进行 ( 如果执行的函数大于 5，可再设定 max_workers 的数值 )。

import time
from concurrent.futures import ThreadPoolExecutor

def test(n):
    for i in range(n):
        print(i, end=' ')
        time.sleep(0.2)

executor = ThreadPoolExecutor()  # 設定一個執行 Thread 的啟動器

a = executor.submit(test, 2)     # 啟動第一個 test 函式
b = executor.submit(test, 3)     # 啟動第二個 test 函式
c = executor.submit(test, 4)     # 啟動第三個 test 函式
executor.shutdown()              # 關閉啟動器 ( 如果沒有使用，則啟動器會處在鎖住的狀態而無法繼續 )

# 0 0 0 1 1 1 2 2 3

上述的做法，可以改用 with...as 的方式 ( 有点类似 open 的 with )。

import time
from concurrent.futures import ThreadPoolExecutor

def test(n):
    for i in range(n):
        print(i, end=' ')
        time.sleep(0.2)

with ThreadPoolExecutor() as executor:    # 改用 with...as
    executor.submit(test, 2)
    executor.submit(test ,3)
    executor.submit(test, 4)

# 0 0 0 1 1 1 2 2 3

上述的范例，也可以改用 map 的做法：

import time
from concurrent.futures import ThreadPoolExecutor

def test(n):
    for i in range(n):
        print(i, end=' ')
        time.sleep(0.2)

with ThreadPoolExecutor() as executor:
    executor.map(test, [2,3,4])

# 0 0 0 1 1 1 2 2 3

输入文字，停止函数执行

透过平行任务处理的方法，就能轻松做到“输入文字，停止正在执行的函数”，以下方的例子而言，run 是一个具有“无穷循环”的函数，如果不使用平行任务处理，在 run 后方的程序都无法运作 ( 会被无穷循环卡住 )，而 keyin 是一个具有“input”指令的函数，如果不使用平行任务处理，在 keyin 后方的程序也无法运作 ( 会被 input 卡住 )，因此如果使用 concurrent.futures，就能让两个函数同时运行，搭配全域变数的做法，就能在输入特定指令时，停止另外函数的运作。

import time
from concurrent.futures import ThreadPoolExecutor

a = True               # 定義 a 為 True

def run():
    global a           # 定義 a 是全域變數
    while a:           # 如果 a 為 True
        print(123)     # 不斷顯示 123
        time.sleep(1)  # 每隔一秒

def keyin():
    global a           # 定義 a 是全域變數
    if input() == 'a':
        a = False      # 如果輸入的是 a，就讓 a 為 False，停止 run 函式中的迴圈

executor = ThreadPoolExecutor()
e1 = executor.submit(run)
e2 = executor.submit(keyin)
executor.shutdown()

ProcessPoolExecutor

ProcessPoolExecutor 会透过 Process 的方式建立多个 Executors ( 执行器 )，执行并处理多个程序，ProcessPoolExecutor 有四个参数，最常用的为 max_workers：

使用 ProcessPoolExecutor 后，就能使用 Executors 的相关方法：

ProcessPoolExecutor 的用法基本上和 ThreadPoolExecutor 很像，但 ProcessPoolExecutor 主要会用做处理比较需要运算的程序，ThreadPoolExecutor 会使用于等待输入和输出 ( I/O ) 的程序，两者执行后也会有些差别，ProcessPoolExecutor 执行后最后是显示运算结果，而 ThreadPoolExecutor 则是显示过程。

import time
from concurrent.futures import ProcessPoolExecutor

def test(n):
    for i in range(n):
        print(i, end=' ')
        time.sleep(0.2)
    print()

with ProcessPoolExecutor() as executor:
    executor.map(test, [4,5,6])

如果是使用 ThreadPoolExecutor 则会如下图的结果：

此外，Python 3.5 之后 map() 方法多了 chunksize 参数可以使用，该参数只对 ProcessPoolExecutor 有效，可以提升处理大量可迭代对象的执行效能，chunksize 默认 1，数值越大效能越好 ( 以电脑本身 CPU 的效能为主 )。

import time
from concurrent.futures import ProcessPoolExecutor

def test(n):
    for i in range(n):
        print(i, end=' ')
        time.sleep(0.2)
    print()

with ProcessPoolExecutor() as executor:
    executor.map(test, [4,5,6], chunksize=5)  # 設定 chunksize

小结

Python 的 concurrent.futures 内建函数库是一个相当方便的函数库，不仅可以让原本同步的执行变成异步，大幅减少工作时间，用法上也比使用 multiprocessing、threading、asyncio 容易得多，是相当推荐的内建函数库。

意见反馈