并发编程是提升 Python 程序效率的关键技能，尤其在处理多任务场景时作用显著。本文将系统介绍 Python 中主流的并发实现方式，帮助你根据场景选择最优方案。

一、多线程编程（threading）

核心特点：共享进程内存，适合 I/O 密集型任务（如网络请求、文件读写），受 GIL（全局解释器锁）限制，无法真正并行执行 CPU 密集型任务。

基础用法

import threading

import time

def task(name):

print(f"任务 {name} 开始")

time.sleep(2) # 模拟I/O操作（不占用CPU）

print(f"任务 {name} 完成")

# 创建并启动线程

threads = []

for i in range(3):

t = threading.Thread(target=task, args=(f"Thread-{i}",))

threads.append(t)

t.start()

# 等待所有线程结束

for t in threads:

t.join()

print("所有任务完成")

线程同步（解决资源竞争）

当多个线程操作同一资源时，需用锁保证数据一致性：

from threading import Lock

import time

lock = Lock()

counter = 0

def increment():

global counter

with lock: # 自动获取和释放锁，避免死锁风险

temp = counter

time.sleep(0.1) # 模拟处理过程

counter = temp + 1

# 多线程计数

threads = [threading.Thread(target=increment) for _ in range(10)]

for t in threads:

t.start()

for t in threads:

t.join()

print(f"最终计数：{counter}") # 正确输出10

二、多进程编程（multiprocessing）

核心特点：独立内存空间，可绕过 GIL 实现真正并行，适合 CPU 密集型任务（如计算、图像处理），但进程间通信成本较高。

基础用法

from multiprocessing import Process

import os

def compute_square(number):

result = number * number

print(f"进程 {os.getpid()} 计算：{number} 的平方是 {result}")

return result

if __name__ == '__main__': # Windows系统必须加此判断

numbers = [1, 2, 3, 4, 5]

processes = []

for num in numbers:

p = Process(target=compute_square, args=(num,))

processes.append(p)

p.start()

for p in processes:

p.join() # 等待进程结束

print("所有进程完成")

进程池（高效管理多进程）

from multiprocessing import Pool

def process_data(data):

return data * 2 # 模拟数据处理

if __name__ == '__main__':

with Pool(4) as pool: # 4个工作进程

inputs = [1, 2, 3, 4, 5, 6, 7, 8]

results = pool.map(process_data, inputs) # 批量处理

print(results) # 输出 [2, 4, 6, 8, 10, 12, 14, 16]

三、异步编程（asyncio）

核心特点：单线程内的协程并发，通过事件循环实现非阻塞 I/O，效率高于多线程，适合高并发 I/O 场景（如异步爬虫、API 服务）。

基础用法

import asyncio

async def fetch_data(task_id, delay):

print(f"任务 {task_id} 开始")

await asyncio.sleep(delay) # 非阻塞等待（不占用线程）

print(f"任务 {task_id} 完成")

return f"结果-{task_id}"

async def main():

# 并发运行多个协程

tasks = [

fetch_data(1, 2),

fetch_data(2, 1),

fetch_data(3, 3)

]

results = await asyncio.gather(*tasks) # 等待所有任务完成

print(f"所有任务结果：{results}")

asyncio.run(main()) # 启动事件循环（Python 3.7+）

异步网络请求示例

import aiohttp

import asyncio

async def fetch_url(url):

async with aiohttp.ClientSession() as session: # 异步HTTP客户端

async with session.get(url) as response:

return await response.text() # 非阻塞读取响应

async def main():

urls = [

'https://www.example.com',

'https://www.python.org',

'https://www.github.com'

]

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

pages = await asyncio.gather(*tasks) # 并发请求

for url, content in zip(urls, pages):

print(f"{url} 返回 {len(content)} 字节")

asyncio.run(main())

四、高级并发工具（concurrent.futures）

提供统一接口，可无缝切换线程池/进程池，简化并发代码。

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

import time

def task(n):

time.sleep(1) # 模拟任务耗时

return n * n

# 线程池（适合I/O密集型）

with ThreadPoolExecutor(max_workers=3) as executor:

results = list(executor.map(task, range(5))) # 批量执行

print(f"线程池结果：{results}")

# 进程池（适合CPU密集型）

with ProcessPoolExecutor(max_workers=3) as executor:

results = list(executor.map(task, range(5)))

print(f"进程池结果：{results}")

五、并发方式选择指南

任务类型 推荐方式 适用场景

I/O 密集型 异步编程（asyncio） 高并发网络请求、API服务

I/O 密集型 多线程（threading） 简单I/O任务（如文件处理）

CPU 密集型 多进程（multiprocessing） 数学计算、图像/视频处理

代码简洁优先 concurrent.futures 快速实现，需灵活切换线程/进程

六、实战案例：并发文件下载器

import os

import time

import requests

from concurrent.futures import ThreadPoolExecutor

def download_file(url, save_path):

"""下载文件并返回结果信息"""

start_time = time.time()

try:

response = requests.get(url, stream=True) # 流式下载

with open(save_path, 'wb') as f:

for chunk in response.iter_content(chunk_size=8192):

if chunk:

f.write(chunk)

file_size = os.path.getsize(save_path) / 1024 # 转换为KB

耗时 = time.time() - start_time

return f"成功：{url}（{file_size:.2f}KB，耗时{耗时:.2f}秒）"

except Exception as e:

return f"失败：{url}（错误：{str(e)}）"

def main():

# 待下载文件列表

files = [

("https://example.com/file1.zip", "file1.zip"),

("https://example.com/file2.zip", "file2.zip"),

("https://example.com/file3.zip", "file3.zip"),

]

# 并发下载（线程池适合网络I/O）

with ThreadPoolExecutor(max_workers=3) as executor:

# 提交所有任务

futures = [executor.submit(download_file, url, path) for url, path in files]

# 获取结果

for future in futures:

print(future.result())

if __name__ == '__main__':

main()

七、常见问题与解决方案

1. 死锁：避免嵌套锁，使用 with 语句自动释放锁，设置超时时间（如 lock.acquire(timeout=5)）。

2. 资源竞争：对共享资源操作加锁（线程）或用队列（进程），优先使用线程安全的数据结构。

3. 异常处理：在并发任务中捕获异常，避免单个任务失败导致整体崩溃：

from concurrent.futures import as_completed

with ThreadPoolExecutor() as executor:

futures = {executor.submit(task, i): i for i in range(5)}

for future in as_completed(futures):

try:

result = future.result()

except Exception as e:

print(f"任务 {futures[future]} 出错：{e}")

else:

print(f"任务 {futures[future]} 结果：{result}")

通过合理选择并发方式，可显著提升 Python 程序的执行效率。实际开发中需结合任务类型、代码复杂度和维护成本综合考量，优先使用高级接口（如 asyncio、concurrent.futures）简化实现。

#p y t hon##python打卡##python自学##分享我的头条荣誉##爆料##在头条记录我的2025#