当前位置： 技术文章>> 如何在 Python 中结合 asyncio 和 concurrent.futures 处理并发任务？

文章标题：如何在 Python 中结合 asyncio 和 concurrent.futures 处理并发任务？
  • 文章分类: 后端
  • 6490 阅读

在Python中，asyncio 和 concurrent.futures 是处理并发任务的两个强大库，它们各自有不同的应用场景和优势。asyncio 主要用于编写单线程的并发代码，通过协程（coroutine）实现非阻塞的IO操作，特别适合于处理IO密集型任务。而 concurrent.futures 则提供了高层次的API来异步执行可调用对象，包括ThreadPoolExecutor和ProcessPoolExecutor，适用于CPU密集型任务或者需要并行处理多个独立任务的情况。将这两者结合使用，可以充分发挥Python在并发编程方面的优势。

理解 asyncio 和 concurrent.futures 的基础

asyncio

asyncio 是Python 3.4及以后版本中引入的，用于编写单线程并发代码。它基于事件循环（event loop），所有协程的执行都依赖于这个事件循环。协程通过await关键字来挂起执行，直到某个异步操作完成。asyncio 非常适合用于IO密集型任务，如网络请求、文件读写等。

concurrent.futures

concurrent.futures 是Python 3.2中引入的一个模块，它提供了高层次的API来异步执行可调用对象。ThreadPoolExecutor和ProcessPoolExecutor是其中最常用的两个类，分别用于在线程池和进程池中异步执行任务。ThreadPoolExecutor适合IO密集型任务，因为它可以在多个线程之间共享全局解释器锁（GIL）的释放时间；而ProcessPoolExecutor则适用于CPU密集型任务，因为它可以完全避免GIL的限制，通过多进程实现真正的并行计算。

结合使用 asyncio 和 concurrent.futures

尽管asyncio和concurrent.futures在设计上有所不同，但在某些情况下，你可能需要将它们结合使用以满足复杂的并发需求。以下是一些场景和策略：

场景一：在 asyncio 协程中调用同步代码

当你需要在asyncio协程中调用一些不支持异步的同步代码时，可以使用concurrent.futures.ThreadPoolExecutor来在单独的线程中执行这些同步代码。这样做可以避免阻塞事件循环。

import asyncio
from concurrent.futures import ThreadPoolExecutor

async def run_in_thread(func, *args):
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(ThreadPoolExecutor(), func, *args)

async def main():
    # 假设 heavy_lifting 是一个耗时的同步函数
    result = await run_in_thread(heavy_lifting, arg1, arg2)
    print(f'Result: {result}')

# 示例同步函数
def heavy_lifting(arg1, arg2):
    # 模拟耗时操作
    import time
    time.sleep(2)
    return arg1 + arg2

# 运行 asyncio 程序
asyncio.run(main())

场景二：在 asyncio 程序中管理多个 concurrent.futures 线程池

如果你的应用需要频繁地在多个线程中执行不同的任务，并且这些任务与asyncio协程有交互，你可以考虑在asyncio程序中管理多个ThreadPoolExecutor实例。

import asyncio
from concurrent.futures import ThreadPoolExecutor

# 创建一个线程池
executor = ThreadPoolExecutor(max_workers=5)

async def run_task(task_func, *args):
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(executor, task_func, *args)

async def main():
    # 同时启动多个任务
    tasks = [
        run_task(heavy_lifting, 'arg1', 1),
        run_task(another_task, 'arg2', 2),
        # ... 其他任务
    ]
    results = await asyncio.gather(*tasks)
    print(results)

# 其他同步函数
def another_task(arg, num):
    import time
    time.sleep(1)
    return f'{arg} processed {num}'

# 运行 asyncio 程序
asyncio.run(main())

场景三：在 concurrent.futures 线程池中调用 asyncio 协程

虽然这种情况较为少见且不推荐（因为asyncio协程应该在事件循环中运行），但在某些特殊情况下，你可能需要在ThreadPoolExecutor的线程中启动或操作asyncio事件循环。这通常涉及到复杂的状态管理和错误处理，因此应谨慎使用。

注意事项

  • 避免在ThreadPoolExecutor中启动多个事件循环：每个Python进程应只有一个活动的事件循环。
  • 资源管理：确保ThreadPoolExecutor和ProcessPoolExecutor在使用完毕后被正确关闭，以避免资源泄露。
  • 错误处理：在并发编程中，错误处理变得尤为重要。确保你的代码能够优雅地处理各种异常情况。
  • 性能考虑：虽然ThreadPoolExecutor可以在一定程度上提高IO密集型任务的性能，但它并不能完全替代asyncio在IO密集型场景下的优势。在设计系统时，应根据具体任务类型选择最合适的并发模型。

总结

通过结合使用asyncio和concurrent.futures，Python程序可以更加灵活地处理各种并发任务。asyncio适用于IO密集型任务，而concurrent.futures则适用于需要并行处理多个独立任务或执行CPU密集型任务的情况。在实际开发中，应根据任务的具体需求和系统架构选择合适的并发模型，并合理设计代码以充分发挥它们的优势。在码小课网站中，我们将深入探讨更多关于并发编程的高级话题，帮助开发者更好地掌握这些强大的工具。

推荐文章