深入解析 Python 并发编程：从线程到协程的实战指南

2026-02-14 09:39:51 0条评论 2次阅读 0人点赞

在现代软件开发中，仅仅编写出能够运行的代码往往是不够的。随着数据量的增加和业务逻辑的复杂化，我们经常面临程序运行缓慢、响应迟钝的挑战。你也许遇到过这样的情况：一个脚本在处理网络请求或文件读写时，大部分时间都在“发呆”，等待外部响应；或者在进行复杂的科学计算时，CPU 一直在满负荷运转，但处理速度依然不够快。

这正是我们需要引入并发的原因。在这篇文章中，我们将深入探讨 Python 中的并发编程机制。我们将不仅局限于理论概念，更会通过实际的代码示例，带你一步步了解如何使用多线程、多进程和异步 IO 来显著提升程序的性能。无论你是想优化 Web 爬虫的抓取速度，还是想加速大规模的数据处理任务，这篇文章都将为你提供实用的见解和解决方案。

什么是并发编程？

并发是计算机科学中一个核心概念，它指的是系统在同一时间段内处理多个任务的能力。为了更准确地理解它，我们需要将其与另一个容易混淆的概念——“并行”——区分开来。

并发关乎的是结构。它允许程序处理多个任务，但这并不意味着这些任务必须在同一瞬间一起执行。想象一下，你正在一边做饭一边接电话。你可能切两下菜，然后接一句电话，再回来继续切菜。这两个任务在“重叠”的时间段内都在进行，但你的注意力在它们之间切换。
并行则关乎执行。它指的是两个或多个任务确实在同一时刻发生。这通常需要硬件的支持，比如多核 CPU。如果回到做饭的例子，如果你有一个助手专门负责接电话，而你只负责切菜，你们两人就在并行工作。

并发与并行的主要区别

为了更直观地理解，让我们用 Python 的术语来对比一下：

  • 并发：适用于 I/O 密集型任务。当程序需要频繁等待外部操作（如网络请求、磁盘读写）时，利用并发可以让 CPU 在等待期间去处理其他任务，从而最大化资源利用率。
  • 并行：适用于 CPU 密集型任务。当程序需要进行大量的数学运算或逻辑判断时，通过多核 CPU 同时运行多个任务，可以真正缩短计算时间。

Python 中的并发实现工具

Python 为我们提供了强大的标准库来实现并发，主要分为以下三大类，每一类都有其独特的适用场景：

  • 多线程： threading 模块。线程是轻量级的，共享进程的内存空间。这使得它们非常擅长处理 I/O 密集型任务（例如下载多个网页），因为线程切换的开销很小。但需要注意，由于 Python 的全局解释器锁（GIL）的存在，同一时刻只能有一个线程在执行 Python 字节码，因此它不适合用于 CPU 密集型任务。
  • 多进程： multiprocessing 模块。每个进程都有自己独立的内存空间和 Python 解释器。这绕过了 GIL 的限制，使得程序可以充分利用多核 CPU 的优势。因此，它是处理 CPU 密集型任务（如视频编码、复杂数据计算）的首选方案。
  • 异步 I/O： asyncio 库。这是一种基于事件循环和协程的并发模型。它完全由用户态的代码来调度任务，不需要操作系统的线程或进程介入。它在处理海量 I/O 操作时效率极高，且资源消耗极低。

实现并发编程的步骤

在开始编写代码之前，让我们先梳理一下实现并发系统的通用步骤：

  • 性能分析：首先确定你的程序瓶颈在哪里。如果是频繁的等待，选择多线程或异步 IO；如果是疯狂的计算，选择多进程。
  • 导入库：根据你的选择，导入 INLINECODEb1fcaf2d、INLINECODE41b3bc24 或 asyncio。
  • 定义任务：将你的业务逻辑封装成独立的函数或方法，确保任务之间尽量减少依赖。
  • 创建并启动执行单元：创建线程、进程或任务对象，并调用启动方法。
  • 同步与通信：如果任务间需要共享数据或协调顺序，使用锁、信号量或队列等机制，以防止竞态条件导致的数据错误。
  • 等待与清理：确保主程序能够等待所有后台任务完成后再退出，以保证数据完整性。

代码示例与实战解析

接下来，让我们通过具体的代码示例来演示如何在 Python 中实现这些并发模式。我们将从最简单的多线程开始。

示例 1：基于线程的并发 (Threading)

多线程最适合处理 I/O 密集型任务。在这个例子中，我们将模拟两个耗时的任务：打印数字和打印字母。为了模拟 I/O 阻塞，我们在每次打印之间人为增加了延迟。

import threading
import time

def print_numbers():
    """
    模拟一个执行耗时任务的线程函数：打印数字 1 到 5。
    在真实场景中，time.sleep 可以被网络请求或文件读写操作替代。
    """
    for i in range(1, 6):
        print(f"[线程 1] 正在打印数字: {i}")
        time.sleep(1)  # 模拟 I/O 操作或计算延迟

def print_letters():
    """
    模拟另一个执行耗时任务的线程函数：打印单词 Python。
    """
    for letter in ‘Python‘:
        print(f"[线程 2] 正在打印字母: {letter}")
        time.sleep(1)  # 模拟 I/O 操作或计算延迟

if __name__ == "__main__":
    # 记录开始时间以计算总耗时
    start_time = time.time()

    # 创建线程对象，target 参数指定线程要执行的函数
    thread1 = threading.Thread(target=print_numbers)
    thread2 = threading.Thread(target=print_letters)

    print("--- 启动并发线程 ---")
    # 启动线程，开始执行
    thread1.start()
    thread2.start()

    # join() 方法会让主程序等待这两个线程执行完毕后再继续
    thread1.join()
    thread2.join()

    print("--- 所有线程执行完毕 ---")
    print(f"总耗时: {time.time() - start_time:.2f} 秒")

代码解析：

在这个例子中，你可能会注意到输出是交错的（数字和字母混杂出现），并且总耗时接近于单个任务的时间，而不是两者之和。这证明了两个任务是并发执行的。

示例 2：基于进程的并发 (Multiprocessing)

如前所述，由于 GIL 的存在，多线程无法利用多核 CPU 来加速计算任务。让我们看看如何使用 multiprocessing 模块来并行处理 CPU 密集型工作。

import multiprocessing
import time

def cpu_bound_task(n):
    """
    一个 CPU 密集型任务：计算平方和。
    这将占用 CPU 资源，适合使用多进程。
    """
    total = 0
    for i in range(n):
        total += i * i
    print(f"进程 {multiprocessing.current_process().name} 计算完成: {total}")
    return total

if __name__ == "__main__":
    start_time = time.time()
    
    # 创建进程列表
    processes = []
    num_tasks = 4  # 我们将并行运行 4 个计算任务
    n = 10000000   # 计算量

    print(f"--- 在 {num_tasks} 个核上并行启动 {num_tasks} 个进程 ---")
    
    # 创建并启动进程
    for i in range(num_tasks):
        p = multiprocessing.Process(target=cpu_bound_task, args=(n,))
        processes.append(p)
        p.start()

    # 等待所有进程完成
    for p in processes:
        p.join()

    print(f"--- 所有进程执行完毕 ---")
    print(f"多进程总耗时: {time.time() - start_time:.2f} 秒")

    # 对比：如果是串行执行（仅供参考逻辑，实际运行注释掉）
    # print("执行串行对比计算...")
    # start_s = time.time()
    # for i in range(num_tasks):
    #     cpu_bound_task(n)
    # print(f"串行总耗时: {time.time() - start_s:.2f} 秒")

实战见解：

如果你在多核机器上运行这段代码，你会发现多进程版本的总耗时远远小于单个任务耗时乘以任务数的时间。进程之间拥有独立的内存空间，这意味着它们互不干扰，可以真正实现并行计算。

示例 3：使用 Pool (进程池) 优化

当任务数量非常多时，手动创建成百上千个进程会消耗大量系统资源。这时，使用 Pool（进程池）是更好的选择。它可以管理固定数量的工作进程，并分配任务给它们。

from multiprocessing import Pool
import os
import time

def heavy_computation(x):
    """
    模拟一个复杂的计算任务，稍微睡眠以模拟负载
    """
    print(f"工作进程 {os.getpid()} 正在处理数据: {x}")
    time.sleep(1) # 模拟处理耗时
    return x * x

if __name__ == "__main__‘:
    data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
    start_time = time.time()

    # 创建一个包含 4 个工作进程的进程池
    # with 语句会自动管理进程池的关闭和资源释放
    with Pool(processes=4) as pool:
        # map 方法会将 data 列表中的每个元素分配给进程池中的进程处理
        # 它会阻塞直到所有结果都准备好
        results = pool.map(heavy_computation, data)

    print(f"结果: {results}")
    print(f"Pool 总耗时: {time.time() - start_time:.2f} 秒")

通过使用 INLINECODEc5b28b84，我们可以像使用内置的 INLINECODE5aa5d05a 函数一样轻松地实现并行化，代码既简洁又高效。

示例 4：基于协程的并发 (Asyncio)

异步编程是现代 Python 开发中非常重要的一部分。让我们使用 INLINECODE51f3f71f 来重写第一个示例。注意这里的“延迟”不再是阻塞式的 INLINECODE3d490429，而是非阻塞的 await asyncio.sleep。

import asyncio
import time

# 定义一个协程函数
async def print_numbers_async():
    print("[协程 1] 开始打印数字")
    for i in range(1, 6):
        print(f"[协程 1] 正在打印数字: {i}")
        # await 关键字挂起当前协程，允许事件循环去运行其他任务
        await asyncio.sleep(1) 
    print("[协程 1] 完成")

async def print_letters_async():
    print("[协程 2] 开始打印字母")
    for letter in ‘Python‘:
        print(f"[协程 2] 正在打印字母: {letter}")
        await asyncio.sleep(1)
    print("[协程 2] 完成")

async def main():
    # 创建任务列表
    # asyncio.gather 并发运行多个任务，并等待它们全部完成
    await asyncio.gather(
        print_numbers_async(),
        print_letters_async()
    )

if __name__ == "__main__":
    print("--- 启动异步并发 ---")
    start_time = time.time()
    # 运行主程序
    asyncio.run(main())
    print(f"--- 协程执行完毕 ---")
    print(f"总耗时: {time.time() - start_time:.2f} 秒")

为什么使用 Asyncio？

在这个例子中，所有的代码实际上都是在同一个线程内运行的！当 INLINECODEad64e228 执行到 INLINECODE84e016f1 时，它并没有阻塞整个程序，而是告诉事件循环：“我这里要等 1 秒，你可以去执行 print_letters_async 了”。这种机制使得我们在单线程中也能处理成千上万个并发的 I/O 操作，资源利用率极高。

示例 5：实际应用场景 (使用 ThreadPoolExecutor)

在实际的 Web 开发中，我们经常需要从多个 URL 获取数据。这是一个典型的 I/O 密集型场景。使用 concurrent.futures 模块可以使代码更加现代化和易于管理。

import concurrent.futures
import requests
import time

# 模拟一个耗时的网络请求函数
def fetch_url_data(url):
    try:
        # 模拟网络延迟
        response = requests.get(url, timeout=5)
        return f"URL: {url} - 状态码: {response.status_code}"
    except Exception as e:
        return f"URL: {url} - 错误: {str(e)}"

if __name__ == "__main__":
    urls = [
        "https://www.google.com",
        "https://www.github.com",
        "https://www.python.org",
        "https://www.stackoverflow.com"
    ]

    print("--- 开始并发网络请求 (ThreadPoolExecutor) ---")
    start_time = time.time()

    # 使用 ThreadPoolExecutor 上下文管理器
    # max_workers 指定了线程池的大小
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        # 使用 executor.submit 提交任务到线程池
        futures = [executor.submit(fetch_url_data, url) for url in urls]
        
        # as_completed 返回一个迭代器，当任务完成时产出 Future 对象
        for future in concurrent.futures.as_completed(futures):
            print(future.result())

    print(f"--- 所有请求完成 ---")
    print(f"总耗时: {time.time() - start_time:.2f} 秒")
    print("注意：如果串行执行，耗时将是所有请求时间的总和。")

常见问题与最佳实践

在编写并发代码时，你可能会遇到一些常见的陷阱：

  • 竞态条件：当多个线程或进程试图同时修改同一个共享变量时，可能会导致数据损坏。解决方法是使用 Lock（锁）来确保同一时间只有一个线程可以访问该变量。
  • 死锁：如果两个线程互相等待对方持有的锁，程序就会永久卡死。避免死锁的最佳实践是按照固定的全局顺序获取锁，或者使用超时机制。
  • GIL 的误解：请记住，多线程在 Python 中并不是全能的。如果你的任务是计算密集型的，请务必使用多进程，否则多线程反而会因为上下文切换的开销而降低性能。
  • 过度并发：创建成千上万个线程或进程会耗尽系统内存。对于大规模 I/O 并发，请优先选择 INLINECODE155c46de 或使用 INLINECODE1e7757c4 / ProcessPoolExecutor 的限制版本。

总结与后续步骤

在本文中，我们一起探索了 Python 并发编程的核心概念和实现方式。我们了解到：

  • 并发适合 I/O 密集型任务，如网络爬虫和文件操作，主要工具是 INLINECODE2d55d41d 和 INLINECODE93c6bed1。
  • 并行适合 CPU 密集型任务，如数据分析，主要工具是 multiprocessing。
  • 代码的可读性和线程安全同样重要，合理使用锁和队列是关键。

要真正掌握这些技能，我们建议你尝试修改上面的示例代码。例如，尝试把串行运行的代码改成并行运行，用 time 模块观察性能提升的差异。你甚至可以尝试去构建一个简单的异步 Web 爬虫，将这些知识应用到实际项目中。现在，你已经具备了优化 Python 代码性能的强大工具，去构建更高效的系统吧！

声明：本站所有文章，如无特殊说明或标注，均为本站原创发布。任何个人或组织，在未征得本站同意时，禁止复制、盗用、采集、发布本站内容到任何网站、书籍等各类媒体平台。如若本站内容侵犯了原著者的合法权益，可联系我们进行处理。如需转载，请注明文章出处豆丁博客和来源网址。https://shluqu.cn/49073.html
点赞
0.00 平均评分 (0% 分数) - 0 票

相关文章美国1G带宽/1T流量高速vps $17.99/年