python常用库

pathlib路径操作

用于面向对象地操作路径，通常使用 pathlib 库中的 Path 类：from pathlib import Path

构造路径对象

• Path(r"/xxxx/b.txt")构造路径对象，路径建议写为原始字符串 r"xxx" 防止各种斜杆被转义
• Path(r"a") / "b" / "c.txt"使用/运算符拼接路径，自动处理不同平台的路径分隔符
• Path.cwd()获取当前工作目录的 Path 对象
• Path.home()获取当前用户主目录的 Path 对象

路径与文件名

• .parent / .name / .stem / .suffix获取父目录、完整文件名、无后缀文件名、后缀。
• .with_name("new.txt") / .with_stem("new") / .with_suffix(".txt")修改文件名(含后缀) / 不含后缀的文件名 / 后缀
• .resolve()解析为绝对路径(基于当前工作目录)，会解析路径中存在的.、.. 和符号链接 并替换为对应路径。如 Path('../exist.txt').resolve()，文件不存在时报错，可以使用.resolve(strict=False)关闭路径/文件存在检查
• .absolute()转换为绝对路径，但不解析.、.. 和符号链接
• .relative_to(other)返回相对于 other 的相对路径
• .is_absolute()判断是否为绝对路径

文件/目录操作

文件/目录创建、删除、重命名

• .touch()创建空文件，或更新文件时间戳
• .unlink(missing_ok=True)删除文件或符号链接，missing_ok=True 文件不存在不报错
• .mkdir(parents=True, exist_ok=True)创建目录，parents=True 递归创建父目录，exist_ok=True 已存在不报错
• .rmdir()删除空目录，删除非空目录参考 shutil.rmtree(p)
• .rename(new_path) / .replace(new_path)重命名或移动路径；replace() 在目标存在时会覆盖。
  文件查找
• .exists() / .is_file() / .is_dir()判断路径是否存在、是否为文件、是否为目录。
• .iterdir()遍历目录下所有直接子项，返回 Path 对象迭代器
• .glob("*.txt")按通配模式查找当前目录下的文件，返回生成器，.rglob("*.txt")递归匹配所有子目录下的文件

读写文件

• .read_text(encoding="utf-8") / .write_text("text", encoding="utf-8")直接读取/写入文本文件，适合小文件。
• .read_bytes() / .write_bytes(data)直接读取/写入二进制数据

其他

.stat()、.chmod(mode)、symlink_to()、hardlink_to()、stat()、owner()、group()、权限相关操作等系统属性功能。

shutil高层文件操作

shutil（shell utilities）是高层文件操作库，可以删除非空目录、跨磁盘复制文件等

文件/目录复制

• shutil.copy(src, dst)复制文件（含权限），不含元数据，如：shutil.copy("a.txt","ac.txt")
• shutil.copy2(src, dst)复制文件，同时保留元数据（修改时间等）
• shutil.copytree(src, dst)递归复制整个目录树
• shutil.copyfile(src, dst) / .copymode(src, dst) / copystat(src, dst)仅复制文件内容 / 权限 / 状态信息

移动与删除

• shutil.move(src, dst)移动文件或目录（可跨磁盘）
• shutil.rmtree(p)递归删除整个目录树（含非空目录）

磁盘与归档

• shutil.disk_usage(p)获取磁盘使用情况，返回 (total, used, free)
• shutil.which("python")查找可执行文件路径，类似 which 命令
• shutil.make_archive("输出文件名", "压缩格式", "root_dir","target_dir")进入root_dir，将该目录下的target_dir目录打包为 zip 等压缩文件，如果只给root_dir不给target_dir，则打包整个root_dir
• shutil.unpack_archive(filename, extract_dir)解压归档文件

sys

sys 是 Python 标准库，提供与解释器和运行环境交互的功能

sys.exit()

• sys.exit(arg)退出程序。arg=0/None 表示正常退出；非 0 表示异常；也可传入一个字符串，如：sys.exit("发生错误")，此时会向标准错误输出该信息并以非 0 状态退出，本质上是在抛出 SystemExit 异常，即类似于执行 raise SystemExit(arg)
• python的 site 模块也提供了exit()方法，这是为 Python 交互式解释器（REPL） 准备的便捷退出方式，在某些不加载 site 模块的环境中可能不存在(如：通过 python -S 禁用 site 模块时)，因此不应该用在正式代码程序中

部分属性

• sys.argv命令行参数列表，argv[0] 是脚本名
• sys.stdin标准输入流，如从标准输入读取：for line in sys.stdin: print(line.strip())
• sys.stdout标准输出流
• sys.stderr标准错误流
• sys.versionPython 版本字符串，可用sys.version_info返回版本元组，可比较
• sys.platform操作系统标识（'linux'、'win32'、'darwin'）
• sys.prefixPython 安装目录

部分方法

• sys.getsizeof(obj)返回对象占用的字节数
• sys.getrefcount(obj)返回对象的引用计数
• sys.intern(s)将字符串加入驻留池，加速比较

OS

os 是 Python 标准库中与操作系统交互的底层模块，主要处理路径、目录、文件、环境变量、进程等，处理文件/目录应该优选 pathlib 库

操作系统信息

• os.name操作系统类型，"nt" 为 Windows，"posix" 为类 Unix
• os.system("cmd")执行简单的命令，如：os.system("cls")清空控制台，执行复杂命令使用subprocess
• os.cpu_count()获取 CPU 核心数

环境变量

• os.environ.get("KEY")读取环境变量；不存在时返回 None，也可使用 os.environ["KEY"] 强制读取
• os.environ["KEY"] = "value"设置当前进程的环境变量；子进程可以继承

目录操作

• os.getcwd()获取当前工作目录
• os.chdir(path)切换当前工作目录
• os.listdir(path)返回目录下的文件名列表；简单场景可用，路径处理通常优先 Path.iterdir()
• os.walk(path)复杂目录遍历时使用，返回 (当前目录, 子目录, 文件)；

subprocess执行外部命令

subprocess 用于创建和管理子进程，替代旧的 os.system 等

subprocess.run

• 执行简单的命令：result = subprocess.run(['ls', '-l'])，或是 shell 内置命令：subprocess.run("cls",shell=True)

• 完整API：

  result = subprocess.run(["cmd", "arg1"], * , stdin=None, input=None, stdout=None, stderr=None, capture_output=False, shell=False, cwd=None, timeout=None, check=False, encoding=None, errors=None, text=None, universal_newlines=None...)

• 执行外部程序并等待结束，执行后状态码可通过 result.returncode 查看

• 单个命令无参数不需要[]，带参数的命令通过列表传参：["命令", "参数1", "参数2"....]

• 默认情况下，命令的输入输出继承父进程，输出通常默认输出到stdout（默认为终端），变量result不捕获输出，无法通过result.stdout/result.stderr 访问这些输出

• capture_output=True捕获标准输出和标准错误输出，而不再输出到终端，之后可以通过 result.stdout / result.stderr 处理这些输出，在使用该选项时建议添加 text = True 来让输出打印为文本形式而非十六进制 bytes

• stdin, stdout, stderr, input 分别指定被执行程序的标准输入、标准输出和标准错误文件句柄，可以为None，当它们的值为 PIPE 时，等价于 capture_output=True

• shell = True：通过 shell 执行命令，可以广泛使用管道、重定向等，但应该注意命令注入风险，不应该拼接任何用户输入，如：subprocess.run('ls -l | grep .py', shell=True)

• text=True 以文本形式返回，而非 bytes

• cwd 指定命令执行的工作目录

• timeout=10：限制等待时间，单位为s，超时后杀死进程并抛出 TimeoutExpired。

• check=True：外部程序返回非 0 退出码时自动抛出 CalledProcessError

• encoding, errors, text指定编码或输出方式

• env 自定义环境变量字典

• result.stdout / result.stderr / result.returncode：分别获取标准输出、标准错误、退出码（0 表示成功，非 0 表示失败）stdout和stderr只有设置了 capture_output=True 时才有效

# 捕获输出
result = subprocess.run(
    ['echo', 'hello'],
    capture_output=True,
    text=True
)
print(result.stdout)      # 'hello\n'
print(result.stderr)      # ''
print(result.returncode)  # 0

# 检查错误
try:
    result = subprocess.run(['ls', '/nonexist'], check=True, capture_output=True, text=True)
except subprocess.CalledProcessError as e:
    print(f"返回码: {e.returncode}")
    print(f"错误: {e.stderr}")

# 超时处理
try:
    subprocess.run(['sleep', '10'], timeout=3, check=True)
except subprocess.TimeoutExpired:
    print("超时！")

# 处理输出
r = subprocess.run(['ls'], capture_output=True, text=True)
if r.returncode == 0:
    lines = r.stdout.splitlines()
    print(f"共 {len(lines)} 个文件")
else:
    print(f"出错：{r.stderr}")

subprocess.Popen

• subprocess.Popen(...)：用于创建并管理一个正在运行的子进程；适合长时间运行、实时读取输出、需要与进程交互的场景，具体参考文档
• Popen(...).poll() / .wait()：检查进程是否结束 / 等待进程结束

argparse解析命令行参数

argparse 是 Python 标准库中用于解析命令行参数的模块

• 创建解析器：parser = argparse.ArgumentParser(
  prog='myprogram', # 程序名
  description='程序描述', # 帮助信息前的描述
  epilog='结尾说明', # 帮助信息后的说明
  )

• 添加参数：parser.add_argument(
  'name', # 参数名（位置参数或选项名）
  action='store', # 动作类型
  nargs=None, # 参数个数
  const=None, # 与 action='store_const' 配合
  default=None, # 默认值
  type=str, # 指定转换类型
  choices=None, # 可选值列表
  required=False, # 是否必须（仅可选参数）
  help='帮助信息', # 帮助文本
  metavar='NAME', # 帮助中显示的参数名
  dest='target', # 存储属性名（默认由参数名推导）
  )

import argparse
def main():
    parser = argparse.ArgumentParser(
        prog='demo',
        description='示例程序',
        formatter_class=argparse.ArgumentDefaultsHelpFormatter
    )
    # 位置参数
    parser.add_argument('input', help='输入文件')
    # 可选参数
    parser.add_argument('-o', '--output', default='out.txt', help='输出文件')
    parser.add_argument('-n', '--num', type=int, default=10, help='数量')
    parser.add_argument('-v', '--verbose', action='store_true', help='详细输出')
    parser.add_argument('--tags', nargs='*', default=[], help='标签列表')
    parser.add_argument('--mode', choices=['fast', 'slow'], default='fast')
    # 互斥组
    group = parser.add_mutually_exclusive_group()
    group.add_argument('--quiet', action='store_true')
    group.add_argument('--debug', action='store_true')
    args = parser.parse_args()
    print(args)
if __name__ == '__main__':
    main()

logging日志处理

• logging.basicConfig(level=logging.INFO, format="...")：配置基础日志系统；简单脚本通常在程序开头配置一次。
• logging.debug() / info() / warning() / error() / critical()：按严重程度记录日志。
• logger = logging.getLogger(__name__)：获取当前模块的 logger；多模块项目通常使用这种方式。
• logger.info("value=%s", value)：使用参数化日志，而不是提前拼接字符串

re正则表达式库

类方法

• re.match(pattern, string)从开头匹配，成功返回Match对象，失败返回 None
• re.search(pattern, string)扫描整个字符串，返回第一个匹配
• re.findall(pattern, string)返回所有匹配的列表
• re.finditer(pattern, string)返回所有匹配的迭代器（元素为Match对象）
• re.sub(pattern, repl, string)替换匹配内容，返回新字符串
• re.subn(pattern, repl, string)同 sub，但返回 (新字符串, 替换次数)
• re.split(pattern, string)按模式分割字符串，返回列表
• re.compile(pattern)编译正则，返回 Pattern 对象（可复用）
• re.fullmatch(pattern, string)整个字符串完全匹配才返回Match对象
• re.escape(string)转义字符串中的特殊字符

Match对象常用方法

上述库方法中很多方法返回的是 Match 对象，如：m = re.search(r'(\d+)-(\d+)', 'abc 12-34 def')，它们有以下常用方法：

• m.group()返回整个匹配，如：'12-34'
• m.group(1)如果正则中进行了分组，返回第1个分组，如：'12'
• m.groups()所有分组元组，如：('12','34')
• m.groupdict()命名分组的字典
• m.start()匹配起始位置，如：4
• m.end()匹配结束位置，如：9
• m.span()起止位置元组，如：(4, 9)

Pattern 对象方法

类似于Java，Python支持编译(compile)正则，方便多次复用，提升效率：

re.compile(pattern)编译正则，返回 Pattern 对象，该对象可以调用以下方法，其中 pos / endpos 是切片范围：

• p.match(string[, pos[, endpos]])从 pos 位置开头匹配，返回Match对象或 None
• p.search(string[, pos[, endpos]])扫描整个字符串，返回第一个匹配，返回Match 或 None
• p.fullmatch(string[, pos[, endpos]])整个字符串完全匹配，返回Match 或 None
• p.findall(string[, pos[, endpos]])返回所有匹配，返回列表
• p.finditer(string[, pos[, endpos]])返回所有匹配，返回迭代器（Match）
• p.sub(repl, string, count=0)替换，返回新字符串
• p.subn(repl, string, count=0)替换并计数，返回(新字符串, 次数)
• p.split(string, maxsplit=0)按模式分割，返回列表
  Pattern对象的属性
• p.pattern编译时用的正则字符串
• p.flags编译时的标志（整数）
• p.groups捕获分组数量
• p.groupindex命名分组 → 组号的字典

p = re.compile(r'(?P\d{4})-(?P\d{2})')
p.pattern      # '(?P\\d{4})-(?P\\d{2})'
p.groups       # 2
p.groupindex   # {'year': 1, 'month': 2}

datetime

datetime：面向对象的日期时间表示与运算，支持时区、格式化、算术运算，该库创建的日期时间对象是不可变的，推荐导入方式：

• 常用模块：from datetime import datetime, timedelta, timezone
• 完整导入：from datetime import datetime, date, time, timedelta, timezone

创建datetime对象

• datetime.now()获取当前时间
• datetime.now(timezone.utc)当前 UTC 时间
• date.today()获得当前日期
  手动指定时间
• datetime(年, 月, 日, 时, 分, 秒)创建一个指定日期的datetime对象，至少需要年月日参数，可以在末尾额外添加一个参数 microsecond
• datetime.fromisoformat("2024-05-20T14:30:59")从 ISO 格式的字符串中获得时间信息
• datetime.fromtimestamp(ts)将一个时间戳 ts (float类型) 转为 datetime
• datetime.strptime("2024-05-20 14:30:59", "%Y-%m-%d %H:%M:%S") 从字符串获得时间信息，第二个参数需要指定解析格式，返回一个 datetime 对象
• date(年, 月, 日)：创建一个datetime.date对象
• time(时, 分, 秒)：创建一个datetime.time对象，不常用，更常用 time 库

获取单个字段

以下方法是实例对象身上的方法，假设：dt = datetime.now()

• dt.year / .month / .day / .hour / .minute / .second / .microsecond 获取对应字段
• dt.weekday() 返回一个值，周一=0，周日=6
• dt.isoweekday() 返回一个值，周一=1，周日=7
  时间戳：返回的是 float 类型的秒数
• dt.timestamp() 将 datetime 转为 Unix 时间戳

格式化

输出一个字符串

• dt.isoformat()转为 ISO 格式，常用于程序之间交换日期时间，日志、文件存储日期信息
• dt.strftime("%Y-%m-%d %H:%M:%S") 按指定格式输出日期，日期格式参考Linux命令的date

timedelta

timedelta用于进行时间与日期的算术运算，处理两个日期或时间之间的差值，支持与 date / datetime / time 进行加减运算，支持自身之间的加减乘除与比较

构造方法

• timedelta(days=0, seconds=0, microseconds=0, milliseconds=0, minutes=0, hours=0, weeks=0) 所有参数可负数，它们在内部统一存储为 days、seconds、microseconds 三个字段
  timedelta(hours=2, minutes=30) 返回 datetime.timedelta(seconds=9000)，存储为秒
• 两个日期对象的运算也会生成一个 timedelta 对象：

  • datetime - datetime：dt = datetime.now() - datetime(2020,10,1)
  • date - date：如 date(2024, 5, 21) - date(2024, 5, 20)

对象属性

• timedelta内部只存储.days、.seconds、.microseconds三个字段，如果要计算两个日期之间差的小时数，需要手动通过days或total_seconds()属性换算
• td.days 返回差额的天数，可为负
• td.seconds 返回差额移除天以后的剩余秒数，计算的是"不足一天"的剩余秒数，要获取总秒数请用 total_seconds()
• td.microseconds返回差额中的微秒数
• td.total_seconds()返回总秒数（含小数）
• str(td) 输出为字符串表示，如：'1 day, 2:30:00'

支持的算术运算

• 两个 timedelta 对象支持 +-*/、abs()等数学运算，还支持==、<=、>等比较运算，如：timedelta(hours=6) * 2；timedelta(days=1) - timedelta(hours=6)；timedelta(days=1) == timedelta(hours=24) # True
• timedelta与日期对象运算可以生成新的时间，

  • datetime ± timedelta：datetime.now() + timedelta(days=1) 当前时间延后一天的日期
  • date ± timedelta：date.today() + timedelta(days=1)

time

time库提供底层时间访问，关于日期和时间功能和datetime库重合，重点使用time提供的时间戳、睡眠与性能计时功能。

为了防止time库和datetime库中的time类在导入时相互覆盖，推荐导入方式：

• time.time()返回当前 Unix 时间戳，类型为 float
• time.time_ns()当前纳秒级时间戳（整数）
• time.sleep(secs)让程序休眠，指定一个秒数，float类型
• time.perf_counter()高精度计时器，可用于测量程序耗时
• time.process_time()计算程序占用 CPU 时间(不含睡眠时间)

collections

collections 是 Python 标准库中提供专用容器数据类型的模块，用于补充 list、dict、set 等内置容器

Counter计数器

Counter 用于统计可迭代对象中各元素出现的次数，本质上是 dict 的子类

• 构造：Counter(iterable) 返回一个字典的子类，可以使用字典的大部分方法访问结果，如：c = Counter("hello")
• c['元素']获得该元素出现的次数，不存在时返回0
• c.update(iterable)增加计数
• c.subtract(iterable)减少计数
• c.most_common(n)出现次数最多的 n 个元素，如：c.most_common(2) # [('l', 2), ('h', 1)]
• c.total()所有计数之和
• 计数器之间支持 +-&| 运算，& 取各元素取较小计数，|各元素取较大计数

defaultdict带默认值的字典

defaultdict 在普通 dict 基础上增加了缺少键时自动创建默认值的能力

• defaultdict构造时可以传入一个默认值工厂函数，在不给键时，会根据对应工厂函数创建默认值
  defaultdict(list) # 缺少键 → [] defaultdict(int) # 缺少键 → 0 defaultdict(set) # 缺少键 → set()
• 操作方法和普通字典类型，注意：访问不存在的键会创建该键，如：d["new"] # 会产生 "new": []

deque双端队列

deque（double-ended queue）适合需要从两端频繁插入、删除的场景，特别适合固定大小的历史记录、滑动窗口、队列、栈，对比list：

• list：尾部 append/pop 高效，头部 insert/pop(0) 较慢

• deque：两端 append/pop 都高效

• 构造：deque([iterable[, maxlen]])如果不传入数据将构造一个空的deque，可以指定maxlen来指定队列最大长度，当加入新元素时会根据加入方向挤走另一端的数据

• q.append(x)右端添加

• q.appendleft(x)左端添加

• q.pop()右端删除

• q.popleft()左端删除

• q.extend(iterable)右端批量添加

• q.extendleft(iterable)左端批量添加

• q.rotate(1)向右循环移动 n 步，如果 n 是负数，就向左循环，n默认值为1

• q.clear()清空

• q.count(元素)队列中指定元素的个数

• q.maxlenDeque的最大尺寸，如果没有限定的话就是 None

其他

collections还包含其他诸多如：namedtuple(带字段名的元组)、ChainMap(组合多个映射)等数据类型，参考文档：collections --- 容器数据类型

urllib.parse操作URL

urllib.parse 用于处理 URL 的解析、拼接、编码以及查询参数

urlsplit/urlparse解析URL

urlsplit(url)：将 URL 拆分为scheme://netloc/path?query#fragment形式，返回 ParseResult 命名元组

• r.scheme协议/方案，如： 'https'
• r.netloc网络位置，完整形式为 user:pass@www.example.com:8080，包含用户名、密码、主机、端口号

  • r.hostname主机，如：'example.com'
  • r.port端口号，如：8080
  • r.username用户名
  • r.password密码

• r.path请求路径，如：'/path'
• r.query查询字符串，如：'a=1'
• r.fragment片段/锚点，如：'top'

urlparse(url)比 urlsplit 多拆分出一个params字段，绝大多数现代网站不使用 params，这个字段通常是空字符串''：

urlunsplit/urlunparse拼接出URL

urlunsplit(parts)将 urlsplit() 得到的组件重新组合成 URL，urlparse()对应的拼接方式为urlunparse()

处理相对链接

urljoin(base, url)根据基础 URL 解析相对 URL，注意：第二个参数如果本身是绝对 URL，会直接使用它

parse_qs/parse_qsl解析查询字符串为字典

• parse_qs(query)将URL中的查询字符串部分解析为字典，由于URL中一个 key 可能出现多次，因此字典的值为一个列表
• parse_qsl(query)将URL中的插叙字符串部分解析为列表，列表值为元组，出现多次的 key 将被解析为两个元组

urlencode(query)生成查询字符串

urlencode(query)将字典或键值对转换为 URL 查询字符串

URL 编码与解码

编码

• quote(string)对 URL 中的字符串进行百分号编码，空格编码为 %20，如：quote("Python 教程") # 'Python%20%E6%95%99%E7%A8%8B'
• quote_plus(string)与 quote() 类似，但空格编码为 +，如：quote_plus("Python 教程") # 'Python+%E6%95%99%E7%A8%8B'
  解码
• unquote(string)对百分号编码进行解码，如：unquote("Python%20%E6%95%99%E7%A8%8B") # 'Python 教程'
• unquote_plus(string)解码百分号编码，同时将 + 解码为空格，如：unquote_plus("Python+%E6%95%99%E7%A8%8B") # 'Python 教程'

tempfile临时文件/目录

tempfile 用于创建临时文件和目录，程序结束后可自动或手动清理，适合缓存、中间数据、测试等场景，该库中的方法通常直接搭配with使用

TemporaryFile()匿名临时文件

创建一个临时文件对象，使用完毕后自动删除，适合只需要一个临时文件对象，不需要知道临时文件路径的场景

• mode：打开模式，默认 "w+b"
• encoding：文本模式下指定编码
• suffix：文件名后缀
• prefix：文件名前缀
• dir：指定临时文件所在目录

NamedTemporaryFile()有文件名的临时文件

创建临时文件，并提供实际的文件路径，可被其他进程/程序访问

• mode：打开模式，默认 "w+b"
• suffix：文件名后缀
• prefix：文件名前缀
• dir：指定临时文件所在目录
• delete：关闭时是否删除，默认 True，注意：Windows 下 delete=True 时文件在关闭前无法被其他进程再次打开

TemporaryDirectory()临时目录

创建一个临时目录，退出 with 后自动删除整个目录及其内容

• suffix：文件名后缀
• prefix：文件名前缀
• dir：指定临时文件所在目录
• tmpdir.name：目录名

concurrent.futures并发编程

concurrent.futures是 Python 标准库中的并发编程模块，它将OS中的线程（threading）和进程（multiprocessing）封装起来，通过统一的Executor（执行器）模型提交任务，并通过Future（未来对象） 获取异步执行结果，让开发者以统一的方式实现并行任务处理，适合处理：

• 大量 I/O 操作阻塞问题（网络请求、文件读写、爬虫）
• CPU 密集型任务并行计算
• 批量任务调度
• 异步获取执行结果

核心特点

• 提供 ThreadPoolExecutor（线程池执行器）和 ProcessPoolExecutor（进程池执行器）:

  • ThreadPoolExecutor：适合I/O 密集型任务（网络请求、文件读写、数据库操作等）。这类任务大部分时间处于 I/O 等待状态，CPU 实际占用时间较短。当某个线程等待 I/O 时，操作系统可以调度其他线程执行，从而提高整体吞吐量。线程之间共享进程资源，创建和上下文切换成本较低，因此适合大量等待型任务。
  • ProcessPoolExecutor：适合CPU 密集型任务（数值计算、图像处理、数据分析等）。这类任务需要持续占用 CPU 进行计算，线程切换无法减少计算时间。CPython 存在全局解释器锁（GIL），同一进程中的多个线程执行 Python 字节码时，需要竞争 GIL，无法充分利用多个 CPU 核心并行计算。而每个进程拥有独立的 Python 解释器和 GIL，可以运行在不同 CPU 核心上，实现真正的并行计算。缺点是进程创建、上下文切换以及进程间通信成本更高

• 使用 Future 对象异步管理任务结果
• 统一 API，切换线程/进程只需换执行器类名
• 自动管理线程/进程的创建、调度和销毁

# 推荐导入方式
from concurrent.futures import (
    ThreadPoolExecutor,
    ProcessPoolExecutor,
    Future,
    as_completed,
    wait,
    FIRST_COMPLETED,
    FIRST_EXCEPTION,
    ALL_COMPLETED,
    TimeoutError,
    CancelledError,
)

执行器

concurrent.futures提供了 ThreadPoolExecutor（线程池执行器）和 ProcessPoolExecutor（进程池执行器）分别用于创建线程池和进程池，Executor 是它们的基类，两个执行器共有的方法有：

• e.submit(fn, *args, kwargs)提交单个任务，返回 Future 对象，参数 fn 为所执行的函数，*args为位置参数，kwargs为关键字参数
• e.map(fn, *iterables, timeout=None, chunksize=1)批量提交，返回结果迭代器，按提交顺序返回结果
• e.shutdown(wait=True, cancel_futures=False)关闭执行器，释放线程/进程资源（Py3.9+ 支持 cancel_futures），wait=True会等待任务完成，wait=False将立即返回，但通常使用width上下文管理器自动关闭，一般不手动执行

ThreadPoolExecutor

线程池执行器适合处理：HTTP请求、爬虫、文件IO、数据库访问，接收参数max_workers用于指定最大同时运行多少线程

from concurrent.futures import ThreadPoolExecutor

def task(x):
    return x * 2

# max_workers 默认 = min(32, os.cpu_count() + 4)
with ThreadPoolExecutor(max_workers=5) as executor:
    future = executor.submit(task, 10)
    print(future.result())  # 20

ProcessPoolExecutor

进程池执行器适合处理图像处理、数据分析等大量计算任务，接收参数max_workers用于指定最大同时运行多少进程，推荐的进程池数量 = CPU核心数，进程池要求 fn 和参数可 pickle 序列化，且必须在 if __name__ == '__main__': 保护下使用

from concurrent.futures import ProcessPoolExecutor

# max_workers 默认 = os.cpu_count()
with ProcessPoolExecutor(max_workers=4) as executor:
    results = executor.map(str.upper, ['a', 'b', 'c'])
    print(list(results))  # ['A', 'B', 'C']

Future未来对象

Future表示一个尚未完成的任务结果，该任务可能：正在执行/已完成/发生异常，因此将由Future负责保存：返回值、异常、执行状态，Future 是 submit() 的返回值。在进行异常处理时，submit 后异常不会立即抛出，必须调用 result() 才能捕获，以下是Future对象的方法：

• f.result(timeout=None)用于获取执行结果，如果任务未完成，会阻塞等待直到完成，异常会重新抛出，可以指定超时时间
• f.done()判断是否已完成（含取消/异常）
• f.cancelled()是否被取消
• f.running()是否正在运行（Py3.9+）
• f.cancel()尝试取消（未开始才能尝试取消，已经开始执行的线程无法通过该方法强制停止），返回True表示取消成功，返回False表示任务已经运行，无法取消
• f.exception(timeout=None)获取异常对象，无异常返回 None
• f.add_done_callback(fn)添加完成回调，任务结束自动执行函数，常用于日志记录、状态通知、善后处理
  def callback(future): print( future.result() ) future.add_done_callback(callback)
• f.set_result() / set_exception()一般仅自定义 Executor 使用
• f._state内部状态（PENDING/RUNNING/CANCELLED/FINISHED）

as_completed()

区别于map(按提交顺序返回结果)，as_completed()用于按任务完成顺序获取 Future，用于“谁先完成谁先处理”的场景:
as_completed(fs, timeout=None)传入多个Future，按完成顺序进行处理

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def task(n):
    time.sleep(n)
    return n

with ThreadPoolExecutor() as ex:
    futures = [ex.submit(task, i) for i in [3, 1, 2]]
    for f in as_completed(futures):
        print(f.result())   # 输出顺序：1, 2, 3

wait()

wait()函数用于等待多个Future：
wait(fs, timeout=None, return_when=ALL_COMPLETED)阻塞等待，返回 (done集合, not_done未完成集合) 两个集合，return_when支持的值有：

• ALL_COMPLETED(默认,全部完成才返回)
• FIRST_COMPLETED(任一任务完成即返回)
• FIRST_EXCEPTION(任一任务抛异常即返回)

from concurrent.futures import wait, FIRST_COMPLETED

done, not_done = wait(futures, timeout=5, return_when=FIRST_COMPLETED)

异常类

• concurrent.futures.TimeoutError result()/exception() 超时；Py3.11+ 等同内置 TimeoutError
• concurrent.futures.CancelledError 对已取消的 Future 调用 result()
• concurrent.futures.BrokenExecutor 执行器损坏，无法提交新任务
• concurrent.futures.InvalidStateError Future 状态非法操作

注意事项

• 线程池大小：I/O 密集型可远超 CPU 核数（如 50~100）；进程池建议 = CPU 核数。
• 异常处理：submit 后异常不会立即抛出，必须调用 result() 才能捕获。
• map 是惰性的：不会立即执行，迭代时才提交任务。
• 避免嵌套进程池：可能导致死锁。
• shutdown(wait=True) 会等待所有任务完成；配合 with 会自动调用。
• Python 3.9+ 才能用 shutdown(cancel_futures=True)。
• 进程池函数必须定义在模块顶层，不能是局部函数或 lambda。

模板

# 乱序收集，最常用：submit + as_completed
with ThreadPoolExecutor(max_workers=10) as ex:
    futures = {ex.submit(fetch, url): url for url in urls}
    for f in as_completed(futures):
        url = futures[f]
        try:
            data = f.result()
        except Exception as e:
            print(f"{url} 失败: {e}")

# map（顺序收集，简洁）
with ProcessPoolExecutor() as ex:
    for result in ex.map(cpu_task, range(100), chunksize=10):
        print(result)

# 批量发起爬虫请求
import requests
from concurrent.futures import ThreadPoolExecutor

def fetch(url):
    response = requests.get(url)
    return response.text

urls = [
    "https://example.com",
    "https://example.org"
]

with ThreadPoolExecutor(max_workers=5) as executor:

    futures = [
        executor.submit(fetch, url)
        for url in urls
    ]

    for future in futures:
        print(
            future.result()
        )

#多个worker线程请求，主线程收集 Future 结果并写入，不需要文件锁
import requests
from concurrent.futures import ThreadPoolExecutor
 
def fetch(url):
    response = requests.get(url)
    return response.text
 
urls = [
    "https://example.com",
    "https://example.org"
]

with ThreadPoolExecutor(max_workers=10) as executor:
    futures = [
        executor.submit(fetch, url)
        for url in urls
    ]

    with open("result.csv", "w", encoding="utf-8") as f:
        for future in as_completed(futures):
            result = future.result() # 响应拿到html
            data=fn(result) # 解析html中的数据
            f.write(data)

# 每个线程持创建独立的session并复用
import threading
import requests
from concurrent.futures import ThreadPoolExecutor

_local = threading.local()

def get_session():
    """每个线程第一次调用时才创建自己的 Session"""
    if not hasattr(_local, "session"): # 通过threading.local()检查线程是否已有session
        _local.session = requests.Session()
        _local.session.headers.update({"User-Agent": "..."})
    return _local.session

def fetch_page(i):
    s = get_session()          # 获得session
    r = s.get(f"https://httpbin.org/get?page={i}", timeout=10)
    return r

with ThreadPoolExecutor(max_workers=5) as ex:
    futures = [ex.submit(fetch_page, i) for i in range(1, 10)]
    for f in futures:
        print(f.result())