在python的整体生态中,虽然已经有很多库支持了异步调用,如可以使用httpx或者aiohttp代替requests库发起http请求,使用asyncio.sleep 代替time.sleep, 但是依然还有很多优秀的第三方库是不支持异步调用也没有可代替的库,那么如何在FastAPI中调用这种没有实现异步的库但是又不阻塞整个系统呢?
我们来详细看一下各种情况。本文将使用await asyncio.sleep(5) 来模拟实现了异步操作的库。time.sleep(5) 来模拟只有同步操作的库。
import time from fastapi import FastAPI, Request import asyncio import threading import uvicorn app = FastAPI() @app.middleware("http") async def cal_time(req: Request, call_next): start = time.time() response = await call_next(req) process_time = time.time() - start print(f"接口{req.url.path} use {process_time} sec.") return response @app.get("/test1") def test1(): print(threading.current_thread(), "test1") return "test1" @app.get("/test2") def test2(): print(threading.current_thread(), "test2") time.sleep(5) return "test2" if __name__ == '__main__': uvicorn.run(app, host="127.0.0.1", port=8000)上面有的fastapi 定义了两个接口,test1接口未执行任何IO操作,只返回一个字符串,用来验证其它接口是否会阻塞test1接口,test2 接口,这里使用time.sleep(5) 来模拟一个阻塞操作。并且同时打印了test1和test2中的线程信息。并且添加了一个中间件cal_time,打印每个请求的耗时。
<WorkerThread(AnyIO worker thread, started 11720)> test2 <WorkerThread(AnyIO worker thread, started 22508)> test1 接口/test1 use 0.0019936561584472656 sec. INFO: 127.0.0.1:56758 - "GET /test1 HTTP/1.1" 200 OK 接口/test2 use 5.0042359828948975 sec. INFO: 127.0.0.1:56757 - "GET /test2 HTTP/1.1" 200 OK由于先请求的test2接口,所以先打印出了test2接口的线程信息,在11720 线程上执行。
import time from fastapi import FastAPI, Request import asyncio import threading import uvicorn app = FastAPI() @app.middleware("http") async def cal_time(req: Request, call_next): start = time.time() response = await call_next(req) process_time = time.time() - start print(f"接口{req.url.path} use {process_time} sec.") return response @app.get("/test1") def test1(): print(threading.current_thread(), "test1") return "test1" @app.get("/test2") async def test2(): print(threading.current_thread(), "test2") await asyncio.sleep(5) return "test2" if __name__ == '__main__': uvicorn.run(app, host="127.0.0.1", port=8000)这里修改test2方法,使用 await asyncio.sleep(5) 模拟异步阻塞操作,由于这里使用了await, 所以需要将test2改为async def,这时还是先访问test2接口,再访问test1接口,此时test2中的异步IO操作并依然没有阻塞test1接口,我们再来看一下请求的打印输出。
<_MainThread(MainThread, started 4501536256)> test2 <WorkerThread(AnyIO worker thread, started 123145549803520)> test1 接口/test1 use 0.003657102584838867 sec. INFO: 127.0.0.1:56481 - "GET /test1 HTTP/1.1" 200 OK 接口/test2 use 5.0049920082092285 sec. INFO: 127.0.0.1:56476 - "GET /test2 HTTP/1.1" 200 OK1.首先访问test2接口,打印了test2接口的线程信息。
<WorkerThread(AnyIO worker thread, started 11720)> test2 <WorkerThread(AnyIO worker thread, started 22508)> test1我们看到两次请求是在不同的线程中运行的,所以即使某个接口请求中存在阻塞操作,也不会影响到其它的线程。
<_MainThread(MainThread, started 4501536256)> test2 <WorkerThread(AnyIO worker thread, started 123145549803520)> test1依然是在两个不同的线程中运行的,所以也不会相互阻塞。
<WorkerThread(AnyIO worker thread, started 11720)> test2 <_MainThread(MainThread, started 4501536256)> test2
第一次在使用def 定义函数时,是在WorkerThread中运行的,第二次使用async def 显示是在MainThread 中运行的,这也就说明了一个问题,在python 中,使用async def 定义的函数是运行在协程中,而多个协程是在一个主线程中的。
import time from fastapi import FastAPI, Request import asyncio import threading import uvicorn from uvicorn.config import LOGGING_CONFIG app = FastAPI() @app.middleware("http") async def cal_time(req: Request, call_next): start = time.time() response = await call_next(req) process_time = time.time() - start print(f"接口{req.url.path} use {process_time} sec.") return response @app.get("/test1") def test1(): print(threading.current_thread(), "test1") return "test1" @app.get("/test2") async def test2(): print(threading.current_thread(), "test2") time.sleep(5) return "test2" if __name__ == '__main__': LOGGING_CONFIG["formatters"]["access"][ "fmt"] = '%(asctime)s %(levelprefix)s %(client_addr)s - "%(request_line)s" %(status_code)s' uvicorn.run(app, host="127.0.0.1", port=8000)
这次我们打印一下请求的具体时间,由于默认的uvicorn 日志没有打印具体的时间,所以这里通过LOGGING_CONFIG重新定义一下日志的格式。在test2函数这时我们使用time.sleep(5) 替换掉await asyncio.sleep(5), 这时依然使用async def 定义test2函数,再访问一下test2和test1接口。
<_MainThread(MainThread, started 3768)> test2 接口/test2 use 5.008129358291626 sec. 2023-11-29 11:33:12,724 INFO: 127.0.0.1:64011 - "GET /test2 HTTP/1.1" 200 OK <WorkerThread(AnyIO worker thread, started 20056)> test1 接口/test1 use 0.006971836090087891 sec. 2023-11-29 11:33:12,730 INFO: 127.0.0.1:64012 - "GET /test1 HTTP/1.1" 200 OK输出分为两部分,一个是访问test2的输出,一个是访问test1的输出。
在FastAPI中,如果使用async def 定义的函数,里面的IO操作均要实现异步操作await,如果要使用同步的IO操作,需要使用def 定义函数。简单来讲用下图表示