1 前言

在上一篇有有提到关于异步到同步的转换。

fastapi框架系列-异步函数转换普通函数

那既然可以相互转换的，那如何的把同步的代码也转为为协程的呢？

在这里我们需要说一下，我们为什么需要把同步转为异步呢？因为目前大部分的库有可能还没有对应的aio的实现，假设比如我们的场景的http请求库（目前异步的已经有了，如aiohttp,httpx等，这里仅仅是假设）requests，如果我们的项目整个过程都是使用异步的方式来进行处理，那么我们如何在异步的逻辑中使用同步的，而又不因为产生阻塞的呢？毕竟我们的知道协程是单进程单线程的，如果调用同步的话，不用怀疑，肯定是会阻塞的，所以我们的就需要做转换了。

各种库示例演示

BackgroundTasks案例代码演示：

其实关于异步到同步的转换在FastApi框架比较明显的地方就是后台任务，通过后台任务的源码我们可以分析到，它对对我们的同步的函数如何转换已经有明显的示例，如BackgroundTasks源码：

import asyncio
import typing

from starlette.concurrency import run_in_threadpool


class BackgroundTask:
    def __init__(
        self, func: typing.Callable, *args: typing.Any, **kwargs: typing.Any
    ) -> None:
        self.func = func
        self.args = args
        self.kwargs = kwargs
        self.is_async = asyncio.iscoroutinefunction(func)

    async def __call__(self) -> None:
        if self.is_async:
            await self.func(*self.args, **self.kwargs)
        else:
            await run_in_threadpool(self.func, *self.args, **self.kwargs)

在上述的代码中，我们可以看到他们是通过对函数判断是否是is_async来处理，如果是协程函数则直接的

 await self.func(*self.args, **self.kwargs)

如果是普通函数那么他就进行使用

await run_in_threadpool(self.func, *self.args, **self.kwargs)

进行包裹之后返回具体一个协程对象。

然后继续顺着run_in_threadpool的源码查看，

async def run_in_threadpool(
    func: typing.Callable[..., T], *args: typing.Any, **kwargs: typing.Any
) -> T:
    if contextvars is not None:  # pragma: no cover
        # Ensure we run in the same context
        child = functools.partial(func, *args, **kwargs)
        context = contextvars.copy_context()
        func = context.run
        args = (child,)
    elif 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)

我们可以看到他是通过

await anyio.to_thread.run_sync(func, *args)

来完成最终的转换。如果有需要，我们可以参考这个来做自己的转换库。

还有一个比较有点意思的库awaits，通过装饰器转换

asgiref 同步到异步的转换

我们上面提到asgiref的转换是相互，同样的asgiref也提供了对应的转换。

asgiref 官网文档地址为：

https://asgi.readthedocs.io/en/latest/ 具体案例：

from fastapi import FastAPI,Request
from fastapi.responses import PlainTextResponse,HTMLResponse
from asgiref.sync import sync_to_async
import requests
# 定义我们的APP服务对象
app = FastAPI()

def getdata():
    return requests.get('http://www.baidu.com').text

@app.get("/get/access_token")
async def access_token(reques:Request):
    asds= await sync_to_async(func=getdata)()
    return HTMLResponse(asds)


if __name__ == '__main__':
    import uvicorn
    uvicorn.run('kkk:app', host="127.0.0.1", port=8100, debug=True, reload=True)

需要注意点是sync_to_async返回的是携程的函数，所以后面需要转换为携程对象，需要加（）

await说明科普：

• await后面跟的是一个协程对象或Awaitable，它主要的目的声明程序在此处需要挂起且等待协程控制流的返回。

asyncer Fastapi框架大佬自己写的库

asyncer和asgiref一样，它不仅可以把同步代码异步化，还可以把异步代码同步化。我们上面提到asgiref的转换是相互，同样的asgiref也提供了对应的转换。

asyncer 官网文档地址为：

https://asyncer.tiangolo.com/

库的安装：

pip isntall asyncer

通过官网示例，我们使用起来也很方便，这里我列举官方的示例，如下示例是同步转换异步的示例：

from fastapi import FastAPI,Request
from fastapi.responses import PlainTextResponse,HTMLResponse
from asyncer import asyncify
import requests
# 定义我们的APP服务对象
app = FastAPI()

def do_sync_work(name):
    return requests.get(f'http://www.baidu.com?name={name}').text

@app.get("/get/access_token")
async def access_token(reques:Request):
    # asds= await sync_to_async(func=getdata)()
    message = await asyncify(do_sync_work)(name="World")
    return HTMLResponse(message)


if __name__ == '__main__':
    import uvicorn
    uvicorn.run('kkk:app', host="127.0.0.1", port=8100, debug=True, reload=True)

需要注意点是asyncify返回的是携程的函数，所以后面需要转换为携程对象，需要加（）

await说明科普：

• await后面跟的是一个协程对象或Awaitable，它主要的目的声明程序在此处需要挂起且等待协程控制流的返回。

有点意思的库awaits，通过装饰器形式

还有一个比较有点意思的库awaits，通过装饰器转换

awaits 官网文档地址为：

https://github.com/pomponchik/awaits

感兴趣的可以直接的查阅官网地址进行使用，这个是库感觉是值得研究一下的。

aioify 让每个函数都可以异步且等待的库

aioify库按他自己说法是：让每个函数都可以异步且等待。

库的安装：

pip isntall aioify

具体示例：

from fastapi import FastAPI,Request
from fastapi.responses import PlainTextResponse,HTMLResponse
from aioify import aioify
import requests
# 定义我们的APP服务对象
app = FastAPI()

def do_sync_work(name):
    return requests.get(f'http://www.baidu.com?name={name}').text

@app.get("/get/access_token")
async def access_token(reques:Request):
    # asds= await sync_to_async(func=getdata)()
    message = await aioify(do_sync_work)(name="World")
    print(message)
    return HTMLResponse(message)


if __name__ == '__main__':
    import uvicorn
    uvicorn.run('kkk:app', host="127.0.0.1", port=8100, debug=True, reload=True)

其实多数的库是一样的。直接进行转换即可使用。它还可以当做装饰器直接的同步函数装饰之后，进行使用，如下示例代码：

from fastapi import FastAPI,Request
from fastapi.responses import PlainTextResponse,HTMLResponse
from aioify import aioify
import requests
# 定义我们的APP服务对象
app = FastAPI()

@aioify
def do_sync_work(name):
    return requests.get(f'http://www.baidu.com?name={name}').text

@app.get("/get/access_token")
async def access_token(reques:Request):
    # asds= await sync_to_async(func=getdata)()
    message = await do_sync_work(name="World")
    print(message)
    return HTMLResponse(message)


if __name__ == '__main__':
    import uvicorn
    uvicorn.run('kkk:app', host="127.0.0.1", port=8100, debug=True, reload=True)

总结-同步到异步转换的本质

各库的实现各有千秋，但本质内部源码殊途同归！同步到异步转换的本质其实就是把同步的函数通过线程的方式进行调用，如下的代码示例：

 def asyncify(
    function: Callable[T_ParamSpec, T_Retval],
    *,
    cancellable: bool = False,
    limiter: Optional[anyio.CapacityLimiter] = None
) -> Callable[T_ParamSpec, Awaitable[T_Retval]]:

    async def wrapper(
        *args: T_ParamSpec.args, **kwargs: T_ParamSpec.kwargs
    ) -> T_Retval:
        partial_f = functools.partial(function, *args, **kwargs)
        return await anyio.to_thread.run_sync(
            partial_f, cancellable=cancellable, limiter=limiter
        )

    return wrapper

其实都是通过await anyio.to_thread.run_sync来进行处理。

同步和异步转换时候关键概念：

以上仅仅是个人结合自己的实际需求，做学习的实践笔记！如有笔误！欢迎批评指正！感谢各位大佬！

结尾

END

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

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

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

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