一.介绍Flask 是一款基于 Python 开发的轻量级 Web 应用框架核心依赖 WerkzeugWSGI 工具集负责路由、请求响应处理和 Jinja2模板引擎负责页面渲染属于典型的「微框架Micro-Framework」。它的设计理念是「核心精简、按需扩展」框架本身只提供 Web 开发的基础能力复杂功能通过第三方扩展实现开发者可以自由选择技术栈灵活度极高。安装pip install flask简单示例from flask import Flask app Flask(__name__) app.route(/demo) def hello(): return Hello, Flask! if __name__ __main__: app.run(debugTrue)程序运行后得到网址。访问网址时这是因为仅仅访问了网址若要显示具体内容则需要在网址上加/demo本篇将从无耗时版耗时版来解析。二.无耗时版2.1获取请求数据2.1.1获取 URL 查询参数Query Stringage request.args.get(age) pwd request.args.get(pwd) print(age, pwd)2.1.2获取表单数据Form Dataxx request.form.get(xx) yy request.form.get(yy) print(xx,yy)2.1.3JSON 格式2.1.3.1获取 JSON 格式的请求数据print(request.json)request.json是一个属性它会自动解析请求体中的 JSON 数据并返回 Python 字典或列表等。要求请求的Content-Type必须是application/json否则会返回None。2.1.3.2 返回 JSON 响应return jsonify({status:Ture,data:sucessful})jsonify() 是 Flask 提供的一个函数它会把传入的字典转换成 JSON 字符串并自动设置响应头的 Content-Type 为 application/json。浏览器浏览器里只能直接发起 GET 请求默认是模拟不出来post请求所以需要用到postmanpostman下载官网https://www.postman.com/downloads/根据需求自己进行测试 URL 参数测试表单提交测试 JSON 数据等一系列操作。2.2直接访问api并返回数据接收一个字符串 拼接密钥 MD5 签名并立即返回签名的完整接口逻辑。import hashlib from flask import Flask, request, jsonify app Flask(__name__) app.route(/bili, methods[POST]) def bili(): ordered_string request.json.get(ordered_string) if not ordered_string: return jsonify({status: False, error: 参数错误}) # 拼接密钥生成MD5签名 encrypt_string ordered_string 560c52ccd288fed045859ed18bffd973 obj hashlib.md5(encrypt_string.encode(utf-8)) sign obj.hexdigest() return jsonify({status: True, data: sign}) if __name__ __main__: app.run(host127.0.0.1, port5000)对于该代码来说postman测试时请求方法POSTURLhttp://127.0.0.1:5000/biliBody 选择raw→JSON发送可得到回复2.3 文件版本授权2.3.1 读取白名单DB_FILE db.txt def get_user_dict(): info_dict {} try: with open(DB_FILE, r, encodingutf-8) as f: for line in f: line line.strip() if not line: continue parts line.split(\t) if len(parts) 2: token parts[0] username parts[1] info_dict[token] username except FileNotFoundError: pass return info_dict2.3.2 在接口中校验 tokenapp.route(/bo, methods[POST]) def bo(): token request.args.get(token) if not token: return jsonify({status: False, error: 认证失败缺少token参数}), 401 # 加载白名单 user_dict get_user_dict() if token not in user_dict: return jsonify({status: False, error: 认证失败token无效}), 401 # ... 后续业务逻辑如签名计算文件授权db.txt的作用详解 核心用途接口身份鉴权登录校验 这个 db.txt 是白名单凭证库用来控制谁能调用你的 MD5 加密接口 只有文件里登记过的 token才能成功请求接口 没登记的 token、乱填的 token直接返回 401 认证失败拒绝访问 每个 token 绑定一个用户名方便区分是谁在调用接口。2.4 数据库版本2.4.1 跨域from flask_cors import CORS CORS(app)前端页面如 http://localhost:8080调用后端 http://127.0.0.1:5000浏览器同源策略拦截请求报跨域错误。 flask-cors 专门解决该问题。2.4.2 数据库配置DB_CONFIG { host: 127.0.0.1, port: 3306, user: root, password: 密码, database: 数据库名, charset: utf8mb4 }2.4.3 数据库查询def fetch_one(sql, paramsNone): conn None cursor None try: conn pymysql.connect(**DB_CONFIG) cursor conn.cursor() cursor.execute(sql, params or ()) return cursor.fetchone() except MySQLError as e: app.logger.error(f数据库异常: {str(e)}) return None finally: if cursor: cursor.close() if conn: conn.close()每次调用都创建新连接执行完立即关闭。2.4.4 接口核心逻辑app.route(/boo, methods[POST]) def boo(): try: token request.args.get(token) if not token: return jsonify({status: False, error: 认证失败缺少token参数}), 401 # 数据库校验 token同时取出用户名 user_row fetch_one(SELECT token, name FROM flask WHERE token%s, (token,)) if not user_row: return jsonify({status: False, error: 认证失败token无效}), 401 username user_row[1] # JSON 格式校验、参数校验 ... # MD5 签名计算 ... # 返回结果 except Exception as e: return jsonify({status: False, error: 服务器内部错误}), 500从 URL 参数获取 token数据库白名单验证SELECT token, name FROM flask WHERE token%s获取用户名后附在响应中方便识别调用者2.5 数据库连接池2.5.1 创建全局连接池from dbutils.pooled_db import PooledDB import pymysql POOL PooledDB( creatorpymysql, maxconnections10, # 连接池最大连接数 mincached2, # 启动时预建的最小空闲连接 maxcached6, # 最大空闲连接数0无限制 blockingTrue, # 连接数满时是否阻塞等待 maxusage1000, # 单个连接复用次数上限 ping1, # 取出连接时先 ping 保活 host127.0.0.1, port3306, userroot, password密码, database数据库名, charsetutf8mb4 )dbutils提供数据库连接池解决原生驱动每次请求新建 / 销毁连接的性能缺陷。2.5.2 使用连接import pymysql.cursors def fetch_one(sql, paramsNone): conn None cursor None try: conn POOL.connection() # 从池里取连接 cursor conn.cursor(pymysql.cursors.DictCursor) # 字典游标 cursor.execute(sql, params or ()) return cursor.fetchone() except Exception as e: app.logger.error(fDB error: {e}) return None finally: if cursor: cursor.close() if conn: conn.close() # 不是真关闭是把连接归还给池2.5.3 对比原生 pymysqlDBUtils 连接池每次请求新建连接频繁 TCP 握手并发卡顿连接复用减少开销高并发稳定异常易出现连接泄漏数据库连接数爆满自动归还、自动检测断线重连无连接上限控制流量大时数据库崩溃限制最大连接数保护 MySQL三.耗时版3.1 接受任务并加入队列app.route(/boo, methods[POST]) def boo(): # 1. 解析请求体为 JSON req_data request.get_json() # 2. 获取 ordered_string 参数默认空字符串 ordered_string req_data.get(ordered_string, ) # 3. 参数校验不能为空 if not ordered_string: return jsonify({status: False, error: 参数错误}), 400 # 4. 生成全局唯一任务 ID tid str(uuid.uuid4()) # 5. 创建 Redis 连接每次请求新建生产环境建议用连接池 redis_conn redis.Redis(**REDIS_CONFIG) # 6. 构造任务字典并序列化为 JSON 字符串 task_info json.dumps({ tid: tid, ordered_string: ordered_string }) # 7. 将任务推入 Redis 列表的左侧队头 redis_conn.lpush(task_queue, task_info) # 8. 立即返回任务 ID不等待实际处理 return jsonify({ status: True, data: { tid: tid, message: 任务已提交等待处理 } })UUID用于生成全局唯一标识符为什么是 lpush将新任务推入 List 的左侧头部而后台 Worker 使用 brpop 从右侧尾部弹出。效果先入队的任务在 List 更靠右的位置会被优先取出从而保证 FIFO先进先出 顺序。3.2 worker处理后台 Worker一个独立的守护线程启动时连接 Redis无限循环地从 task_queue 阻塞式弹出任务brpop执行模拟的耗时操作time.sleep(2)然后将结果推入 result_queue。初始化 Redis 连接并保持长连接这条连接在整个 Worker 生命周期内只创建一次并反复使用。Redis作用一个基于内存的“数据结构服务器”主要用来解决高并发下关系型数据库如 MySQL的读写瓶颈提升系统响应速度。def worker(): 后台工作线程从任务队列取任务处理后放入结果队列 try: redis_conn redis.Redis(**REDIS_CONFIG) logging.info(Worker 启动成功等待任务...) while True: # 1. 阻塞式从任务队列右侧弹出任务 task_data redis_conn.brpop(task_queue, timeout1) if task_data: queue_name, task_json task_data task json.loads(task_json) tid task.get(tid) ordered_string task.get(ordered_string) logging.info(fWorker 开始处理任务: {tid}) # 2. 模拟耗时处理 time.sleep(2) result f处理完成: {ordered_string} # 3. 将结果放入结果队列 result_data json.dumps({ tid: tid, result: result, status: completed }) redis_conn.lpush(result_queue, result_data) logging.info(fWorker 完成任务: {tid}) except Exception as e: logging.error(fWorker 异常: {e})为什么用 brpop 而不是 rpoprpop 是无阻塞的如果队列为空立即返回 None这样 Worker 就会在 while True 中高速空转白白消耗 CPU。brpop 在无任务时让线程进入休眠不占用 CPU有新任务时立即唤醒既高效又节省资源。Worker 与 Web 接口的协作全景客户端 → POST /boo带 ordered_stringWeb 线程 → 生成 tidlpush 到 task_queue立即返回 tidWorker 线程 → brpop 拿到任务执行耗时操作lpush 结果到 result_queue客户端 → 轮询 GET /result/tid遍历 result_queue 查找匹配的 tid获取结果四.附1.Redis下载Redis下载https://github.com/tporadowski/redis/releases2.状态码常见状态码状态码含义使用场景200OK 请求成功正常返回业务数据400Bad Request 参数非法缺少参数、参数格式错误401Unauthorized 未认证token 不存在、token 失效403Forbidden 权限不足登录成功但无操作权限404Not Found 接口不存在访问不存在的路由500Internal Server Error代码报错、数据库异常