ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 数据库 > 数据库开发技术 > 深入FastAPI异步编程-之并发编程（9）-asyncio.as_completed的补充说明

深入FastAPI异步编程-之并发编程（9）-asyncio.as_completed的补充说明

原创 数据库开发技术 作者：程序员小钟同学 时间：2024-01-18 17:21:52 0 删除 编辑

asyncio.as_completed 的并发任务阻断

在前面中，我以及讨论过关于asyncio.gather处理并发的问题，也粗略分析它在使用的上可能遇到的一些问题。

在前面我们知道asyncio.gather是需要进行等待任务组的执行完成才会返回。那有可能有些时候，我们需要并发的执行任务，但是又不希望一个一个等待任务的完成后才进行处理,此时我们就也用上asyncio.as_completed()。

asyncio.as_completed()和asyncio.gather（）的优势上就是我们不需要等待所有的任务组都处理完成就可以处理，我们一个一个的迭代检索一下在asyncio.as_completed()执行的任务是否已完成。也就是说我们的as_completed() 会返回一个future的迭代器，你可以按需迭代这个future迭代器来获取完成的事件循环中的协程。这意味着你可以根据事件循环的进度来处理结果，而不需要等待所有协程完成，这对于资源密集型或长时间运行的任务特别有用。

通常如果我们的业务逻辑中有一个依赖其他协程结果的任务时，可以使用 as_completed() 来确保这些依赖的结果一旦准备好就能立即处理。如下的示例：

import asyncio
async def async_function1():
    await asyncio.sleep(5)
    return "第一个任务需要5秒的时间，所以会最后输出"
async def async_function2():
    await asyncio.sleep(2)
    return "第二个任务只需要2秒的时间，所以会先输出"
async def main():
    tasks = [async_function1(), async_function2()]
    # 使用 as_completed() 获取完成的协程
    for task in asyncio.as_completed(tasks):
        result = await task  # 等待任务完成并获取结果
        print(result)  # 立即处理结果
asyncio.run(main())

执行后输出的结果为：

第二个任务只需要2秒的时间，所以会先输出
第一个任务需要5秒的时间，所以会最后输出

从输出结果看，当我们的任务2执行完成后不会等待1的处理完成就会返回了！所有可以很好处理有并发需求，但是又不需要等待所有并发任务都完成的时候才返回的情况。

如果我们存在一个并发的任务，当有一个完成后就取消其他的任务的。如下的一个阻断其他任务运行的一个示例：

import asyncio


async def async_function1():
    was_cancelled = False
    try:
        # 这里执行一些异步操作
        await asyncio.sleep(1)
    except asyncio.CancelledError:
        # 捕获 CancelledError 异常
        was_cancelled = True
    finally:
        # 这里执行任何需要在任务结束时执行的代码
        if was_cancelled:
            print("任务1以及被取消")
        else:
            print("任务1处理完成")

    return "任务1已完成"


async def async_function2():
    was_cancelled = False
    try:
        # 这里执行一些异步操作
        await asyncio.sleep(2)
    except asyncio.CancelledError:
        # 捕获 CancelledError 异常
        was_cancelled = True
    finally:
        # 这里执行任何需要在任务结束时执行的代码
        if was_cancelled:
            print("任务2以及被取消")
        else:
            print("任务2处理完成")

    return "任务2已完成"


async def cancel_other_tasks(finished_task):
    #  如果 finished_task 包含在任务组内，则取消其他任务
    tasks_to_cancel = [task for task in renwugor if task != finished_task]
    for task in tasks_to_cancel:
        #调用任务的cancel()方法。这个方法会抛出一个asyncio.CancelledError异常到协程中
        task.cancel()

renwugor = []


async def main():
    # 给任务命名
    task1 = asyncio.create_task(async_function1(), name="task1")
    task2 = asyncio.create_task(async_function2(), name="task2")
    # 使用 as_completed() 来迭代任务
    renwugor.append(task1)
    renwugor.append(task2)
    for task in asyncio.as_completed(renwugor):
        # 等待任务完成
        finished_task = await task
        # 检查任务结果并处理
        await cancel_other_tasks(finished_task)
        # print(f"{finished_task.get_name()} finished")

asyncio.run(main())

运行上面的代码，我们输出的结果为:

任务1处理完成
任务2以及被取消

注意：取消一个协程并不意味着它会立即停止。协程必须在执行到一个可以安全取消的点时才能响应取消请求。例如，如果协程正在执行无法中断的I/O操作，那么即使调用了cancel()，协程也可能无法立即停止。

另外：但在 Python 3.8 及更高版本中，推荐使用 asyncio.create_task() 函数代替 asyncio.ensure_future() 函数。

番外篇ThreadPoolExecutor()中的的as_completed。

在同步的线程并发中，ThreadPoolExecutor也有类似的提高用于迭代返回已完成的 Future 对象的的as_completed方法。它同样的也是用于处理一组并发任务，并在给定的超时时间内等待任务完成。如下示例代码所示：

import concurrent.futures
import time

def task(n):
    # 模拟耗时操作
    time.sleep(n)
    return n

def main():
    # 创建 ThreadPoolExecutor 对象
    executor = concurrent.futures.ThreadPoolExecutor()

    # 提交任务到线程池
    futures = [executor.submit(task, i) for i in range(1, 6)]

    # 使用 as_completed 迭代已完成的 Future 对象
    for future in concurrent.futures.as_completed(futures, timeout=3):
        try:
            result = future.result()
            print(f'任务： {result} completed')
        except concurrent.futures.TimeoutError:
            print('超时处理)

if __name__ == '__main__':
    main()

上面代码中我们通过创建线程池执行器（ThreadPoolExecutor），并将一组任务提交到线程池里面进行处理。然后，接着我们使用 as_completed 函数迭代已完成的 Future 对象，并设置超时时间为 3 秒。

最终我们可以看到，如果某个任务在超时时间内完成，我们通过 future.result() 获取任务的结果，并打印任务完成的消息。如果超时时间内没有任务完成，将抛出 TimeoutError 异常。

asyncio.as_completed 的并发任务的超时处理

asyncio.as_completed中提供了一个timeout的参数，它主要用于表示，设定任务组执行的超时时间，当任务组的执行耗时超过了设定的时间，则会引发超时。如下示例代码所示：

import asyncio


async def async_function1():
    was_cancelled = False
    try:
        # 这里执行一些异步操作
        await asyncio.sleep(2)
    except asyncio.CancelledError:
        # 捕获 CancelledError 异常
        was_cancelled = True
    finally:
        # 这里执行任何需要在任务结束时执行的代码
        if was_cancelled:
            print("任务1以及被取消")
        else:
            print("任务1处理完成")

    return "任务1已完成"


async def async_function2():
    was_cancelled = False
    try:
        # 这里执行一些异步操作
        await asyncio.sleep(3)
    except asyncio.CancelledError:
        # 捕获 CancelledError 异常
        was_cancelled = True
    finally:
        # 这里执行任何需要在任务结束时执行的代码
        if was_cancelled:
            print("任务2以及被取消")
        else:
            print("任务2处理完成")

    return "任务2已完成"


async def cancel_other_tasks(finished_task):
    # 如果 finished_task 包含在任务组内，则取消其他任务
    tasks_to_cancel = [task for task in renwugor if task != finished_task]
    for task in tasks_to_cancel:
        task.cancel()

renwugor = []


async def main():
    # 给任务命名
    task1 = asyncio.create_task(async_function1(), name="task1")
    task2 = asyncio.create_task(async_function2(), name="task2")
    # 使用 as_completed() 来迭代任务
    renwugor.append(task1)
    renwugor.append(task2)
    for task in asyncio.as_completed(renwugor,timeout=1):
        # 等待任务完成
        try:
            finished_task = await task
            # 检查任务结果并处理
            await cancel_other_tasks(finished_task)
        except asyncio.TimeoutError:
            print("任务超时")
        except asyncio.CancelledError:
            print("任务被取消")

asyncio.run(main())

通过运行代码观察输出的结果：

任务超时
任务超时
任务1以及被取消
任务2以及被取消

结合aiohttp的应用示例： 

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as response:
        return response.status

async def main():
    urls = [
        'https://www.baidu.com',
        'https://www.qq.com',
        'https://www.hao123.com'
    ]

    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        # 基于同一个回话并发执行任务
        for coro in asyncio.as_completed(tasks):
            try:
                result = await coro
                print(f'相应结果 from {result}...')
            except Exception as e:
                print(f'Error: {e}')

asyncio.run(main())

限量并发的使用

我们还可以使用了信号量 semaphore 来限制同时运行的任务数量，并使用 asyncio.as_completed 来获取已完成的任务。如下示例代码：

import asyncio

async def task():
    # 模拟一个耗时操作
    await asyncio.sleep(2)
    return '任务完成'

async def main():
    sem = asyncio.Semaphore(5)  # 同时运行的任务数量限制为5
    async def limited_task():
        async with sem:
            return await task()

    for future in asyncio.as_completed([limited_task() for _ in range(10)]):
        result = await future
        print(result)

asyncio.run(main())

下小结我们再有针对性展开关于asyncio.wait的一些其他的补充说明。

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

结尾

END

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

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

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

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

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


原文作者：程序员小钟同学；如有侵权，请联系管理员删除。

0
0
上一篇： 深入FastAPI异步编程-之并发编程（11）-异步中的?同步原语
下一篇： 深入FastAPI异步编程-之并发编程（10）-?asyncio.wait 更精细化并发补充说明
请登录后发表评论 登录
全部评论
程序员小钟同学

注册时间：2021-02-20

  • 博文量
    224
  • 访问量
    111699

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593