Python WEB框架FastAPI （二）

---

最近一直在使用fastapi，随着使用的深入发现我对于它的了解还是太少了，以至于踩了一些坑。所以在这里记录一下，愿看到的小伙伴不迷路。

• 路径传参
• 并发问题

一、路径传参

这是对上一个传参知识的补充，除了通过request对象传参以及参数名传参，还可以通过请求路径传参。这也是开发中常用的传参方式，请看以下代码：

@app.get("/test/{id}")
def test(id):
    print(f"收到请求！{id}")
    return id

代码就不用解释了吧。相信各位都能看懂。

二、并发问题

这个部分是我今天要聊的一个重点，确实花了一些时间。

1、async关键字会导致请求阻塞

当方法添加了async关键字时，请求将被串行，后进的请求会等待前一项请求结束才能够进方法。

@app.get("/test3")
async def test():
    now_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    print(now_time, "/test3")
    time.sleep(10)
    return "OK"

浏览器开启两个tag访问/test3，后台只会打印一个/test3，另一个10s中之后才会被打印出来。

2、不使用async，相同的请求会被阻塞

如果两个请求完全相同，则会阻塞等待前一个请求结束。

@app.get("/test4")
def test(request: Request):
    now_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    print(now_time, request.url)
    time.sleep(10)
    return "OK"

浏览器开两个tag，访问 /test4，第二个请求会在 10s后进入，跟async使用时效果相同。

分别用edge和chrome访问 /test，两个请求则会同时进入。

另外如果是同一浏览器请求，但是携带的参数不同，结果也会同时进入：

综上，如果要实现并发，去掉async关键字即可，完全一样的请求会被阻塞，个人认为也是正常的，也能够防止恶意攻击。

3、使用线程池控制并发量

创建全局线程池，并发量为2

# 创建全局线程池
thread_pool = ThreadPoolExecutor(max_workers=2)


def long_running_task(id):
    print(f"{id} 开始执行！")
    time.sleep(10)
    return "OK"

@app.get("/test/{id}")
def test(request: Request, id):
    print(f"{id} 请求进入！")
    future = thread_pool.submit(long_running_task, id)
    # 等待任务结束
    res = future.result()
    return res

当四个请求同时访问，请求会同时进入，但是只会同时处理两个请求：

以上便是本次的一个学习笔记，欢迎大家留言探讨！
最后奉上完整的测试源码：

import time
from argparse import ArgumentParser
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime

import uvicorn
from fastapi import FastAPI, Request

app = FastAPI()

# 创建全局线程池
thread_pool = ThreadPoolExecutor(max_workers=2)


def long_running_task(id):
    print(f"{id} 开始执行！")
    time.sleep(10)
    return "OK"

@app.get("/test/{id}")
def test(request: Request, id):
    print(f"{id} 请求进入！")
    future = thread_pool.submit(long_running_task, id)
    # 等待任务结束
    res = future.result()
    return res



@app.get("/test3")
async def test():
    now_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    print(now_time, "/test3")
    time.sleep(10)
    return "OK"


@app.get("/test4")
def test(request: Request):
    now_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    print(now_time, request.url)
    time.sleep(10)
    return "OK"


if __name__ == "__main__":
    # 创建解析器
    parser = ArgumentParser()
    # 添加命令行参数
    parser.add_argument('--host', default="0.0.0.0", type=str, help='Server bound address')
    parser.add_argument('--port', default=8000, type=int, help='Port number')
    # 解析命令行参数
    args = parser.parse_args()
    # 启动服务器
    uvicorn.run(app=app, host=args.host, port=args.port)