面试题：Python等效函数调用之异步编程中的函数替代

Question

在Python异步编程中，`asyncio`库是常用的工具。`asyncio.create_task()`函数用于创建一个异步任务。假设我们要实现一个简单的异步任务调度，在某些场景下，可以使用`ensure_future()`函数来达到类似的效果。请深入分析这两个函数在异步函数调用上的等效性，包括它们对事件循环的影响、任务优先级设置以及异常处理机制等方面的差异。同时，编写一个复杂的异步应用场景代码示例，涵盖多个异步任务的创建、调度和协同工作，展示如何正确地使用这两个函数实现等效功能并体现出各自的特点。

Answer 1

asyncio.create_task() 与 ensure_future() 的等效性及差异分析

对事件循环的影响

• asyncio.create_task()：
  • 它是Python 3.7 引入的新特性，直接与事件循环交互，创建的任务会被立即排入事件循环的任务队列中。在没有显式获取事件循环时，它会自动使用当前运行的事件循环（如果存在），若不存在则抛出 RuntimeError。
  • 例如在以下代码中，task = asyncio.create_task(coro()) 直接将任务排入当前事件循环。

import asyncio


async def coro():
    await asyncio.sleep(1)
    print('Task completed')


async def main():
    task = asyncio.create_task(coro())
    await task


if __name__ == '__main__':
    asyncio.run(main())

• ensure_future()：
  • 它在Python 3.4 就已引入，同样是将协程包装成 Future 对象并排入事件循环。不过在Python 3.7 之前，使用 ensure_future() 时通常需要手动获取事件循环并将任务添加进去。在Python 3.7 及之后版本，它会自动使用当前运行的事件循环（如果存在）。
  • 如下代码，在Python 3.7 之前可能需要类似这样手动获取事件循环：

import asyncio


async def coro():
    await asyncio.sleep(1)
    print('Task completed')


loop = asyncio.get_event_loop()
future = asyncio.ensure_future(coro())
loop.run_until_complete(future)
loop.close()

任务优先级设置

• asyncio.create_task()：本身没有直接设置任务优先级的功能，任务按照加入事件循环任务队列的顺序依次执行。但可以通过自定义任务队列和调度算法来实现优先级调度。
• ensure_future()：同样没有内置的优先级设置功能，也是按照加入事件循环任务队列的顺序执行。在需要设置优先级时，也需要借助外部机制实现，例如自定义优先级队列，并在将任务加入事件循环前进行优先级排序。

异常处理机制

• asyncio.create_task()：当任务内部发生未处理的异常时，该异常会一直向上传播，直到被 await 该任务的地方捕获。如果在任务结束时仍未被捕获，事件循环会将其作为未处理异常记录并可能打印警告信息。

import asyncio


async def coro():
    raise ValueError('Custom error')


async def main():
    task = asyncio.create_task(coro())
    try:
        await task
    except ValueError as e:
        print(f'Caught error: {e}')


if __name__ == '__main__':
    asyncio.run(main())

• ensure_future()：异常处理机制与 asyncio.create_task() 类似，异常会在 await 任务的地方被捕获。不过在早期版本中，如果手动管理事件循环，需要在事件循环运行结束后检查任务的异常情况，否则异常可能被忽略。例如在上述手动管理事件循环的代码中，若任务 future 发生异常，在 loop.run_until_complete(future) 之后需要额外检查 future.exception() 来处理异常。

复杂异步应用场景代码示例

import asyncio


async def task1():
    await asyncio.sleep(1)
    print('Task 1 completed')


async def task2():
    await asyncio.sleep(2)
    print('Task 2 completed')


async def task3():
    await asyncio.sleep(3)
    print('Task 3 completed')


async def main_create_task():
    tasks = [
        asyncio.create_task(task1()),
        asyncio.create_task(task2()),
        asyncio.create_task(task3())
    ]
    await asyncio.gather(*tasks)


async def main_ensure_future():
    loop = asyncio.get_running_loop()
    futures = [
        loop.create_future() for _ in range(3)
    ]
    loop.create_task(task1()).add_done_callback(lambda f: futures[0].set_result(None) if not futures[0].done() else None)
    loop.create_task(task2()).add_done_callback(lambda f: futures[1].set_result(None) if not futures[1].done() else None)
    loop.create_task(task3()).add_done_callback(lambda f: futures[2].set_result(None) if not futures[2].done() else None)
    await asyncio.gather(*futures)


if __name__ == '__main__':
    print('Using create_task:')
    asyncio.run(main_create_task())
    print('Using ensure_future:')
    asyncio.run(main_ensure_future())

在这个示例中，main_create_task 函数使用 asyncio.create_task() 创建并调度了三个异步任务，asyncio.gather() 用于等待所有任务完成。main_ensure_future 函数使用 ensure_future() 的等效功能（这里使用 loop.create_task() 模拟 ensure_future() 的功能，因为 ensure_future() 在Python 3.7 之后自动使用当前运行的事件循环，与 create_task 类似，但这里展示早期手动管理的思路），同样创建并调度了三个异步任务，也使用 asyncio.gather() 等待所有任务完成。通过这种方式展示了两者在复杂异步场景下实现等效功能及各自的特点。create_task 更简洁直观，而早期使用 ensure_future() 时需要更多手动管理事件循环和任务的操作。