线程池与进程池执行器
本课程讨论了线程及线程池执行器的各种API。
线程池与进程池执行器
在前面的部分中，我们学习了如何创建和管理线程与进程。然而，手动管理这些实体对开发者来说可能很繁琐。Python 通过提供一个接口来缓解这一负担，该接口抽象了启动和销毁线程或进程的细节。concurrent.futures 模块提供了 Executor 接口，可用于将任务提交给线程或进程。其两个子类分别是：
ThreadPoolExecutor
ProcessPoolExecutor
任务可以同步或异步地提交到池中执行。
ThreadPoolExecutor
ThreadPoolExecutor 使用线程来执行提交的任务。让我们看一个简单的例子。
使用 ThreadPoolExecutor
1from concurrent.futures import ThreadPoolExecutor 2from threading import current_thread 3 4 5def say_hi(item): 6 7 print("\nhi " + str(item) + " 在线程 ID " + current_thread().name 中执行", flush=True) 8 9 10if __name__ == '__main__': 11 executor = ThreadPoolExecutor(max_workers=10) 12 lst = list() 13 for i in range(1, 10): 14 lst.append(executor.submit(say_hi, "guest" + str(i))) 15 16 for future in lst: 17 future.result() 18 19 executor.shutdown()
我们创建了一个最多包含十个线程的线程池。接下来，在循环中提交任务以供执行。submit() 方法的第一个参数是一个可调用对象，它会使用后续参数被调用。如果你查看输出，会发现任务是由具有不同名称的线程执行的。submit 调用返回我们称为 未来对象（future） 的东西。Future 类代表可调用对象的执行过程。注意，调用 future.result() 是阻塞的。 有趣的是，如果我们把第一个 for 循环中的代码改为如下形式，执行就会变成串行的。
1 for i in range(1, 10): 2 future = executor.submit(say_hi, "guest" + str(i)) 3 future.result()
通常我们不会采用这种任务提交方式。我们的目标是能够并行处理多个任务，并通过未来对象查询任务的进度。例如，如果你的程序需要从互联网下载多个 MP3 文件，那么这些下载任务就可以相互独立地并行执行。又或者你编写一个脚本，将多个文件从源目录复制到目标目录，这类任务也适合并行处理。
程序末尾，我们调用了执行器的 shutdown() 方法。默认情况下，执行器会等待所有未来对象完成，并释放相关资源。在调用 shutdown 后再提交任务将引发异常。如果调用时传入 wait=False，则调用变为非阻塞。但整个 Python 程序不会退出，直到所有未来对象都完成为止。
1from concurrent.futures import ThreadPoolExecutor 2from threading import current_thread 3 4 5def say_hi(item): 6 7 print("\nhi " + str(item) + " 在线程 ID " + current_thread().name 中执行", flush=True) 8 9 10if __name__ == '__main__': 11 executor = ThreadPoolExecutor(max_workers=10) 12 lst = list() 13 for i in range(1, 10): 14 lst.append(executor.submit(say_hi, "guest" + str(i))) 15 16 for future in lst: 17 future.result() 18 19 executor.shutdown()
map 方法
map() 方法返回一个迭代器，其中包含将某个函数应用于一组值的结果。函数和值作为参数传递给 map() 调用。结果按输入值的顺序返回。考虑以下代码片段：
map() 示例
1 executor = ThreadPoolExecutor(max_workers=10) 2 3 it = executor.map(square, (1, 2, 3, 4, 5, 6, 7, 8, 9, 10), 4 chunksize=1, timeout=2)
在示例中，square() 是一个可调用对象。内部，迭代器的 __next__() 方法会调用未来对象的 result()，这是一个阻塞调用。如果某个未来对象的结果尚未准备好，迭代就会被阻塞。为避免这种情况，我们可以在 map 调用中指定超时时间；如果在超时秒数内结果仍未就绪，则会抛出 Timeout 异常。我们可以遍历已完成的未来对象，但一旦遇到尚未完成计算的首个未来对象，迭代就会被阻塞。
在下面的例子中，我们通过让第五个计算耗时十秒来模拟超时异常，而 map 调用中的超时设置为两秒。运行此代码后，你会看到前四个结果被打印出来，随后抛出超时异常。
另外请注意，如果不指定 map 调用中的 timeout 参数，迭代将在第五次尝试时被阻塞，直到第五个未来对象完成。
1from concurrent.futures import ThreadPoolExecutor 2from threading import current_thread 3import time 4 5 6def square(item): 7 if item == 5: 8 time.sleep(10) 9 return item * item 10 11 12if __name__ == '__main__': 13 executor = ThreadPoolExecutor(max_workers=10) 14 15 it = executor.map(square, (1, 2, 3, 4, 5, 6, 7, 8, 9, 10), 16 chunksize=1, timeout=2) 17 18 for sq in it: 19 print(sq) 20 21 executor.shutdown()
在上述两个示例中，尽管主线程在收到超时异常后退出，但整个 Python 程序不会退出，直到执行第五个计算的线程也结束。如果该线程是守护线程（daemon thread），程序就可以在不等待线程完成的情况下退出。最后，请注意，当使用线程池时，chunksize 参数不会产生任何影响。
ProcessPoolExecutor
进程池与线程池非常相似，唯一的区别是它是一个由进程组成的池，用于执行任务，而不是线程。下面我们重写之前的一个示例，改用进程池而非线程池。
1from concurrent.futures import ProcessPoolExecutor 2import multiprocessing 3from multiprocessing import current_process 4from threading import current_thread 5 6import os 7 8 9def say_hi(item): 10 print( 11 "\nhi " + str(item) + " 在线程 ID " + current_thread().name + " 中，进程 ID " + str(os.getpid()) + "，进程名 " + current_process().name, 12 flush=True) 13 14 15if __name__ == '__main__': 16 print("主进程 ID " + str(os.getpid())) 17 multiprocessing.set_start_method('spawn') 18 executor = ProcessPoolExecutor(max_workers=10) 19 lst = list() 20 for i in range(1, 10): 21 lst.append(executor.submit(say_hi, "guest" + str(i))) 22 23 for future in lst: 24 future.result() 25 26 executor.shutdown()
1from concurrent.futures import ProcessPoolExecutor 2import multiprocessing 3from multiprocessing import current_process 4from threading import current_thread 5 6import os 7 8 9def say_hi(item): 10 print( 11 "\nhi " + str(item) + " 在线程 ID " + current_thread().name + " 中，进程 ID " + str(os.getpid()) + "，进程名 " + current_process().name, 12 flush=True) 13 14 15if __name__ == '__main__': 16 print("主进程 ID " + str(os.getpid())) 17 multiprocessing.set_start_method('spawn') 18 executor = ProcessPoolExecutor(max_workers=10) 19 lst = list() 20 for i in range(1, 10): 21 lst.append(executor.submit(say_hi, "guest" + str(i))) 22 23 for future in lst: 24 future.result() 25 26 executor.shutdown()
请注意，我们可以通过适当设置启动方法，使进程池以“派生”（spawn）或“复制”（fork）的方式创建进程。如果将上面示例中第 18 行的启动方法改为 fork，输出将显示进程是通过复制方式创建的。
我们也可以在进程池中使用 map() API。下面是使用进程池重写的线程池示例。
1from concurrent.futures import ProcessPoolExecutor 2import os 3import time 4 5 6def square(item): 7 print("在进程 ID " + str(os.getpid()) + " 中执行", flush=True) 8 return item * item 9 10 11if __name__ == '__main__': 12 executor = ProcessPoolExecutor(max_workers=10) 13 14 it = executor.map(square, (1, 2, 3, 4, 5, 6, 7, 8, 9, 10), chunksize=1) 15 16 for sq in it: 17 print(sq) 18 19 executor.shutdown()
在使用 map() 时，线程与进程之间唯一的重大区别在于 chunksize 参数的影响。在上面的例子中，我们将 chunksize 设置为 1，这意味着每个平方运算由不同的进程完成。如果将 chunksize 改为 5，则只需两个进程即可完成十个输入值的平方计算。根据具体使用场景，较大的 chunksize 可能带来更快的执行速度，因为减少了频繁创建和销毁进程所耗费的时间。