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

首页 > 数据库 > 数据库开发技术 > 深入FastAPI异步编程-之并发编程（5）~Future对象

深入FastAPI异步编程-之并发编程（5）~Future对象

原创 数据库开发技术 作者：程序员小钟同学 时间：2024-01-08 00:00:00 19 删除 编辑

Future

在前面我们介绍了关于Task任务，而在Python的asyncio库中我们的Task的父类来自于Future，而Future是什么呢？它既然是Task的父类，自然也包含了Task的一些场景的功能。

下面我们介绍一下Future。Future用于表示一个异步操作的最终结果，代表着一个未来的值。它可以在异步调用完成后被设置为一个值或者一个异常。Future对象可以被等待，以便获取最终结果。

Future可以处于以下几种状态之一：

  • pending（等待状态）
  • running（运行状态）
  • done（完成状态） cancelled（取消状态）

这些状态反映了异步操作的生命周期，通常我们在编写代码的话，不会直接定义它。下面我的简单创建一下Future对象并进行相关状态切换，如下示例所示：


import asyncio

async def main():
    # 创建一个Future对象
    future = asyncio.Future()

    # 检查Future对象的状态
    print(f"刚创建的时候Future's state: {future}")

    # 将Future对象设置为已完成状态
    future.set_result('设置为已完成状态')

    # 检查Future对象的状态
    print(f"设置值后Future's state: {future}")

    # 尝试将已完成的Future对象再次设置为已完成状态
    try:
        future.set_result('对象再次设置为已完成状态')
    except Exception as e:
        print(f"已经存在值，无法再次设置Error: {e}")

        # 检查Future对象的状态
    print(f"取消前的Future's state: {future}")
    # 将Future对象设置为已取消状态
    future.cancel()

    # 检查Future对象的状态
    print(f"取消后的Future's state: {future}")

asyncio.run(main())

运行代码输出的结果为：

刚创建的时候Future's state: <Future pending>
设置值后Future's state: <Future finished result='设置为已完成状态'>
已经存在值，无法再次设置Error: invalid state
取消前的Future's state: <Future finished result='设置为已完成状态'>
取消后的Future's state: <Future finished result='设置为已完成状态'>

在上面的示例代码中，我们首先创建了一个 Future 对象，它的初始状态是 pending。接着我们使用 set_result 方法将 Future 对象设为 successful 状态并提供了结果 '设置为已完成状态'。然后我们尝试再次使用 set_result 来设置一个已经完成的 Future 对象，这会引发一个 InvalidStateError，因为 Future 对象一旦完成（无论是通过 set_result还是 set_exception）就不能再被改变。最后，我们调用 cancel 方法将 Future 对象的状态设置为 cancelled。

我们还可以给我们的future添加关于回调处理，如下示例代码所示：



async def main():
    # 创建一个Future对象
    future = asyncio.Future()

    # 添加一个完成回调，当Future对象完成时执行的回调函数，打印Future对象的结果
    future.add_done_callback(lambda f: print(f"Future结果：{f.result()}"))

    # 模拟异步操作，并设置Future的结果
    await asyncio.sleep(1)
    future.set_result("Hello, Future!")


# 运行主协程
asyncio.run(main())

在上面的代码中：

  • 我们通过future = asyncio.Future() 创建了一个Future对象。

  • 然后接着我们通过future.add_done_callback(lambda f: print(f"Future结果：{f.result()}")) 添加了一个完成回调。这个回调函数将在Future对象完成时执行，不管它是成功还是失败。这里使用lambda表达式创建了一个匿名函数，这个函数接收future作为参数，并打印出它的结果。

  • 再然后我们通过await asyncio.sleep(1) 模拟了一个异步操作。asyncio.sleep是一个异步函数，它将暂停当前协程的执行1秒钟。await关键字用于挂起当前协程，直到被挂起的协程完成。

  • 接着我们继续进行future.set_result("Hello, Future!") 设置了Future的结果。这通常是在异步操作实际完成时由异步库或框架自动调用的。在这里，我们手动调用它，模拟Future完成了异步操作，并直接滴返回了结果"Hello, Future!"。

在实际的异步编程中，Future对象通常不需要手动创建和设置结果。asyncio提供了更高级的抽象，如我们前面说的协程或任何。比如async和await，以及asyncio.create_task来处理异步任务，这些通常更加方便和高效。

需注意的地方点，假如我们需要在取消 Future 对象的时候执行一些特定操作处理的话，需要经过特殊机制进行在考虑在取消 Future 对象之前手动抛出一个异常，并在相应的取消回调函数中捕获该异常来获取取消原因，再进行处理。如下示例代码所示：

import asyncio


async def main():
    # 创建一个Future对象
    future = asyncio.Future()

    # 添加一个完成回调，当Future对象完成时执行的回调函数，打印Future对象的结果
    future.add_done_callback(lambda f: print(f"Future结果：{f.result()}"))

    # 定义取消回调函数
    def cancel_callback(f):
        try:
            # 在取消回调中手动抛出异常
            raise asyncio.CancelledError("Future已取消")
        except asyncio.CancelledError as e:
            # 捕获取消异常并打印取消原因
            print(f"Future已取消，原因：{str(e)}")

    # 添加取消回调
    future.add_done_callback(cancel_callback)

    # 模拟异步操作，并设置Future的结果
    await asyncio.sleep(1)
    future.set_result("Hello, Future!")

    # 取消Future对象
    future.cancel()


# 运行主协程
asyncio.run(main())

上面的代码输出结果为：

Future结果：Hello, Future!
Future已取消，原因：Future已取消

我们思考一下假如我们没有设置相关Future对象的结果就开始等待的话，会出现什么情况呢？如下示例代码：

import asyncio

async def main():
    # 获取当前事件循环
    loop = asyncio.get_running_loop()
    # 创建future对象
    futtest = loop.create_future()
    print("创建future对象")

    # 等待任务最终结果
    print("等待任务最终结果~~~")
    result = await futtest
    print("任务最终结果输出~~~",result)


asyncio.run(main())

执行上面的代码后你会发现我们的一直等待着阻塞往下执行！！！！futtest的结果，那么就会一直等待。我们修改代码后如下：

import asyncio


async def sleeper(futtest):
    # 等待一段时间，模拟异步操作
    await asyncio.sleep(1)
    # 设置future的结果
    futtest.set_result("在另外的协程中，设置future的结果")


async def main():
    # 获取当前事件循环
    loop = asyncio.get_running_loop()
    # 创建future对象
    futtest = loop.create_future()
    print("创建future对象")

    # 使用另一个协程来设置future的结果
    asyncio.create_task(sleeper(futtest))

    # 等待任务最终结果
    print("等待任务最终结果~~~")
    result = await futtest
    print(result)
    print("任务最终结果输出~~~",result)


asyncio.run(main())

最终会输出如下的结果：

创建future对象
等待任务最终结果~~~
在另外的协程中，设置future的结果
任务最终结果输出~~~ 在另外的协程中，设置future的结果

在上面的代码中我们我们定义了一个新的协程 sleeper，它接受一个 Future 对象作为参数，并在其中进行异步操作，最后调用 set_result(result) 方法来设置结果，最终才可以不会一直阻塞着等待。

Future 和 任务

通常我们创建了一个任务的时候，就已经包含了创建Future。也就是说我们的Task包含了一个空的Future，当任务执行结束的时候，我们的Future会被进行值的设置。如下的示例所示：

import asyncio

async def my_coroutine():
    await asyncio.sleep(2)  # 模拟异步操作
    return '协程执行结果完成'

async def main():
    task = asyncio.create_task(my_coroutine())  # 创建任务并自动创建Future对象
    # 返回两个Future的集合：(done, pending)。其中done集合是已经完成的Future对象的集合，pending集合是尚未完成的Future对象的集合。
    done, pending = await asyncio.wait({task})  # 等待任务完成，并获取已完成的Task集合
    for task in pending:
        future = task.result()  # 获取与任务关联的Future对象
        print('pending',future)  # 打印Future对象的结果
    for task in done:
        future = task.result()  # 获取与任务关联的Future对象
        print('done',future)  # 打印Future对象的结果

asyncio.run(main())


在上面的代码中：

  • 首先我们定义了一个异步协程my_coroutine。在这个协程内部中，使用await asyncio.sleep(2)来模拟一个异步操作，当完成后return '协程执行结果完成' 也是说当my_coroutine协程完成时，它将返回一个字符串结果。

  • 接着我们通过task = asyncio.create_task(my_coroutine()) 使用asyncio.create_task来创建一个任务。然后传入到asyncio.wait({task})中，这里我们使用asyncio.wait来等待一个任务集合中的所有任务完成。其中我们的asyncio.wait接受一个任务集合作为参数，并返回两个集合：done和pending。

    • done集合包含已经完成的任务，
    • pending集合包含尚未完成的任务。

  • 而对于future = task.result() 则是用于获取与任务关联的Future对象的结果。这个方法会阻塞，直到Future对象有结果可用。

应用示例

常见的一个应用示例：

import asyncio
import aiohttp


async def fetch_page(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.text()


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

    # 创建多个Future对象，用于处理每个网站的HTML页面内容
    futures = [asyncio.ensure_future(fetch_page(url)) for url in urls]

    # 等待多个Future对象完成
    pages = await asyncio.gather(*futures)

    # 把页面内容信息和地址，并将它们打包成二元组 (city, population)打印每个网站的页面内容 用于遍历 url 和 page 列表，
    for url, page in zip(urls, pages):
        print(f"{url} 页面内容：{page[:100]}...")


# 运行主协程
asyncio.run(main())

在上面的代码中，我们首先定义了一个 fetch_page() 函数，其参数是一个URL。

  • 在该函数内部，使用 aiohttp.ClientSession() 创建一个异步HTTP客户端会话对象 session。然后，使用 session.get(url) 发起异步GET请求，并使用 await resp.text() 获取响应的文本内容。

  • 接着我们使用列表推导式创建了一个包含多个 fetch_page(url) 函数调用的 Future 对象列表 futures。

  • 再让通过asyncio.ensure_future() 函数用于将 fetch_page(url) 转换为一个 Future 对象。

  • 使用 asyncio.gather(*futures) 等待多个 Future 对象完成，并将它们的结果以列表的形式返回，赋值给 pages 变量。这里使用 *futures 将 futures 列表进行解包

  • 最后，使用 zip(urls, pages) 迭代 urls 和 pages 列表，将每个网站的URL和对应的页面内容一一配对，然后使用 print() 函数打印每个网站的URL和前100个字符的页面内容。

在上面的代码中，我们主要涉及一个代码段如下：

# 创建多个Future对象，用于处理每个网站的HTML页面内容
    futures = [asyncio.ensure_future(fetch_page(url)) for url in urls]

从字面上看，我们可以得知asyncio.ensure_future() 函数是将一个协程对象转换为一个 Future 对象来开看待。

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

结尾

END

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

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

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

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

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


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

0
0
上一篇： 深入FastAPI异步编程-之并发编程（7）番外篇~Windows系统下事件循环插曲
下一篇： 深入FastAPI异步编程-之并发编程（4）~任务、任务并发、asyncio.wait、asyncio.gather 等
请登录后发表评论 登录
全部评论
程序员小钟同学

注册时间：2021-02-20

  • 博文量
    224
  • 访问量
    111698

最新文章

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

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

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

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