做网站维护要多少钱一年,企业网站建设开始起步文章,wordpress 产品购买,wordpress的搜索结果Python WEB框架FastAPI #xff08;二#xff09; 最近一直在使用fastapi#xff0c;随着使用的深入发现我对于它的了解还是太少了#xff0c;以至于踩了一些坑。所以在这里记录一下#xff0c;愿看到的小伙伴不迷路。
路径传参并发问题
一、路径传参
这是对上一个传参…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_workers2)def long_running_task(id):print(f{id} 开始执行)time.sleep(10)return OKapp.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 datetimeimport uvicorn
from fastapi import FastAPI, Requestapp FastAPI()# 创建全局线程池
thread_pool ThreadPoolExecutor(max_workers2)def long_running_task(id):print(f{id} 开始执行)time.sleep(10)return OKapp.get(/test/{id})
def test(request: Request, id):print(f{id} 请求进入)future thread_pool.submit(long_running_task, id)# 等待任务结束res future.result()return resapp.get(/test3)
async def test():now_time datetime.now().strftime(%Y-%m-%d %H:%M:%S)print(now_time, /test3)time.sleep(10)return OKapp.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 OKif __name__ __main__:# 创建解析器parser ArgumentParser()# 添加命令行参数parser.add_argument(--host, default0.0.0.0, typestr, helpServer bound address)parser.add_argument(--port, default8000, typeint, helpPort number)# 解析命令行参数args parser.parse_args()# 启动服务器uvicorn.run(appapp, hostargs.host, portargs.port)
