Python并发编程入门：从`threading`到`asyncio`，运维脚本性能提升指南

当你的Shell脚本还在串行等待时，隔壁组的Python脚本已经用并发把批量任务时间缩短了5倍。

作为IT运维/开发人员，你是否遇到过这些场景：批量SSH检查服务器状态耗时过长、需要同时拉取多个API接口数据、处理大量日志文件时CPU利用率极低。单线程串行执行是罪魁祸首。今天，我们深入Python并发编程的三种主流方案，并给出可直接落地的代码。

 

一、为什么你的脚本慢？先分清I/O密集与CPU密集

 

在写并发代码前，必须明确任务类型：

 

• I/O密集型：网络请求、文件读写、数据库查询。瓶颈在等待I/O完成，CPU大部分时间空闲。

• CPU密集型：复杂计算、数据压缩、加密解密。瓶颈在CPU执行速度。

 

重要结论：Python的threading由于GIL（全局解释器锁），无法利用多核并行执行CPU密集型代码。因此：

• 对于I/O密集型 → 使用 threading 或 asyncio

• 对于CPU密集型 → 使用 multiprocessing 或直接调用concurrent.futures.ProcessPoolExecutor

 

二、`threading`：最直观的并发模型（适合I/O密集）

 

threading 通过创建线程来并发执行。虽然受GIL限制，但在等待I/O时线程会释放GIL，因此依然能大幅提升效率。

 

经典案例：批量检测端口连通性

 

import threading

import socket

from concurrent.futures import ThreadPoolExecutor

def check_port(host, port, timeout=2):

　　"""检测指定主机端口是否开放"""

　　try:

　　　　with socket.create_connection((host, port), timeout=timeout):

　　　　　　return f"{host}:{port} OPEN"

　　except (socket.timeout, ConnectionRefusedError):

　　　　return f"{host}:{port} CLOSED"

hosts = ["192.168.1.1", "192.168.1.2", "192.168.1.3"]

ports = [22, 80, 443, 3306]

# 使用线程池，最大并发10个任务

with ThreadPoolExecutor(max_workers=10) as executor:

　　tasks = [(h, p) for h in hosts for p in ports]

　　results = list(executor.map(lambda x: check_port(*x), tasks))

for res in results:

　　print(res)

 

注意：threading 适合任务数小于1000的场景。如果任务过多，频繁创建线程会带来开销。

 

三、`asyncio`：单线程事件循环，更轻量高效（I/O密集）

 

asyncio 是Python 3.4+的标准库，使用协程和事件循环，在单线程内实现并发。它比线程更轻量，能轻松支持数万并发连接。适合高并发网络请求场景。

 

实战：并发调用多个REST API

 

import asyncio

import aiohttp

async def fetch_url(session, url):

　　"""异步获取URL内容"""

　　async with session.get(url) as response:

　　　　if response.status == 200:

　　　　　　data = await response.text()

　　　　　　return f"{url} -> {len(data)} bytes"

　　　　else:

　　　　　　return f"{url} -> HTTP {response.status}"

async def main():

　　urls = [

　　　　"https://api.example.com/health",

　　　　"https://api.example.com/metrics",

　　　　"https://api.example.com/version"

　　]

　　async with aiohttp.ClientSession() as session:

　　　　# 并发执行所有请求

　　　　tasks = [fetch_url(session, url) for url in urls]

　　　　results = await asyncio.gather(*tasks)

　　　　for res in results:

　　　　　　print(res)

if __name__ == "__main__":

　　asyncio.run(main())

 

关键点：

• 必须使用异步库（如aiohttp替代requests）

• 用asyncio.gather()聚合多个协程

• 注意：协程内不能有阻塞调用（比如time.sleep），要用await asyncio.sleep()替代

 

四、`concurrent.futures`：统一接口，灵活切换（推荐）

 

concurrent.futures 提供了高层抽象，让你用同一套代码选择线程池或进程池。

 

场景：混合任务（I/O + CPU）——下载日志并计算MD5

 

import hashlib

import requests

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

# 模拟下载日志文件（I/O密集型）

def download_log(url):

　　resp = requests.get(url, timeout=5)

　　return resp.content

# 计算MD5（CPU密集型，但数据量小，用线程亦可）

def calc_md5(data):

　　return hashlib.md5(data).hexdigest()

def process_url(url):

　　# 第一步：下载（I/O）

　　content = download_log(url)

　　# 第二步：计算（CPU）

　　return calc_md5(content)

urls = [

　　"http://logserver/app1.log",

　　"http://logserver/app2.log",

　　"http://logserver/app3.log"

]

# 如果I/O占比高，用线程池；如果CPU占比高，改用ProcessPoolExecutor

with ThreadPoolExecutor(max_workers=5) as executor:

　　md5_list = list(executor.map(process_url, urls))

for md5 in md5_list:

　　print(f"MD5: {md5}")

 

切换技巧：只需将ThreadPoolExecutor换成ProcessPoolExecutor，代码其余部分不变。但注意进程池会消耗更多内存，且参数必须可序列化。

 

五、实战对比与选择建议

 

| 方案 | 适用场景 | 并发上限 | 代码复杂度 | 调试难度 |

|------|---------|---------|-----------|---------|

| threading | 中小规模I/O任务 | ~1000线程 | 低 | 中（存在竞态条件） |

| asyncio | 高并发网络请求 | 数万协程 | 中（需理解async/await） | 高（单线程调试） |

| ProcessPoolExecutor | CPU密集型 | 等于CPU核数 | 低 | 低（进程隔离） |

 

运维落地建议：

1. 如果只是临时跑一个批量脚本，优先用ThreadPoolExecutor，简单可靠。

2. 如果是常驻服务（如监控系统），需要处理海量连接，用asyncio。

3. 如果涉及大量数值计算，用multiprocessing，并注意if __name__ == "__main__"保护。

 

六、避免踩坑：并发编程三大雷区

 

• 共享状态：多个线程同时修改同一变量会导致数据错乱。解决方案：使用threading.Lock或改用asyncio（单线程天然安全）。

• 阻塞调用：在协程中调用time.sleep()或requests.get()会阻塞整个事件循环。务必使用await asyncio.sleep()和异步库。

• 过度并发：线程数不是越多越好，通常max_workers设为CPU核数 * (I/O等待时间/CPU执行时间)。经验值：I/O密集用50-100，CPU密集用os.cpu_count()。

 

七、进阶提升：结合`subprocess`处理外部命令

 

运维场景中经常需要并发执行Shell命令。使用asyncio.create_subprocess_exec实现异步调用外部程序：

 

import asyncio

async def run_cmd(cmd):

　　process = await asyncio.create_subprocess_exec(

　　　　*cmd,

　　　　stdout=asyncio.subprocess.PIPE,

　　　　stderr=asyncio.subprocess.PIPE

　　)

　　stdout, stderr = await process.communicate()

　　return cmd[0], process.returncode, stdout.decode().strip()

async def main():

　　cmds = [

　　　　["ping", "-c", "2", "192.168.1.1"],

　　　　["df", "-h"],

　　　　["uptime"]

　　]

　　tasks = [run_cmd(cmd) for cmd in cmds]

　　for cmd, code, output in await asyncio.gather(*tasks):

　　　　print(f"=== {cmd} (exit {code}) ===")

　　　　print(output)

asyncio.run(main())

 

总结

 

Python并发编程的核心不是"多线程"而是"异步等待"——理解你的任务是I/O密集还是CPU密集，然后选择threading、asyncio或ProcessPoolExecutor。从今天起，用concurrent.futures统一你的并发代码，让运维脚本真正"跑起来"。

以上就是“Python并发编程入门：从`threading`到`asyncio`，运维脚本性能提升指南”的详细内容，想要了解更多Python教程欢迎持续关注编程学习网。