首页 > 编程语言 >Python多任务处理方法：多线程与异步编程如何选？

Python多任务处理方法：多线程与异步编程如何选？

来源：互联网 2026-07-29 10:01:18

python资讯合集

Python多任务处理包括多线程、多进程、异步编程、concurrent.futures模块及gevent等协程库。多线程适合I/O密集型但受GIL限制，多进程实现CPU密集型并行，异步编程高效处理高并发I/O，需注意线程安全、进程间通信及异常处理。

Python多线程：threading模块实现I/O密集型任务并发

在Python的世界里，实现多任务处理（也就是并发或并行）有不止一条路可以走。到底选哪条，得看任务本身是“I/O密集型”还是“CPU密集型”，以及你手头的具体需求。下面就把这些常见的方法和它们的实操手法捋一遍，希望能帮你在实际开发中快速找到方向。

1. 多线程（threading模块）

适用场景：I/O密集型任务，比如文件读写、网络请求这类。不过要注意，Python的全局解释器锁（GIL）是个硬伤，多线程在CPU密集型任务上基本使不上劲。

长期稳定更新的攒劲资源： >>>点此立即查看<<<

基本用法：

import threading
import time
def worker(name, delay):
    print(f"线程 {name} 开始")
    time.sleep(delay)
    print(f"线程 {name} 结束")
# 创建多个线程
thread1 = threading.Thread(target=worker, args=("A", 2))
thread2 = threading.Thread(target=worker, args=("B", 3))
# 启动线程
thread1.start()
thread2.start()
# 等待所有线程完成
thread1.join()
thread2.join()
print("所有线程已完成")

常见使用手法：

线程池：用concurrent.futures.ThreadPoolExecutor来管理线程，省心得多。

from concurrent.futures import ThreadPoolExecutor
import time
def worker(name, delay):
    print(f"线程 {name} 开始")
    time.sleep(delay)
    print(f"线程 {name} 结束")
    return f"结果 {name}"
with ThreadPoolExecutor(max_workers=2) as executor:
    future1 = executor.submit(worker, "A", 2)
    future2 = executor.submit(worker, "B", 3)
    print(future1.result())
    print(future2.result())
print("所有线程已完成")

2. 多进程（multiprocessing模块）

适用场景：CPU密集型任务，比如复杂的数学计算、大规模数据处理。多进程能绕过GIL，实现真正的并行计算，这是它的核心优势。

基本用法：

from multiprocessing import Process
import time
def worker(name, delay):
    print(f"进程 {name} 开始")
    time.sleep(delay)
    print(f"进程 {name} 结束")
if __name__ == "__main__":
    process1 = Process(target=worker, args=("A", 2))
    process2 = Process(target=worker, args=("B", 3))
    process1.start()
    process2.start()
    process1.join()
    process2.join()
    print("所有进程已完成")

常见使用手法：

进程池：concurrent.futures.ProcessPoolExecutor可以更高效地管理进程，避免频繁创建销毁的开销。

from concurrent.futures import ProcessPoolExecutor
import time
def worker(name, delay):
    print(f"进程 {name} 开始")
    time.sleep(delay)
    print(f"进程 {name} 结束")
    return f"结果 {name}"
if __name__ == "__main__":
    with ProcessPoolExecutor(max_workers=2) as executor:
        future1 = executor.submit(worker, "A", 2)
        future2 = executor.submit(worker, "B", 3)
        print(future1.result())
        print(future2.result())
    print("所有进程已完成")

3. Python异步编程：asyncio模块实现高并发I/O

适用场景：高并发的I/O密集型任务，比如网络服务器、爬虫，尤其是需要处理大量并发连接，但每个连接处理时间很短的情况。异步编程的核心思想是“等待时不阻塞”。

基本用法：

import asyncio
async def worker(name, delay):
    print(f"任务 {name} 开始")
    await asyncio.sleep(delay)
    print(f"任务 {name} 结束")
    return f"结果 {name}"
async def main():
    tasks = [
        asyncio.create_task(worker("A", 2)),
        asyncio.create_task(worker("B", 3))
    ]
    results = await asyncio.gather(*tasks)
    for result in results:
        print(result)
# 运行异步主函数
asyncio.run(main())

常见使用手法：

协程与事件循环：通过async/await定义协程，事件循环（event loop）负责调度执行。
异步I/O：配合aiohttp、aiomysql等异步库，实现非阻塞的I/O操作。

import asyncio
import aiohttp
async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()
async def main():
    urls = [
        "http://example.com",
        "http://python.org",
        "http://github.com"
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        pages_content = await asyncio.gather(*tasks)
        for content in pages_content:
            print(f"下载了 {len(content)} 字节")
asyncio.run(main())

4. 并发执行（concurrent.futures模块）

适用场景：这个模块是对多线程和多进程的高层封装，适合需要统一管理并发任务的场景，写起来更简洁。

基本用法：

ThreadPoolExecutor：I/O密集型任务的首选。

from concurrent.futures import ThreadPoolExecutor, as_completed
import time
def worker(name, delay):
    print(f"线程 {name} 开始")
    time.sleep(delay)
    print(f"线程 {name} 结束")
    return f"结果 {name}"
with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(worker, f"任务-{i}", i) for i in range(1, 4)]
    for future in as_completed(futures):
        print(future.result())
print("所有线程已完成")

ProcessPoolExecutor：CPU密集型任务的好搭档。

from concurrent.futures import ProcessPoolExecutor, as_completed
import time
def compute(n):
    print(f"计算 {n} 开始")
    time.sleep(2)
    result = n * n
    print(f"计算 {n} 结束")
    return result
with ProcessPoolExecutor(max_workers=2) as executor:
    futures = [executor.submit(compute, i) for i in range(1, 5)]
    for future in as_completed(futures):
        print(f"结果: {future.result()}")
print("所有计算已完成")

5. 协程库（如gevent、eventlet）

适用场景：需要轻量级并发，同时又想利用协程处理I/O操作。相比asyncio，gevent这类库更“自动”，通过打补丁的方式让标准库的阻塞操作变成非阻塞。

基本用法（以gevent为例）：

import gevent
from gevent import monkey
import time
# 打补丁，使标准库中的阻塞操作变为非阻塞
monkey.patch_all()
def worker(name, delay):
    print(f"协程 {name} 开始")
    gevent.sleep(delay)
    print(f"协程 {name} 结束")
def main():
    tasks = [
        gevent.spawn(worker, "A", 2),
        gevent.spawn(worker, "B", 3),
        gevent.spawn(worker, "C", 1)
    ]
    gevent.joinall(tasks)
    print("所有协程已完成")
if __name__ == "__main__":
    main()

Python多任务处理注意事项

在实际开发中，有几个坑不得不防：

1. 线程安全：多线程环境下，共享数据容易出乱子，必须用锁（Lock）或其他同步机制来保护。

import threading
lock = threading.Lock()
shared_resource = 0
def increment():
    global shared_resource
    with lock:
        shared_resource += 1
        print(f"共享资源: {shared_resource}")
threads = [threading.Thread(target=increment) for _ in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

2. 进程间通信（IPC）：多进程之间不像线程那样共享内存，需要通过Queue、Pipe等机制来传递数据。

from multiprocessing import Process, Queue
def worker(q, n):
    q.put(n * n)
if __name__ == "__main__":
    q = Queue()
    processes = [Process(target=worker, args=(q, i)) for i in range(5)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    results = []
    while not q.empty():
        results.append(q.get())
    print("结果:", results)

3. 异常处理：多任务中任何一个任务崩了都可能影响整体，所以一定要捕获并处理异常。

from concurrent.futures import ThreadPoolExecutor
def worker(n):
    if n == 3:
        raise ValueError("错误发生")
    return n * n
with ThreadPoolExecutor(max_workers=2) as executor:
    futures = [executor.submit(worker, i) for i in range(5)]
    for future in futures:
        try:
            result = future.result()
            print(f"结果: {result}")
        except Exception as e:
            print(f"任务出错: {e}")

4. 资源管理：线程池、进程池用完记得关闭，否则容易造成资源泄漏。用with上下文管理器最省心。

5. 选择合适的方法：I/O密集型用多线程或异步，CPU密集型用多进程，这个原则基本不会错。