Python异步任务优化技巧分享

本文深入剖析了在同步Python应用中高效运行异步后台任务的关键技巧，着重解决了`asyncio.create_task`创建的任务未正确`await`导致协程无法完成的问题。针对这一常见痛点，文章首先阐述了`asyncio`协作式多任务的核心机制，强调了`await`的重要性。进而，提出了两种实用的解决方案：一是通过在异步上下文中显式`await`后台任务，确保任务按顺序完成；二是巧妙地结合`threading`模块，在独立的线程中运行`asyncio`事件循环，从而实现异步任务与主同步应用的并行执行。无论您是希望在异步块中集中处理任务，还是需要将异步操作完全解耦到后台并行执行，本文都能为您提供清晰的指导和可行的代码示例，助力您的Python应用实现卓越的性能表现。

本文探讨了在现有同步Python应用中运行异步后台任务的策略。我们首先分析了使用`asyncio.create_task`但未正确`await`导致后台协程无法完成的问题，阐明了`asyncio`的协作式多任务机制。随后，提供了两种解决方案：一是通过在异步上下文中显式`await`后台任务来确保其顺序完成；二是通过结合`threading`模块，在单独的线程中运行`asyncio`事件循环，实现异步任务与主同步应用的并行执行。

理解Asyncio的异步机制

Python的asyncio库提供了一种编写并发代码的方式，它基于协程和事件循环实现协作式多任务。与传统的多线程或多进程不同，asyncio的并发并非真正的并行执行（即在同一时间点执行多个操作），而是通过在I/O操作（如网络请求、文件读写或asyncio.sleep）等待时，将控制权交还给事件循环，从而允许其他协程运行。这种机制被称为“协作式多任务”，意味着协程必须显式地通过await来放弃控制权。

当一个协程通过asyncio.create_task()被调度时，它被添加到事件循环中，但并不会立即执行。事件循环会在适当的时机开始执行它。关键在于，如果一个通过create_task创建的任务没有被await，并且事件循环在其完成之前就退出了，那么该任务可能无法完全执行。

初始问题的分析

考虑以下示例代码，它尝试在一个同步的main函数中，通过asyncio.run启动一个异步任务：

import asyncio
from time import sleep
import sys

async def task():
    for i in range(5):
        print(f"Background task iteration {i}")
        await asyncio.sleep(1)
    print('finished')

async def background_task():
    print("a")
    asyncio.create_task(task()) # 任务被创建但未被await
    print("b")

def main():
    print("Main program started python", sys.version)
    asyncio.run(background_task()) # 运行异步入口
    for i in range(3):
        sleep(3) # 主程序中的同步阻塞
        print(f"Main program iteration {i}")

if __name__ == "__main__":
    main()

运行上述代码，会得到类似以下的输出：

Main program started python 3.11.6 (main, Oct 23 2023, 22:48:54) [GCC 11.4.0]
a
b
Background task iteration 0
Main program iteration 0
Main program iteration 1
Main program iteration 2

从输出可以看出，task()协程只执行了一次迭代就停止了。这是因为：

1. 在background_task中，asyncio.create_task(task())只是将task协程调度到事件循环中，但并没有await它。
2. asyncio.run(background_task())执行完毕后，事件循环会退出。
3. 在task协程第一次遇到await asyncio.sleep(1)时，它会将控制权交还给事件循环。此时，由于background_task已经执行到末尾，asyncio.run所在的事件循环认为没有更多需要await的任务，便会退出。
4. 主程序main中的sleep(3)是同步阻塞的，它会暂停整个线程，这与asyncio的事件循环是独立的，无法让后台的asyncio任务继续执行。因此，task协程在第一次暂停后便没有机会恢复执行。

要确保asyncio.run中的所有任务完成，需要显式地await它们，或者确保事件循环持续运行直到所有任务完成。

解决方案一：在异步上下文中顺序完成任务

如果你的目标是在asyncio.run的整个生命周期内，确保所有被调度的异步任务都完成，那么最直接的方法就是在异步入口函数中await这些任务。

import asyncio
from time import sleep
import sys

async def task():
    for i in range(5):
        print(f"Background task iteration {i}")
        await asyncio.sleep(0.1) # 缩短睡眠时间以便快速观察
    print('finished')

async def background_task_sequential():
    print("a")
    # 创建任务并立即await它，确保其完成
    scheduled_task = asyncio.create_task(task()) 
    print("task scheduled")
    # 可以在此处执行其他异步操作
    # ...
    await scheduled_task # 等待后台任务完成
    print("b")

def main_sequential():
    print("Main program started python", sys.version)
    asyncio.run(background_task_sequential()) # 运行异步入口
    for i in range(3):
        sleep(0.5)
        print(f"Main program iteration {i}")

if __name__ == "__main__":
    main_sequential()

输出：

a
task scheduled
Background task iteration 0
Background task iteration 1
Background task iteration 2
Background task iteration 3
Background task iteration 4
finished
b
Main program iteration 0
Main program iteration 1
Main program iteration 2

解释： 通过在background_task_sequential中添加await scheduled_task，我们确保了asyncio.run会一直等待task()协程执行完毕，然后才继续执行background_task_sequential中的print("b")，并最终退出事件循环。这种方式使得task协程能够完整运行，但它仍然是在asyncio.run的阻塞时间内完成的，对于外部的同步main_sequential函数来说，asyncio.run仍然是一个阻塞操作。

解决方案二：结合线程实现并行后台执行

如果你的同步主程序需要继续执行，而异步任务需要在“后台”真正并行地运行，那么你需要将asyncio事件循环放到一个单独的线程中。这样，主线程可以继续执行同步代码，而另一个线程则负责运行异步事件循环。

import asyncio
from time import sleep
import sys
import threading

async def task():
    for i in range(5):
        print(f"Background task iteration {i}")
        await asyncio.sleep(1) # 恢复较长睡眠时间以观察并行效果
    print('finished')

async def background_task_parallel():
    print("a")
    await task() # 直接await task()，因为整个asyncio.run都在一个线程中运行
    print("b")

def main_parallel():
    print("Main program started python", sys.version)

    # 在一个新线程中运行asyncio事件循环
    # lambda函数用于包装asyncio.run，使其成为threading.Thread的target
    t = threading.Thread(target=lambda: asyncio.run(background_task_parallel()))
    t.start() # 启动新线程

    for i in range(3):
        sleep(3) # 主程序继续执行同步阻塞
        print(f"Main program iteration {i}")

    # 可选：等待后台线程完成
    # t.join() 
    print("Main program finished")

if __name__ == "__main__":
    main_parallel()

输出：

a
Background task iteration 0
Main program started python 3.11.6 (main, Oct 23 2023, 22:48:54) [GCC 11.4.0]
Background task iteration 1
Background task iteration 2
Main program iteration 0
Background task iteration 3
Background task iteration 4
finished
b
Main program iteration 1
Main program iteration 2
Main program finished

解释： 在这个方案中，我们创建了一个新的threading.Thread。这个新线程的目标函数是一个lambda表达式，它调用asyncio.run(background_task_parallel())。这意味着整个asyncio事件循环及其中的background_task_parallel（以及它所await的task协程）都在这个新线程中独立运行。

主线程在启动新线程后，立即继续执行其自身的同步循环（包含sleep(3)）。由于asyncio事件循环运行在不同的线程中，它不会被主线程的sleep阻塞，可以独立地推进其内部的异步任务。从输出可以看出，主程序的迭代和后台任务的迭代是交错进行的，这表明它们在并行执行。

注意事项：

• 线程安全： 当在不同线程中运行代码时，如果它们需要共享数据，必须考虑线程安全问题，例如使用锁（threading.Lock）来保护共享资源。
• 资源管理： 确保后台线程能够正常退出。如果后台任务是长期运行的，可能需要机制来优雅地停止它。
• asyncio.run()的阻塞性： 无论在哪种方案中，asyncio.run()本身都是一个阻塞调用，它会阻塞调用它的那个线程，直到其内部的事件循环完成。这也是为什么在方案二中需要将其放入单独的线程中。

总结

在同步Python应用中集成异步后台任务时，理解asyncio的协作式多任务原理至关重要。

• 如果你希望在asyncio.run的生命周期内确保所有异步任务顺序完成，应在异步入口函数中显式地await通过create_task创建的任务。
• 如果你需要异步任务在不阻塞主同步应用的情况下真正并行运行，最有效的方法是将其asyncio事件循环封装在一个单独的threading.Thread中。

选择哪种方案取决于你的具体需求：是希望在一个集中的异步块中完成所有工作，还是需要将异步操作完全解耦到后台并行执行。

文中关于的知识介绍，希望对你的学习有所帮助！若是受益匪浅，那就动动鼠标收藏这篇《Python异步任务优化技巧分享》文章吧，也可关注golang学习网公众号了解相关技术文章。