线程的替代方案

在开始编写多线程应用之前，先做一个快速回顾：通常来说，多线程是一个好东西。不过，由于Python 的GIL 的限制，多线程更适合于I/O 密集型应用（I/O 释放了GIL，可以允
许更多的并发），而不是计算密集型应用。对于后一种情况而言，为了实现更好的并行性，你
需要使用多进程，以便让CPU 的其他内核来执行。

这里将不再进行详细介绍（这个主题内已经在Core Python Programming 或Core Python
Language Fundamentals 的“执行环境”章节中有所涵盖），对于多线程或多进程而言，threading
模块的主要替代品包括以下几个。

subprocess 模块

这是派生进程的主要替代方案，可以单纯地执行任务，或者通过标准文件（stdin、stdout、
stderr）进行进程间通信。该模块自Python 2.4 版本起引入。

multiprocessing 模块

该模块自Python 2.6 版本起引入，允许为多核或多CPU 派生进程，其接口与threading
模块非常相似。该模块同样也包括在共享任务的进程间传输数据的多种方式。

concurrent.futures 模块

这是一个新的高级库，它只在“任务”级别进行操作，也就是说，你不再需要过分关注
同步和线程/进程的管理了。你只需要指定一个给定了“worker”数量的线程/进程池，提交任
务，然后整理结果。该模块自Python 3.2 版本起引入，不过有一个Python 2.6+可使用的移植
版本，其网址为http://code.google.com/p/pythonfutures。

使用该模块重写后bookrank3.py 会是什么样子呢？假定代码的其他部分保持不变，下面
的代码是新模块的导入以及对_main()函数的修改。

from concurrent.futures import ThreadPoolExecutor
. . .
def _main():
print('At', ctime(), 'on Amazon...')
with ThreadPoolExecutor(3) as executor:
for isbn in ISBNs:
executor.submit(_showRanking, isbn)
print('all DONE at:', ctime())

传递给concurrent.futures.ThreadPoolExecutor 的参数是线程池的大小，在这个应用里就是
指要查阅排名的3 本书。当然，这是个I/O 密集型应用，因此多线程更有用。而对于计算密
集型应用而言，可以使用concurrent.futures.ProcessPoolExecutor 来代替。

当我们得到执行器（无论线程还是进程）之后，它负责调度任务和整理结果，就可以调
用它的submit()方法，来执行之前需要派生线程才能运行的那些操作了。

如果我们做一个到Python 3 的完全移植，方法是将字符串格式化操作符替换为str.format()
方法，自由利用with 语句，并使用执行器的map()方法，那么我们完全可以删除_showRanking()
函数并将其功能混入_main()函数中。示例4-13 的bookrank3CF.py 是该脚本的最终版本。
示例4-13 高级任务管理（bookrank3CF.py）

from concurrent.futures  import ThreadPoolExecutor 
from re import compile
from time import ctime 
from urllib.request  import urlopen as uopen
import urllib.error  
 
# 修正正则表达式 & 添加异常处理
REGEX = compile(br'#([\d,]+)\s+in\s+Books')
AMZN = 'https://www.amazon.com/dp/' 
ISBNs = {
    '0132269937': 'Core Python Programming',
    '0132356139': 'Python Web Development with Django',
    '0137143419': 'Python Fundamentals',
}
 
def getRanking(isbn):
    try:
        # 添加URL路径分隔符处理
        with uopen(f"{AMZN}{isbn}", timeout=10) as page:
            content = page.read() 
            match = REGEX.findall(content) 
            return match[0].decode('utf-8') if match else "Ranking N/A"
    except (urllib.error.URLError,  IndexError, ValueError) as e:
        return f"Error: {str(e)}"
 
def _main():
    print(f'At {ctime()} on Amazon...')
    with ThreadPoolExecutor(max_workers=3) as executor:
        # 修正线程池调用 & 结果映射 
        results = executor.map(getRanking,  ISBNs)
        for isbn, ranking in zip(ISBNs, results):
            print(f'- {ISBNs[isbn]} ranked {ranking}')
    print(f'All DONE at: {ctime()}')
 
if __name__ == '__main__':
    _main()

运行结果

At Fri Aug  1 13:26:34 2025 on Amazon...
- Core Python Programming ranked Error: HTTP Error 500: Internal Server Error
- Python Web Development with Django ranked Error: HTTP Error 500: Internal Server Error
- Python Fundamentals ranked Error: HTTP Error 500: Internal Server Error
All DONE at: Fri Aug  1 13:26:36 2025

相关模块

表4-6 列出了多线程应用编程中可能会使用到的一些模块。