前言

由前面我们知道，关于协程（Coroutine）它是是一种可挂起的函数，它可以被暂停和恢复。在程序的执行过程中可以主动地交出控制权，而不是被动地等待I/O操作完成。它的主要应用场景是处理并发任务，可以在不阻塞线程的情况下进行异步操作。 而前面我们提到了一个案例是如下：

import asyncio
import time


async def say_after(delay, what):
    await asyncio.sleep(delay)  # 异步等待指定时间
    print(what)
async def main():
    print(f"开始时间 at {time.strftime('%X')}")
    await say_after(1, '我是第一名')
    await say_after(2, '我是第二名')
    print(f"结束时间 at {time.strftime('%X')}")
asyncio.run(main())

需要注意一点是：我们的 asyncio.sleep(delay)本质也是一个协程的哦！ 我们通过await asyncio.sleep(delay) 是为了交出控制权，方便执行其他的协程任务。

它执行的结果是顺序性的，如下所示：

我是第一名
我是第二名
结束时间 at 10:43:24

上面的情况是没有并发可言的，所以我们需要引入一个新的概念。

任务（Task）

任务（Task），我们可以定义为:

任务是协程的一种封装，它允许将协程作为异步可调用对象来使用。任务是异步编程中更细粒度的一个概念，通常用于描述一组操作的组合，例如将多个网络请求组合成一个任务。任务可以由事件循环调度器自动管理，并允许异步地执行。

简单的来说，我们的协程对象可能没有相关取消方法或结果状态这些，当我们使用任务包装了协程对象后，就可以获取当这个协程里面的一些状态或对它执行取消，可以获取到结果，可以获取到异常。

任务定义

下面我们开始换另一种方式来，具体的代码改变如下：

import asyncio
import time

async def say_after(delay, what):
    """异步地等待指定时间后打印what"""
    await asyncio.sleep(delay)  # 异步等待指定时间
    print(what)

async def main():
    """主函数，运行异步任务并打印开始和结束时间"""
    print(f"开始时间 at {time.strftime('%X')}")
    task = asyncio.create_task(say_after(1, '我是第一名'))
    print(f"task等待中... at {time.strftime('%X')}")
    task2 = asyncio.create_task(say_after(2, '我是第二名'))
    print(f"task2等待中... at {time.strftime('%X')}")
    print(f"结束时间 at {time.strftime('%X')}")

asyncio.run(main())

我们运行代码，查看输出结果为：

开始时间 at 11:15:03
task等待中... at 11:15:03
task2等待中... at 11:15:03
结束时间 at 11:15:03

从结果可以看得出，我们的任务程序直接运行后，就结束退出了并没有等待任务的处理结果。但是假设我们加上 await + 任务对象 会怎么样？如下代码所示：

async def main():
    """主函数，运行异步任务并打印开始和结束时间"""
    print(f"开始时间 at {time.strftime('%X')}")
    task = asyncio.create_task(say_after(1, '我是第一名'))
    print(f"task等待中... at {time.strftime('%X')}")
    task2 = asyncio.create_task(say_after(2, '我是第二名'))
    print(f"task2等待中... at {time.strftime('%X')}")
    await task
    await task2
    print(f"结束时间 at {time.strftime('%X')}")

查看输出的结果则为:

开始时间 at 11:19:44
task等待中... at 11:19:44
task2等待中... at 11:19:44
我是第一名
我是第二名
结束时间 at 11:19:46

此时我们仔细的对比一下输出结果的时间：

一个是直接 await+协程：

开始时间 at 10:43:21
我是第一名
我是第二名
结束时间 at 10:43:24

对象 一个直接的 await+任务：

开始时间 at 11:19:44
task等待中... at 11:19:44
task2等待中... at 11:19:44
我是第一名
我是第二名
结束时间 at 11:19:46

你会发现await+协程总体好像是两个协程对象的总和也就是3秒，而我们的await+任务的总体耗时是2秒！！！！so，这里我们就知道，它至少少了1秒的执行的时间，所以由此可知起到所谓的并发处理的作用。

需注意一点，假设我们的改的代码如下：

async def main():
    """主函数，运行异步任务并打印开始和结束时间"""
    print(f"开始时间 at {time.strftime('%X')}")
    await asyncio.create_task(say_after(1, '我是第一名'))
    print(f"task等待中... at {time.strftime('%X')}")
    await asyncio.create_task(say_after(2, '我是第二名'))
    print(f"task2等待中... at {time.strftime('%X')}")
    print(f"结束时间 at {time.strftime('%X')}")

asyncio.run(main())

你可以自己试一试运行一下，会发现和我们的前面的await+协程的方式一样，总体耗时是3秒哟！！！

所以需要注意：

await asyncio.create_task(say_after(1, '我是第一名'))
和
task = asyncio.create_task(say_after(1, '我是第一名'))
await task
是有点区别的哦！！！！！！！！！！！！！！！！！！！！！！！！！
是有点区别的哦！！！！！！！！！！！！！！！！！！！！！！！！！
是有点区别的哦！！！！！！！！！！！！！！！！！！！！！！！！！

在前面的示例中我们的创建任务对象是通过如下的方式：

task = asyncio.create_task(my_task())

需注意，当我们使用asyncio.create_task(my_task())之后其实我们的任务已经开始加入到事件循环中，只是等待着伺机而动！如下示例代码所示：

import asyncio
import time

async def say_after(delay, what):
    """异步地等待指定时间后打印what"""
    await asyncio.sleep(delay)  # 异步等待指定时间
    print(what)

async def main():
    """主函数，运行异步任务并打印开始和结束时间"""
    print(f"开始时间 at {time.strftime('%X')}")
    asyncio.create_task(say_after(1, '我是第一名'))
    print(f"task等待中... at {time.strftime('%X')}")
    await say_after(5, '测试等待，看看我们的任务我是第一名是否被执行！！！')
    print(f"结束时间 at {time.strftime('%X')}")

asyncio.run(main())

在上面代码中，我创建了任务asyncio.create_task(say_after(1, '我是第一名'))，但是没有进行await等待，此时如果我们启动了循环事件的话，它也会被司机而调度，为了观察被调度了，我们刻意加了一个 await say_after(5, '测试等待，看看我们的任务我是第一名是否被执行！！！')观察，得出的结果如下：

开始时间 at 11:32:08
task等待中... at 11:32:08
我是第一名
测试等待，看看我们的任务我是第一名是否被执行！！！
结束时间 at 11:32:13

总体的时间也是5秒，假如我们修改一下 asyncio.create_task(say_after(4, '我是第一名'))，如果也是并行的话，那理论上也还是5秒哦（还得看你自己机器哟），不信你可以自己试一试！

所以我们这里可以得出一个结论：

在调用 asyncio.create_task(my_task()) 之后，任务会被放入事件循环的任务队列中。一旦事件循环开始运行，它会按照先进先出的顺序逐个取出任务，并执行任务对应的协程函数。

如果你有GO的基础，应该也能理解GO的协程和我们这里的任务对象其实有点类似，在我们的python中，每个 Task 对应于一个协程（coroutine），任务是 asyncio 中进行并发编程的基本单位。

这里我们先归纳一下关于Task 的特点有哪些：

• 轻量级：Task 对象是轻量级的，创建一个 Task 的开销很小。和GO的协程类似。
• 异步执行：Task 在事件循环中异步执行，不会阻塞主线程。
• 可取消：可以通过 Task 的 cancel() 方法来取消尚未完成的 Task。
• 可获取结果和看进行异常处理：Task 的结果和异常可以通过 Task 对象的 result() 和 exception() 方法获取。

任务的取消

假如对于任务的执行，我们的有时限，需要对超时的任务进行取消处理，那么我们应该怎么进行任务的取消操作呢？首先我们先看如下的代码：

import asyncio
import time

async def say_after(delay, what):
    """异步地等待指定时间后打印what"""
    print(f"任务开始于{time.strftime('%X')}")
    try:
        await asyncio.sleep(delay)  # 异步等待指定时间  # 模拟耗时操作
    except asyncio.CancelledError:
        print('任务已经被取消了！！！！')
    else:
        print(f"任务结束于{time.strftime('%X')}")


async def main():
    """主函数，运行异步任务并打印开始和结束时间"""
    print(f"开始时间 at {time.strftime('%X')}")
    #  为演示任务的时长，我们假设这个任务需要耗时10秒
    task = asyncio.create_task(say_after(10, '我是第一名'))
    print(f"task等待中... at {time.strftime('%X')}")
    await asyncio.sleep(5)  # 异步等待指定时间
    print("执行任务取消")
    task.cancel()
    print("执行任务取消完成")
    # await asyncio.sleep(1)  # 异步等待指定时间
    print(f"结束时间 at {time.strftime('%X')}")

asyncio.run(main())

在上面的代码中，我们定义了一个say_after的协程函数，在这个协程函数内部我们有调用了另一个协程对象，也就是使用了 await asyncio.sleep(10)模拟了一个耗时操作，通过它可以来让任务暂停执行。另外我们还使用了 try-except 语句来捕获关于任务取消的 asyncio.CancelledError 异常，这是一个特殊的异常类，它表示任务已被取消。

所以这个异常的触发是，当我们调用 task.cancel() 方法时，事件循环会发送一个取消请求给任务。

接着我们在 main() 函数中通过 task = asyncio.create_task(say_after(10, '我是第一名'))创建了一个任务 task 并启动它（注意我们只是创建了任务哦，还没使用await），紧接着等待继续调用 5 秒钟后，我们调用了 task.cancel() 方法来取消任务。

上面的代码的执行结果如下：

开始时间 at 12:46:58
task等待中... at 12:46:58
任务开始于12:46:58
执行任务取消
执行任务取消完成
结束时间 at 12:47:03
任务已经被取消了！！！！

需要注意，并非是我们的执行了 task.cancel()之后，我们就会立即触发asyncio.CancelledError 的异常，也就是不会立即停止任务的执行，它只是会在事件循环下一个循环点上才回去执行，比如我们上面的如使用 try-except 捕获 CancelledError 异常里面的 await asyncio.sleep(delay)点。

反过来的话，如何避免任务被取消的操作？也就是说，我们的任务不管出现什么情况，都应该不可以被取消，即使的手动的调用，如下代码所示：

import asyncio

async def test_coroutine():
    """一个延迟执行的协程，会打印出"任务还是执行完成了！"""
    await asyncio.sleep(4)
    print('任务还是执行完成了！')

async def main():
    shielded_task = None
    try:
        task = asyncio.create_task(test_coroutine())  # 创建一个异步任务
        shielded_task = asyncio.shield(task)  # 创建一个受保护的任务
        await asyncio.wait_for(shielded_task, timeout=3)  # 等待受保护的任务完成，超时时间为3秒
    except asyncio.TimeoutError:
        shielded_task.cancel()  # 如果超时，则尝试进行取消受保护的任务
        print('超时了，并且尝试进行取消受保护的任务！！')
    except asyncio.CancelledError:
        print('在超时的时候，我们操作了取消任务，但是因为保护机制，这个取消任务被保护起来了，所以不会执行')

    # 挂起，演示我们的任务的取消是无效的！
    print('任务还在执行，我们取消它')
    await asyncio.sleep(8)  # 模拟一个长时间的任务
    print('挂起，演示我们的任务的取消是无效的！')

asyncio.run(main())  # 运行主函数

最终上面的代码输出结果为：

超时了，并且尝试进行取消受保护的任务！！
任务还在执行，我们取消它
任务还是执行完成了！
挂起，演示我们的任务的取消是无效的！

从输出的结果可以看出，当我们尝试进行任务的消息的时候，还是不能取消，任务还是一样的执行了！！ 对比下面的代码，我们可以知道取消是成功的：

import asyncio

async def test_coroutine():
    """一个延迟执行的协程，会打印出"任务还是执行完成了！"""
    await asyncio.sleep(4)
    print('任务还是执行完成了！')

async def main():
    try:
        task = asyncio.create_task(test_coroutine())  # 创建一个异步任务
        await asyncio.wait_for(task, timeout=3)  # 等待受保护的任务完成，超时时间为3秒
    except asyncio.TimeoutError:
        task.cancel()  # 如果超时，则尝试进行取消受保护的任务
        print('超时了，并且尝试进行取消受保护的任务！！')
    except asyncio.CancelledError:
        print('在超时的时候，我们操作了取消任务，但是因为保护机制，这个取消任务被保护起来了，所以不会执行')

    # 挂起，演示我们的任务的取消是无效的！
    print('任务还在执行，我们取消它')
    await asyncio.sleep(8)  # 模拟一个长时间的任务
    print('挂起，演示我们的任务的取消是的！')

asyncio.run(main())  # 运行主函数

任务的超时

对于不然得协程任务，如果我们需要限制相关处理超时的话，我们需要单独的引入asyncio.timeout来进行处理，它是 Python 的 asyncio 模块中的一个函数，它用于在协程（coroutines）中实现超时机制。也就是说当一个协程在指定的超时时间内没有完成时，asyncio.timeout 可以引发一个 asyncio.TimeoutError 异常。

语法

asyncio.timeout(delay, loop=None)

参数

• delay: 一个表示超时时间（秒）的浮点数或者整数。当设置为 None 时，不会设置超时限制。
• loop: 事件循环（event loop）对象。如果未指定，将使用 asyncio.get_event_loop() 获取当前事件循环。

使用示例：

import asyncio


async def my_coroutine():
    # 等待某个操作完成
    await asyncio.sleep(5)

async def main():
    try:
        async with asyncio.timeout(3):
            # 调用 my_coroutine 函数并等待其完成
            await my_coroutine()
    except asyncio.TimeoutError:
        # 操作超时，捕获 TimeoutError 异常并打印提示信息
        print("操作超时")

# 使用 asyncio.run() 函数运行主函数
asyncio.run(main())  # 运行主函数

在上面的代码中我们尝试执行my_coroutine()，并进行设置了一个3秒的超时时间的机制。如果my_coroutine()在3秒内完成，那么一切正常。但是，如果在3秒内未完成，将引发asyncio.TimeoutError异常。所以我们在try-except语句捕获asyncio.TimeoutError异常，并打印出"操作超时"的消息。

注意哦！！asyncio.timeout并不会真正中止或取消你的异步操作。它只是在超时发生时引发异常，如果需要对超时之后进行相关结果出来，需要我们自己根据需要处理这个异常，例如取消正在进行的操作或执行其他逻辑。

任务的并发处理

上面我们看到我们通过手动的创建任务的方式也可以达到并发效果，当然我们的asyncio其实也封装很多其他的方法来进行并发任务处理。

基于async for 循环并发

如果你有一个可迭代的协程对象集合，你可以使用 async for 循环并发地执行这些协程任务。如下示例代码：

import asyncio  # 导入asyncio模块，用于处理异步操作

async def task1():  # 定义一个异步函数task1，用于执行协程任务1
    # 协程任务1的代码
    print("Task 1正在运行")  # 输出协程任务1正在运行的信息
    await asyncio.sleep(1)  # 模拟协程任务1执行1秒
    print("Task 1已完成")  # 输出协程任务1已完成的信息

async def task2():  # 定义一个异步函数task2，用于执行协程任务2
    # 协程任务2的代码  
    print("Task 2正在运行")  # 输出协程任务2正在运行的信息
    await asyncio.sleep(1)  # 模拟协程任务2执行1秒
    print("Task 2已完成")  # 输出协程任务2已完成的信息

async def main():  # 定义一个异步函数main，用于执行主程序代码
    tasks = [task1(), task2()]  # 创建包含task1和task2的列表tasks
    for task in asyncio.as_completed(tasks):  # 遍历每个任务，使用asyncio.as_completed函数获取完成的任务
        await task  # 执行已完成的任务

asyncio.run(main())  # 运行主程序代码，使用asyncio.run函数执行异步代码，也就是跑一个事件循环

上面的代码输出结果如下：

Task 1 is running
Task 2 is running
Task 1 is done
Task 2 is done

上面我们看到我们通过手动的创建任务的方式也可以达到并发效果，当然我们的asyncio其实也封装很多其他的方法来进行并发任务处理，

关于asyncio.gather() 函数前面我们的文章其实也于提到过，如问文中使用的到的案例：

基于asyncio.gather() 函数

比如下面我们的大佬提出的：

import asyncio

async def coroutine_1():
    print("协程1 开始")
    await asyncio.sleep(1)
    print("协程1 结束")

async def coroutine_2():
    print("协程2 开始")
    await asyncio.sleep(2)
    print("协程2 开始")

async def main():
    task1 = asyncio.create_task(coroutine_1())
    task2 = asyncio.create_task(coroutine_2())
    print("批量执行任务 开始")
    # await asyncio.gather(coroutine_1(), coroutine_2())
    await asyncio.gather(task1,task2)
    print("批量执行任务 结束")

asyncio.run(main())

我们的asyncio.gather() 函数接受多个协程对象作为参数，并返回一个新的协程对象，它将并发地执行这些协程任务，并等待它们全部完成。

asyncio.gather错误处理

首先asyncio.gather() 函数时相关的参数项的说明如下：

• *coroutines: 这是最常用的参数，用于指定要并发运行的协程任务。你可以传递一个或多个协程任务对象作为参数。

• return_exceptions: 默认情况下，asyncio.gather() 在所有任务完成后返回一个结果列表，如果其中某个任务抛出异常，则会立即取消所有其他任务并引发该异常。但是，如果将 return_exceptions 设置为 True，则会在任何任务引发异常时继续运行其他任务，并将异常包装成结果列表中的异常对象。这种处理机制对于在我们需要收集所有任务的处理结果而又不需要中断执行的情况下非常有用。

使用示例如下：

import asyncio

import asyncio


async def main():
    try:
        # 设置超时时间为1.5秒
        async def coro(i):

            print(f"开始执行协程 {i}")
            await asyncio.sleep(i)
            print(f"协程 {i} 执行完毕")
            return i

        async def foo():
            await asyncio.sleep(1)
            return 'foo'

        async def bar():
            await asyncio.sleep(2)
            raise ValueError('bar error')

        results = await asyncio.gather(foo(), bar(), coro(1), return_exceptions=True)
        print(results)
    except asyncio.TimeoutError:
        print('Timeout occurred')


asyncio.run(main())

输出的结果如下：

开始执行协程 1
协程 1 执行完毕
['foo', ValueError('bar error'), 1]

asyncio.gather并发任务的取消操作

合并任务的全员取消，示例代码如下：

import asyncio

async def task1():
    try:
        print('任务1开始。')
        await asyncio.sleep(5)
        print('任务1完成。')
    except asyncio.CancelledError:
        print('任务1被取消。')

async def task2():
    try:
        print('任务2开始。')
        await asyncio.sleep(3)
        print('任务2完成。')
    except asyncio.CancelledError:
        print('任务2被取消。')

async def main():
    tasks = [task1(), task2()]
    gather_task = asyncio.gather(*tasks)

    await asyncio.sleep(2)  # 等待2秒后取消任务
    # 全部任务都取消
    gather_task.cancel()

    try:
        await gather_task
    except asyncio.CancelledError:
        print('合并任务被取消。')

asyncio.run(main())

输出结果如下：

任务1开始。
任务2开始。
任务1被取消。
任务2被取消。
合并任务被取消。

单独取消某个合并中的任务，示例代码如下：

import asyncio

async def task1():
    try:
        print('任务1开始。')  # 任务1开始
        await asyncio.sleep(5)  # 等待5秒
        print('任务1完成。')  # 任务1完成
    except asyncio.CancelledError:
        print('任务1被取消。')  # 任务1被取消

async def task2():
    try:
        print('任务2开始。')  # 任务2开始
        await asyncio.sleep(3)  # 等待3秒
        print('任务2完成。')  # 任务2完成
    except asyncio.CancelledError:
        print('任务2被取消。')  # 任务2被取消

async def main():
    # 创建 Task 对象
    task1_obj = asyncio.create_task(task1())  # 任务1对象
    task2_obj = asyncio.create_task(task2())  # 任务2对象

    await asyncio.sleep(2)  # 等待2秒后取消任务
    task1_obj.cancel()  # 取消任务1

    try:
        await asyncio.gather(task1_obj, task2_obj)  # 同时等待任务1和任务2完成
    except asyncio.CancelledError:
        print('合并任务被取消。')  # 合并任务被取消

asyncio.run(main())  # 运行主函数

输出结果如下：

任务1开始。
任务2开始。
任务1被取消。
任务2完成。

基于 asyncio.wait 函数

asyncio.wait的参数说明如下：

asyncio.wait(aws, *, loop=None, timeout=None, return_when=ALL_COMPLETED)

• aws：一个可迭代的协程对象列表（或任何可迭代对象），表示要等待的任务列表。
• loop：可选参数，指定要使用的事件循环。如果未提供，则使用默认的事件循环。
• timeout：可选参数，指定等待的最长时间（以秒为单位）。如果超时时间到达而任务还未完成，则等待函数会返回。
• return_when：可选参数，指定返回的条件。可以选择以下值之一：
• FIRST_COMPLETED：如果有任何一个任务完成，则立即返回。
• FIRST_EXCEPTION：如果有任何一个任务抛出异常，则立即返回。
• ALL_COMPLETED：如果所有任务都完成，则返回。

最简单示例：

import asyncio

async def coro(i):
    print(f"开始执行协程 {i}")
    await asyncio.sleep(i)
    print(f"协程 {i} 执行完毕")
    return i

async def main():
    tasks = [asyncio.create_task(coro(1)), asyncio.create_task(coro(2)), asyncio.create_task(coro(3))]
    done, pending = await asyncio.wait(tasks)
    for task in done:
        print(f"协程 {task.result()} 完成")

asyncio.run(main())

需要注意上面的代码中我们需要的list列表对象包含的是一个asyncio.create_task的任务的对象，并将任务列表传递给 asyncio.wait()。这个函数会挂起当前任务，等待所有任务完成。上面的代码执行的结果会等待我们的所有的任务的执行完成后会再统一的进行输出。上面输出的结果为：

开始执行协程 1
开始执行协程 2
开始执行协程 3
协程 1 执行完毕
协程 2 执行完毕
协程 3 执行完毕
协程 1 完成
协程 2 完成
协程 3 完成

超时示例：

import asyncio

async def coro(i):
    print(f"开始执行协程 {i}")
    await asyncio.sleep(i)
    print(f"协程 {i} 执行完毕")
    return i

async def main():
    tasks = [asyncio.create_task(coro(1)), asyncio.create_task(coro(2)), asyncio.create_task(coro(3))]
    done, pending = await asyncio.wait(tasks, timeout=2, return_when=asyncio.FIRST_COMPLETED)
    for task in done:
        print(f"协程 {task.result()} 完成")

asyncio.run(main())

再上面的代码加入某个任务的执行超过等待的时间。

等待所有任务完成，如果有任务抛出异常，则立即返回

import asyncio

async def coro(i):
    if i == 2:
        raise ValueError("协程任务错粗错误~~~~~任务抛出异常")
    print(f"开始执行协程 {i}")
    await asyncio.sleep(i)
    print(f"协程 {i} 执行完毕")
    return i

async def main():
    tasks = [asyncio.create_task(coro(1)), asyncio.create_task(coro(2)), asyncio.create_task(coro(3))]
    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_EXCEPTION)
    for task in done:
        if task.exception() is not None:
            print(f"协程 {task.get_coro().__name__} 抛出异常: {task.exception()}")
        else:
            print(f"协程 {task.result()} 完成")

asyncio.run(main())

上述代码执行输出的结果如下：

开始执行协程 1
开始执行协程 3
协程 coro 抛出异常: 协程任务错误~~~~~任务抛出异常

超时和错误结合示例

import asyncio

async def coro(i):
    if i == 2:
        raise ValueError("协程错误~~任务抛出异常")
    print(f"开始执行协程 {i}")
    await asyncio.sleep(i)
    print(f"协程 {i} 执行完毕")
    return i

async def main():
    tasks = [asyncio.create_task(coro(1)), asyncio.create_task(coro(2)), asyncio.create_task(coro(3))]
    done, pending = await asyncio.wait(tasks, timeout=2, return_when=asyncio.FIRST_EXCEPTION)
    for task in done:
        if task.exception() is not None:
            print(f"协程 {task.get_coro().__name__} 抛出异常: {task.exception()}")
        else:
            print(f"协程 {task.result()} 完成")

    for task in pending:
        print(f"协程 {task.get_coro().__name__} 未完成")

asyncio.run(main())

输出结果如下：

开始执行协程 1
开始执行协程 3
协程 coro 抛出异常: 任务抛出异常
协程 coro 未完成
协程 coro 未完成

真实业务逻辑中，上面的代码会存在以下的可能的几种情况，

• 如果在超时时间内有任务完成（即有任务变为已完成状态），那么done集合将包含这些已完成的任务，而pending集合将包含未完成的任务。

• 如果在超时时间内没有任务完成，但至少有一个任务抛出了异常，那么done集合将包含这个抛出异常的任务，而pending集合将包含未完成的任务。

• 如果在超时时间内没有任务完成，也没有任务抛出异常，那么done集合将为空，而pending集合将包含所有的任务。

归纳总结asyncio.wait和asyncio.gather

对于任务的并发执行，我们通常我们同步的处理过程中，一般是通过多线程的方式的，而再异步编程过程中，对于批量任务的并发处理，我们也可以实现类似的多线程的并发，如前面的示例中我们的使用到了asyncio.gather这方法。另外的asyncio还提供的另一种并发任务的执行的方法，asyncio.wait。那么asyncio.wait和asyncio.gather他们都是用于等待多个异步任务的完成，那么它们之间有哪些些区别的呢？（需主要一点，asyncio.gather和asyncio.wait本身也是一个协程任务哦）如下说明所示：

• asyncio.wait函数返回一个Future对象的集合，表示一组任务。可以通过在协程中使用await关键字等待这个集合的完成。在等待期间，可以使用asyncio.wait的return_when参数来指定返回的条件。常见的返回条件有三种：

  通过asyncio.wait返回的集合中，可以使用done和not_done属性分别获取已完成的任务和未完成的任务。

• asyncio.FIRST_COMPLETED：只要任意一个任务完成，就立即返回。
• asyncio.FIRST_EXCEPTION：只要发生任意一个任务的异常，就立即返回。
• asyncio.ALL_COMPLETED：等待所有任务都完成后返回。

• asyncio.gather函数用于同时启动和等待多个协程任务的完成。它接收一组协程对象作为参数，并返回一个Future对象。asyncio.gather的参数可以是任意数量的协程，也可以是协程的可迭代对象（如列表）。返回的Future对象的结果是一个列表，包含了所有协程任务的结果。

  与asyncio.wait不同，asyncio.gather一般会等待所有的任务完成，除非我们需要显式指定取消(cancel)或忽略(return_exceptions)。

所以我们可以得出以下的几点结论：

• asyncio.wait用于等待一组任务的完成，可以通过不同的返回条件来控制返回时机。
• asyncio.gather用于同时启动多个协程任务并等待全部完成，返回一个包含所有任务结果的列表。

asyncio.TaskGroup任务组

在Python 3.10中，引入了asyncio.TaskGroup类作为asyncio模块的新功能之一。asyncio.TaskGroup用于管理一组相关的任务，并提供了一种更高级的方式来并发地执行这些任务。

asyncio.TaskGroup的主要特点和应用场景如下：

• 任务管理: TaskGroup允许同时创建和管理多个Task对象。你可以使用create_task()方法将任务添加到TaskGroup中，并使用cancel_all()方法取消所有的任务。这使得任务的管理更加方便和集中化。

• 并发执行: 通过使用async with语句创建TaskGroup对象，你可以在同一时间并发地执行多个任务。它们会在事件循环中以非阻塞的方式运行，实现非阻塞的并发执行，从而提高程序的吞吐量和性能。

• 超时控制: 使用asyncio.wait_for()函数结合TaskGroup，你可以设置总体超时时间来限制任务的执行时间。如果任务没有在指定的时间内完成，将引发asyncio.TimeoutError异常，你可以在异常处理块中对超时进行处理。

• 结果处理: 可以使用async for循环遍历TaskGroup对象，以获取每个任务的结果。这样，你可以轻松地处理任务的返回值，对结果进行进一步的处理或输出。

• 协程组合: TaskGroup可以与其他asyncio的协程相关类和函数一起使用，例如gather()、as_completed()等。它们可以配合使用，以实现更复杂的并发行为和任务调度。

应用示例：

import asyncio

async def fetch_request(url):
    # 模拟一个耗时操作
    await asyncio.sleep(2)
    return f"返回的处理结果 {url}"

async def main():
    async with asyncio.TaskGroup() as group:
        tasks = [
            group.create_task(fetch_request("https://xxxx.example.com/大爷1")),
            group.create_task(fetch_request("https://xxxx.example.com/大爷2")),
            group.create_task(fetch_request("https://xxxx.example.com/大爷3"))
        ]

        # 设置总体超时时间为 3 秒
        try:
            # 等待同时执行多个任务，超时时间为3秒
            await asyncio.wait_for(asyncio.gather(*tasks), timeout=3)
        except asyncio.TimeoutError:
            # 如果超时则打印操作超时并返回
            print("操作超时")
            return

        # 遍历任务列表
        for task in tasks:
            # 如果任务已完成
            if task.done():
                # 获取任务的结果
                result = task.result()
                # 打印任务结果
                print(result)


asyncio.run(main())

输出结果如下：

返回的处理结果 https://xxxx.example.com/大爷1
返回的处理结果 https://xxxx.example.com/大爷2
返回的处理结果 https://xxxx.example.com/大爷3

asyncio.to_thread之单独的线程中异步运行函数

我们知道，在 asyncio 中，事件循环是并发执行的核心，它负责调度和执行任务。当事件循环运行一个任务时，如果该任务执行了阻塞操作，那么事件循环将无法继续执行其他任务，这会导致性能下降。通过将阻塞操作放到单独的线程中，事件循环可以在等待 I/O 完成的同时继续执行其他任务，从而提高整体性能。

因此在3.9版本中asyncio.to_thread 是 asyncio 模块中的一个函数，它用于将一个协程（coroutine）转换为一个在单独线程中运行的线程任务的机制。通过这样我们可以允许协程执行阻塞操作（例如，I/O密集型操作或长时间运行的任务），而不阻塞整个事件循环。

需要了解，asyncio.to_thread是把阻塞的同步代码转化为异步的线程池执行模式。同步的任务是会被放到一个线程池中执行，然后返回一个协程对象，这个协程对象可以被加入到事件循环中执行。

由上可知我们的asyncio.to_thread可以用于以下一些情况：

• 同步代码转异步: 对于一些同步的阻塞式代码，如果要将其转化为异步非阻塞的代码，通常需要对代码进行重构。使用asyncio.to_thread可以简化这个过程，只需要将同步代码封装为一个函数，然后调用asyncio.to_thread即可。

• 对外部库的支持: 对于某些外部库中没有提供异步接口的函数，可以将其包装成异步函数，然后使用asyncio.to_thread将其放到线程池中执行。这样既可以享受到异步编程的优势，又可以使用现有的库函数。

• asyncio.to_thread 返回的是一个 协程的Task 对象，它表示在单独线程中运行的协程。所以我们可以使用 await 来等待这个任务完成。

• asyncio.to_thread 创建的线程任务是独立于事件循环的，它们在单独的线程中运行，不会影响事件循环的正常工作。

如它源码所示：

async def to_thread(func, /, *args, **kwargs):
    """Asynchronously run function *func* in a separate thread.

    Any *args and **kwargs supplied for this function are directly passed
    to *func*. Also, the current :class:`contextvars.Context` is propagated,
    allowing context variables from the main thread to be accessed in the
    separate thread.

    Return a coroutine that can be awaited to get the eventual result of *func*.
    """
    loop = events.get_running_loop()
    ctx = contextvars.copy_context()
    func_call = functools.partial(ctx.run, func, *args, **kwargs)
    return await loop.run_in_executor(None, func_call)

如应用的官方示例所示：

import asyncio
import time


def blocking_io():
    print(f"start blocking_io at {time.strftime('%X')}")
    # Note that time.sleep() can be replaced with any blocking
    # IO-bound operation, such as file operations.
    time.sleep(1)
    print(f"blocking_io complete at {time.strftime('%X')}")

async def main():
    print(f"started main at {time.strftime('%X')}")

    await asyncio.gather(
        # 在新的线程中执行阻塞的任务
        asyncio.to_thread(blocking_io),
        #  定义一个协程
        asyncio.sleep(1))

    print(f"finished main at {time.strftime('%X')}")


asyncio.run(main())

最终输出结果如下所示：

started main at 17:38:10
start blocking_io at 17:38:10
blocking_io complete at 17:38:11
finished main at 17:38:11

文笔有限，如有笔误或错误！欢迎批评指正！感谢各位大佬！有什么问题也可以随时交流！

结尾

END

简书：https://www.jianshu.com/u/d6960089b087

掘金：https://juejin.cn/user/2963939079225608

公众号：微信搜【程序员小钟同学】

新开QQ群号，欢迎随时加群交流，相互学习。QQ群号：247491107

小钟同学 | 文 【欢迎一起学习交流】| QQ：308711822