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

首页 > 数据库 > 数据库开发技术 > 基于FastAPI框架从零开始搭建权限管理系统--02异步携程篇

基于FastAPI框架从零开始搭建权限管理系统--02异步携程篇

原创 数据库开发技术 作者：程序员小钟同学 时间：2023-11-01 15:36:02 0 删除 编辑

前言

在上一篇中，我们已简单介绍了接下来我们后台权限管理系统所需的一切前端和后端使用到的相关资源信息。在正式步入到后端相关框架编写之前，为了让一些刚入门朋友了解所谓的异步编程的相关特性。本章节，我主要围绕这这个异步和协程这些知识点，再啰嗦多一点。

所谓的异步编程

老话长谈的异步编程无非只是一种对‘编程模式’的定义。以往我们编写相关程序的代码的时候，使用都是一种同步的方式来编写，相关的逻辑代码都是顺序一步一步的执行，即便是遇到相关的IO阻塞等耗时的任务，也必须等待阻塞操作的逻辑的执行的完成，才可以继续往下执行。 那这样等待阻塞任务的这种也需要等待的问题上，相关应对的解决方案就是，我们所谓的异步编程了。

在异步编程里面，前面我们提到的耗时操作，比如IO任务等，通常会在它遇到阻塞的时候，把它放单独拿出来，放到一个单独的线程或进程中去执行处理，这样我们的主线程也就可以继续执行其他操作。

然后当耗时操作完成后，异步编程会香坊设置通过回调函数或其他的方式，通知到原来我们程序运行的主线程中，并告知已操作已完成，接着我们的主线程可以获取结果并进行处理。

好比：高速公路只有一条高速路的情况下，车来车往的，所有的车辆的行驶，都是顺序的排队前行，无法弯道超车。当前车没有油需要加油的时候，我们不可能等前面的车加满油才让后面的车继续前行，所以只能让需要加油的车跑到服务区里面加油先，加完再想办法回到高速路上。

在异步编程中也是一样的道理。耗时的操作就像等缺由了需要加油的汽车一样，会阻塞程序的执行。而异步编程就是让这些耗时的操作在另外的线程或进程中执行，不会阻塞主线程。主线程可以继续执行其他操作，等到耗时操作完成时，通过回调函数或其他方式通知主线程。这样就提高了程序的执行效率和吞吐量，也应对了可能耗时的操作，比如网络请求、文件读写、数据库访问等等IO密集型操作。在python中目前我们最常示使用的异步编程方式也即是我们的今天的主角asyncio了。

asyncio异步I/O框架

关于asyncio它是Python标准库中的一个异步I/O框架，自从Python 3.4版本开始引入。它核心观念主要主要有：

  • 协程（coroutine）
  • 事件循环（event loop）

通过它们来实现相关异步代码逻辑处理,使得开发者可以使用简单直观的语法编写高效的异步代码.

如我们前面提到的，在传统的同步I/O编程模型中，通常我们发起一个请求时，我们的服务器会为这个请求创建一个线程或进程来处理相关业务，而每个请求都需要独立的线程或进程来处理的处理方式，在多请求的并发请求情况下，肯定会导致资源紧张或者浪费，更何况我们知道创建线程或进程都是需要消耗资源的，且系统对线程或进程的调度的切换也是需要时间的，所以整体上执行的效率肯定是不高的。

而异步编程的模型就是为了解决这种过度的创建线程或进程等方式，也可以完成请求之间不再需要独立的线程或进程的的场景而出现的。

进而提出了新的概念，也就是所谓的“协程”，通过使用协程来处理并发请求。协程是一种轻量级的线程，它主要好处在于可以在代码中的某个点暂停执行，并在被唤醒时继续执行。使用协程可以避免线程切换的开销，并能够充分利用计算机的资源。它类似高速路上提供的服务区一样，提供了某种便利的方式，进行回调通知，解决高并发的问题。

而在FastAPI框架中，它也是基于Python的asyncio模块，使用异步协程来处理HTTP请求和响应。所以FastAPI中如果使用异步的特性来处理能够处理大量并发请求，也可以提供高性能的Web服务。谁说PY不可以写web api接口的呢？ 无非是选择题而已。 哈哈~

什么是协程

在讨论什么是协程之前，我们从前面必须认识到一个概念，也就是协程，它是一种轻量级的线程，它是程序员使用某种方式来定义的一种更加轻量的线程。再操作系统中，其实是没有所谓的‘协程’的概念的。

在前面我们提到“协程”主要好处在于可以在代码中的某个点暂停执行，并在被唤醒时继续执行。那这种可以在某点暂停，然后后续还可以再原地唤醒的机制。它提供了一种非阻塞的、高效的方式来处理异步任务。通过使用它，我们就可以在编写业务代码的的时候，可以同时进行多个任务的执行，提高代码的并发性和效率。协程也可以很好地处理事件驱动的编程模型，例如网络请求、IO操作等。

其实也是有相关的来源的。

在Python中，关于协程（coroutine）的定义：

  • 一种通过异步方式进行编程的函数。它可以被挂起和恢复执行，并且可以与其他协程之间进行协作。

它的表现形式是：

  • 协程使用async和await关键字来定义，被定义为协程的函数称为协程函数（coroutine function）。协程函数可以轻松地挂起和恢复执行，而不会阻塞其他任务。

协程挂起和唤醒形式：

  • 协程在执行时，可以通过调用await关键字来暂停执行，等待其他协程的完成或者异步操作的结果。而不是像普通函数那样一次性地执行完毕。

基于此，我们大概可以总结出来协程包括的一些特点包括：

  1. 可以被挂起和恢复执行，不会阻塞其他任务。
  2. 可以通过await关键字暂停执行，等待其他协程的完成或者异步操作的结果。
  3. 可以与其他协程之间进行协作，实现并发执行。
  4. 可以通过async关键字定义协程函数，通过await关键字挂起执行。
  5. 一个协程函数加上（）的时候，则变为一个协程对象

协程和生成器

在python中我们可以实现类似暂停后继续原地唤醒的机制，其实和我们之前了解到一个生成器的概念是类似的，生成器是一种对象，它可以迭代生成值，但是可以在生成值的过程中暂停执行，然后在需要时恢复执行。如下的代码示例：

首先生成器是一种可迭代对象，它可以通过函数的方式来定义并实现。在函数中通过yield语句来生成一个值，当返回后，则此时会暂停函数执行，直到下一次需要这个值的时候再继续执行。每一次迭代都会执行一次函数，并暂停在yield语句之后，直到重新进行迭代时，它会继续执行函数，并从上一次暂停的地方继续执行。

以下是一个生成器的示例：

def my_generator():
    yield 1
    yield 2
    yield 3

该生成器函数可以用在for循环中进行迭代，例如：

gen = my_generator()
for i in gen:
    print(i)

这将会输出以下内容：

1
2
3

上面的代码我们运行起来，会觉得无非是对一个可迭代对象的遍历循环，但是假设如果我们的换另一种遍历方式，如下示例代码所示：

def my_generator():
    yield 1
    yield 2
    yield 3

# 生成一个可迭代的对象
enn= my_generator()

print(next(enn))
print(next(enn))
print(next(enn))
# 已完成了，再继续获取下一个就报错
print(next(enn))

这将会输出以下内容：

Traceback (most recent call last):
  File "D:\code_loacl\fastapi_admin_api_pro\test22222.py", line 12, in <module>
    print(next(enn))
          ^^^^^^^^^
StopIteration
1
2
3

而改为我们前面使用for则不会抛出：StopIteration的错误，原因就是：因为我们的当前的迭代对象中已经没有值了，还继续迭代的时候，会引发StopIteration异常。但是使用可以使用for循环来遍历迭代器的时候，因为它已经自动处理StopIteration异常。所以没有抛出~

从上面的示例可以大概的看的出，我们的协程和生成器中的yield机制是类似的。本质上协程实际上是一种特殊的生成器。而协程也是一种可以在执行过程中暂停/恢复的函数，所以我们可以理解为协程是一种在生成器的基础上，通过使用 async 和 await 关键字来实现一种生成器。

协程和协程对象

在前面我们知道，协程它是一个特殊的函数，它可以暂停执行并在需要时恢复执行。当我们队协程函数加上（）进行处理的时候，此时得到的实例对象，就是一个协程对象则是表示协程的实例。

协程对象可以通过使用 async def 关键字定义，其中包含 await 关键字用于暂停协程的执行。定义协程对象后，可以通过调用它来创建一个协程，然后可使用 await 关键字来等待它的执行结果。

比如下实例代码：

import asyncio

async def countdowntimes():
    print("开始倒计时")
    for i in range(5, 0, -1):
        print("倒计时:",i)
        await asyncio.sleep(1)
    print("完成倒计时")

coro = countdowntimes()  # 创建一个协程对象
asyncio.run(coro)  # 执行协程对象

在上述例子中，countdown() 函数是一个协程对象，它被调用后返回一个协程对象，并被赋值给变量 coro。然后，通过 asyncio.run() 来执行这个协程对象，实现异步的倒计时输出。

asyncio.run()和事件循环

在前面的实例代码中，我们最终执行一个协程对象，是把它丢到asyncio.run(coro)，而我们知道asyncio它核心观念主要主要有：

  • 协程（coroutine）
  • 事件循环（event loop）

所以我们由此可以得出：asyncio.run()它的主要作用是执行给定的协程对象并返回最终结果。也就是说通常我们在使用 asyncio 进行异步编程时，通常需要包装整个程序的入口点，并通过调用 asyncio.run() 来启动事件循环并运行协程对象。

需要说明的是：这里的asyncio.run(coro) 函数做了其实帮我们自动处理了以下几件事情：

  1. 创建一个新的事件循环对象。
  2. 运行给定的协程对象（即 coro）。
  3. 等待协程对象完成并返回结果。
  4. 关闭事件循环并清理相关资源。

而这里我们asyncio.run(coro) 它就扮演着一个事件循环（Event Loop）的角色，这个事件循环（Event Loop）在整个的异步编程中的一个重要概念，它主用的作用是用于驱动和调度异步任务执行的机制。它是我们实现异步编程的核心机制，它提供了一种结构化的方式来处理异步任务，避免了回调地狱和线程管理的复杂性。通过事件循环，可以实现高效的并发执行和响应多个I/O操作，从而提高程序的性能和可伸缩性。

事件循环会时时的负责监听和处理事件，并根据事件的类型执行相应的任务。在异步编程中，事件通常包括网络请求完成、定时器到期、文件可读~可写等等。事件循环会不断地从事件队列中获取事件，并根据事件的类型确定执行相应的回调函数或协程。

在Python的asyncio模块中，事件循环通过执行协程对象来实现异步任务的调度和执行。它能够同时执行多个协程，并使用调度策略来分配系统资源和控制协程的执行顺序。

协程对象和协程任务（asyncio.create_task）

在异步编程中，我们所谈到的协程对象和协程任务是最长剑的概念，首先：

  • 协程对象在前面我们已经说明过，它是协程函数进行执行‘（）’得到的实例化的对象，此时我们可以对该对象进行使用 await 关键字来挂起（暂停）当前协程的执行。
  • 协程任务，则是对协程对象进行一系列分封装操作。进一步对协程对象的封装主要目的是：更利于循环事件对协程对象的调度。在python中，我们的协程任务可以使用 asyncio.create_task() 函数或 loop.create_task() 方法创建相关的任务。

所以我们需要明白一点，在asyncio中，协程对象是异步编程的基本单位，它是通过async/await语法来创建并执行（）得到实例对象。而任务（Task）对象则是在协程对象的基础上进行封装而来的，主要用于在事件循环中进行调度和管理。

我们可以理解为：任务对象是对协程对象的封装，我们可以通过创建任务对象来启动协程对象的执行。 我们对协程对象进行‘任务’化封装主要有下面一些好处：

  1. 异步任务的调度和管理：任务对象可以被事件循环统一地调度和管理。通过将协程对象封装为任务对象，可以将其加入到事件循环的任务队列中，然后我们的事件循环会自动进行任务调度执行和切换，而不是阻塞整个程序。

  2. 可以获取任务对象的状态和结果：任务对象提供了属性和方法来获取和操作任务对象的状态和结果。例如，可以通过task.done()方法来判断任务是否执行完毕，通过task.result()方法来获取协程对象的执行结果等。有这些方式，我们就可以可以方便地进行任务的状态监控、错误处理和结果处理。

  3. 支持任务的取消和超时：任务对象可以通过task.cancel()方法来取消任务的执行。这在某些场景下非常有用，例如当需要提前结束一个任务的执行时，可以使用取消操作。此外，通过设置任务对象的超时时间，可以在任务执行时间超过设定的时间后自动取消任务的执行，避免任务的长时间阻塞。

  4. 支持任务的回调函数：任务对象可以通过task.add_done_callback()方法来添加一个回调函数，在任务完成时被调用。这样可以在任务执行完毕后，对结果进行处理、进行下一步的操作或进行其他特定的处理。

asyncio.create_task 创建任务的应用

asyncio.create_task 函数是Python中用于创建任务的一个异步函数。在Python中，任务是异步编程中的基本组成部分之一，而 asyncio.create_task 函数则提供了一种简单轻量级的方法来创建任务。使用 asyncio.create_task 函数，我们可以简单的创建一个 coroutine 对象并将其封装在一个任务中，然后可以将这个任务添加到事件循环中，这样这个任务就可以异步的运行了。此外，asyncio.create_task 函数也会返回一个 asyncio.Task 对象，你可以使用它来管理这个任务。

当你要在异步程序中运行多个任务时，asyncio.create_task 函数就显得非常方便了。通过创建多个任务并将它们添加到事件循环，你就可以在不阻塞主线程的情况下并发的运行这些任务，从而提高异步程序的效率。

下面是一个演示 asyncio.create_task 函数的应用的例子：

import asyncio

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

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

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

asyncio.run(main())

这个例子中，我们定义了两个异步的 coroutine 对象 —— coroutine_1 和 coroutine_2，它们会分别输出一些内容并暂停一段时间。在 main 函数中，我们通过 asyncio.create_task 函数创建了两个异步任务，并把它们全都的添加到事件循环中，然后接着我们使用 asyncio.gather 函数，我们将这两个任务聚合到一起，并等待它们都完成后程序才会继续运行。在这个例子中，coroutine_1 和 coroutine_2 会同时运行，因为它们是异步的。而通过 asyncio.create_task 函数，我们可以更加方便的创建和管理这些任务。

需要注意：在 asyncio.create_task 的对象在没有await的情况下，代码会立即继续执行后续的代码，而不会等待task完成。但是，如果使用await task来等待task完成，代码会暂停执行并等待task的结果。一旦task完成，即my_task()协程执行完毕，代码会继续执行后续的代码。 下面是一个示例，演示了如何使用await task来等待任务的完成：

import asyncio

async def my_task():
    print("执行任务")
    await asyncio.sleep(1)
    return "任务结果"

async def main():
    task = asyncio.create_task(my_task())
    print("开始执行主任务")
    # result = await task
    print("主任务完成，任务结果为:", task)

asyncio.run(main())

我们通过注释  result = await task 来观察输出的结果： 没有加的情况输出如下：

开始执行主任务
主任务完成，任务结果为: <Task pending name='Task-2' coro=<my_task() running at D:\code_loacl\fastapi_admin_api_pro\test2222222.py:3>>
执行任务

加入加了的情况下：

开始执行主任务
执行任务
主任务完成，任务结果为: 任务结果

从上面可以看得出来，当解除注释并使用await等待task时，它会返回任务的结果。在上面的示例中，我们将任务的结果存储在result变量中，并在后续的代码中使用了该结果。

异步中任务并发执行（asyncio.gather和asyncio.wait）

对于任务的并发执行，我们通常我们同步的处理过程中，一般是通过多线程的方式的，而再异步编程过程中，对于批量任务的并发处理，我们也可以实现类似的多线程的并发，如前面的示例中我们的使用到了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.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的任务的对象，才可以放到await asyncio.wait中。上面的代码执行的结果会等待我们的所有的任务的执行完成后会再统一的进行输出。

超时示例： 

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.gather错误处理

asyncio.gather的错误异常处理和asyncio.wait稍微有些不同,首先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]

同步的阻塞函数转换为异步的非阻塞函数

在异步编程过程中，通常我们遇到一个需求，就是需要把现有一些同步阻塞的函数转换可以在异步协程函数中执行，转化后也不会遇到阻塞的问题，那么此时我们的 await asyncio.to_thread 的作用是将一个同步的阻塞函数转换为异步的非阻塞函数，以便在异步协程中调用。

在异步环境中，当需要调用一个同步的阻塞函数时，直接调用会阻塞当前的事件循环，导致整个程序的执行被阻塞。asyncio.to_thread 函数提供了一种在异步环境中调用同步函数的方式，它会将同步函数的调用封装在一个单独的线程中，并返回一个 Future 对象，从而使得该同步函数被异步化。

asyncio.to_thread 函数的语法如下：

await asyncio.to_thread(func, *args, **kwargs)

其中：

  • func 是要在一个单独的线程中执行的同步函数；
  • *args 和 **kwargs 是传递给 func 的参数。

使用 asyncio.to_thread 函数，可以将原本阻塞的同步函数转换为非阻塞的异步函数，从而在协程中并发地执行它，而不会阻塞整个事件循环。

以下是一个示例代码：

import asyncio
import time

def sync_func(arg1, arg2):
    # 阻塞的同步函数
    time.sleep(6)
    return f'ok：{arg1+arg2}'

async def async_func(arg1, arg2):
    # 异步函数
    result = await asyncio.to_thread(sync_func, arg1, arg2)
    print(result)
    
asyncio.run(async_func(1, 2))

在上面的例子中，sync_func 是一个阻塞的同步函数，里面模拟了同步阻塞休眠，并且最终会把传入的两个参数相加并返回结果。async_func 是一个异步函数，它使用 asyncio.to_thread 将 sync_func 转换为异步执行，并在协程中调用。asyncio.run 函数用于运行异步程序。

此转化和我们下面提到相关在FastAPI框架中处理异步转化相似。后续会提到。

FastAPI框架中的同步和异步

首先我们先看以下的示例代码： 以下是一个使用异步协程和同步方式的示例：

异步示例代码：

from typing import List
import asyncio
from fastapi import FastAPI

app = FastAPI()

async def async_sum(numbers: List[int]) -> int:
    # 定义async_sum协程函数
    total = 0
    for n in numbers:
        total += n
        await asyncio.sleep(1)  # 模拟阻塞任务
    return total

@app.get("/async")
async def get_async_sum(numbers: List[int]) -> int:
    result = await async_sum(numbers)
    return result

在上述示例中，async_sum函数使用异步方式，其中的await asyncio.sleep(1)语句用于模拟长时间运算或网络请求等待。get_async_sum函数使用async关键字定义为异步协程，并在内部通过await async_sum(numbers)调用异步计算函数。

同步方式：

from typing import List
from fastapi import FastAPI

app = FastAPI()

def sync_sum(numbers: List[int]) -> int:
    total = 0
    for n in numbers:
        total += n
    return total

@app.get("/sync")
def get_sync_sum(numbers: List[int]) -> int:
    result = sync_sum(numbers)
    return result

在上述示例中，sync_sum函数使用同步方式顺序计算输入数组中的元素和。get_sync_sum函数作为普通函数调用sync_sum函数，并返回结果。

如何看出上面请求的差异呐？接下来我们修改一下代码直接输出当前的线程的运行的ID：

from fastapi import FastAPI
import threading

app = FastAPI()


@app.get("/async")
async def get_async_sum():
    print('请求的为ID：',threading.current_thread().ident)
    return threading.current_thread().ident


@app.get("/sync")
def get_sync_sum():
    print('请求的为ID：',threading.current_thread().ident)
    return threading.current_thread().ident

if __name__ == "__main__":
    import uvicorn
    import os

    app_modeel_name = os.path.basename(__file__).replace(".py", "")
    uvicorn.run('test22222:app', host='127.0.0.1', port=21161)

然后我们使用一个而多线程并发访问测试验证输出： 多线程并发测试的代码如下：


import requests
import threading

def make_request(url):
    response = requests.get(url)
    print(f"Response: {response.text}")

url = "http://127.0.0.1:21161/sync"  # 替换为实际的接口地址

# 创建多个线程并发请求
num_threads = 10  # 指定线程数量
threads = []
for _ in range(num_threads):
    thread = threading.Thread(target=make_request, args=(url,))
    thread.start()
    threads.append(thread)

# 等待所有线程完成
for thread in threads:
    thread.join()

请求同步返回结果：

Response: 15436
Response: 1543
6Response: 5364
Response: 15436
Response: 2068
Response: 15436
Response: 2068
Response: 2068
Response: 1543
6Response: 2068

输出的是不同的ident。接下我们请求异步返回结果如下：

Response: 17904
Response: 1790
4Response: 17904
Response: 17904
Response: 17904
Response: 17904
Response: 17904
Response: 17904
Response: 17904
Response: 17904

由此我们的可以得出的结论就是异步执行的的时候是在单线程中处理，而我们的同步使用的是多线程方式进行并发任务处理的。

假设我们的在异步中使用同步的代码，如下示例代码所示： 
@app.get("/async")
async def get_async_sum():
    print('请求的为ID：',threading.current_thread().ident)
    time.sleep(5)
    return threading.current_thread().ident

上面的代码中我们在异步中使用了同步的阻塞代码，使用了time.sleep模拟业务耗时任务。然后再并发执行代码，观察输出结果。

然后再修改为：

@app.get("/async")
async def get_async_sum():
    print('请求的为ID：',threading.current_thread().ident)
    await asyncio.sleep(5)
    return threading.current_thread().ident

观察输出结果。对比输出会发现相关输出响应速度是不一样滴哦。 综上我们的可以知道，我们在异步里面调用同步代码的时候，是不可能的（应该是说不太现实）。非如此调用不可的情况，我们也可以经过相关转化处理，如下代码示例:

app = FastAPI()

import requests
import asyncio

@app.get("/async")
async def get_async_sum():
    print('请求的为ID：',threading.current_thread().ident)
    # requests.get('htttps://')
    response = await run_in_threadpool(requests.get, 'https://')
    print(response)
    return threading.current_thread().ident

在上面的示例中，我们引入了asyncio和run_in_threadpool函数。在get_async_sum异步函数中，我们使用了run_in_threadpool函数来将requests.get操作包装到了一个线程池中去运行处理。通过这种方式，我们的requests.get会在一个单独的线程中执行，而不会阻塞事件循环。等待请求完成后 再返回主线程中。

PS：这里的run_in_threadpool函数是由Starlette提供的，它提供了在异步函数中运行阻塞代码的功能。

在上面的代码中似乎我们， 使用run_in_threadpool处理转换也挺方便的，但是我们需要注意一点是，当我们调用的run_in_threadpool的时候，会每次都是创建一个新的线程池来处理任务，所以，这一点我们需要额外的注意。 深入到run_in_threadpool内部的源码，如下：


async def run_in_threadpool(
    func: typing.Callable[P, T], *args: P.args, **kwargs: P.kwargs
) -> T:
    if kwargs:  # pragma: no cover
        # run_sync doesn't accept 'kwargs', so bind them in here
        func = functools.partial(func, **kwargs)
    return await anyio.to_thread.run_sync(func, *args)

从代码中我们可以可以看到，内部的路基也是允许将一个同步的阻塞函数（func）异步地执行在一个线程池中。‘anyio.to_thread’ 和签名我们的也保持一致。只是更换了另一个异步的库。回到我们需要注意点上，在使用使用 run_in_threadpool() 执行异步的任务的时候，可能会导致性能问题，主要是因为它在每次调用时都会创建一个新的线程。这在高并发的情况下，可能会对系统的资源和性能产生一些负面影响。需要需要谨慎的使用。所以我们通常采取能替换就替换为异步IO的库，或者进行消息队列的异步化处理。

关于异步相关一些说明，本节先说到这。 文笔有限，如有笔误或错误！欢迎批评指正！感谢各位大佬！有什么问题也可以随时交流！

结尾

END

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

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

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

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

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


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

0
0
上一篇： 杂谈-FastAPI中的异步后台任务之Celery篇
下一篇： 基于FastAPI框架从零开始搭建权限管理系统--01前言篇
请登录后发表评论 登录
全部评论
程序员小钟同学

注册时间：2021-02-20

  • 博文量
    224
  • 访问量
    111678

最新文章

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

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

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

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