app.py文件--------创建项目、设置路由、启动项目
from flask import Flask
from flask_restful import Api
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor(max_workers=10)
app = Flask(__name__)
app.executor = executor
api = Api(app)
from app.num_check import num_check
#添加路由接口
api.add_resource(num_check, "/num_check/LV/check_test/")
if __name__ == '__main__':
app.run()
num_check.py------利用Resource模板设置get请求的响应信息
from flask_restful import Resource
from flask import request
from flask import current_app
import time
class num_check(Resource):
def __init__(self):
self.GET = request.args.to_dict()
def get(self):
num = request.args.to_dict().get("num")
num = int(num)
time.sleep(3)
t = current_app.executor.submit(num_check_load, num)
check_info = t.result()
if check_info[0]:
return {"code": 0, "info": "", "data": check_info[1]}
else:
return {"code": 1, "info": check_info[1], "data": ""}
def num_check_load(num):
if num > 10:
return [False, "%s 不能大于10" % num]
check_info = "%s 通过" % num
return [True, check_info]
check.py-------测试文件
import requests
from threading import Thread
def requests_result(URL):
ret = requests.get(URL)
if ret.status_code != 200:
print("请求失败")
ret_dect = ret.json()
if ret_dect["code"] == 1:
error_info = ret_dect["info"]
print(error_info)
else:
print(ret_dect["data"])
return ret
threading_fool = []
for i in range(100):
URL = "http://127.0.0.1:5000//num_check/LV/check_test/?num=%s" % i
t = Thread(target=requests_result, args=(URL,))
threading_fool.append(t)
for i in threading_fool:
i.start()
for i in threading_fool:
i.join()
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 [email protected] 举报，一经查实，本站将立刻删除。