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()