python pymysql数据库连接池范例from flask import Flask, jsonify, request import json import pymysql from dbutils.pooled_db import PooledDB app Flask(__name__) POOL PooledDB(creatorpymysql, maxconnections10, mincached 3, maxcached 5, blockingFalse, #是否阻塞 host127.0.0.1, userroot, passwordxxxxx, databasexxxxx) def getUser(type : int, token : str) - dict: user_info_list {} if 1 type : with open(user_info.txt, moder, encodingutf-8) as f: for x in f: x x.strip() #去空格 valid_token, valid_name x.split(,) #拆包 user_info_list[valid_token] valid_name return user_info_list else: db POOL.connection() cursor db.cursor() cursor.execute(fselect token,name from users) result cursor.fetchall() #二维元组直接转字典 return dict(result) # ((f29d7c0e-8a51-437b-9106-27e489bc01a5, 张三), (18ce9072-d65f-4120-b834-72fa61de0cf3, 李四), (75ac4e10-1297-492c-b25d-30e76f195841, 王五)) app.route(/index, methods[POST]) def index(): token request.json.get(token) if not token: dic {success: False, data: token失效} return jsonify(dic) #返回json并设置json响应头 user_info_list getUser(2, token) if token not in user_info_list: dic {success: False, data: token失效} return jsonify(dic) name request.json.get(name) print(token, name) dic {success: True, data: 成功} return jsonify(dic) if __name__ __main__: # 本地调试 app.run(debugTrue, host0.0.0.0, port5001) # 生产debugFALSEpython redis连接池范例import time from flask import Flask, jsonify, request import json import pymysql from dbutils.pooled_db import PooledDB import uuid import redis from redis.connection import ConnectionPool import threading import pickle #字节互转字符串 app Flask(__name__) REDIS_CONFIG { max_connections: 20, # 连接池最大连接数 host: 127.0.0.1, port: 6379, db: 1, password: xxxx, # 有密码填字符串无则None decode_responses: False, # 自动bytes转str不用手动decode socket_timeout: None, # 没有连接超时 retry_on_timeout: True, socket_keepalive: True } # 初始化全局连接池 POOL ConnectionPool(**REDIS_CONFIG) TASK_QUEUE task_queue RESULT_QUEUE result_queue # 获取redis客户端从池里取连接 def get_redis_client() - redis.Redis: return redis.Redis(connection_poolPOOL) # 全局单例client redisClient get_redis_client() def getUser() - dict: user_info_list {} with open(user_info.txt, moder, encodingutf-8) as f: for x in f: x x.strip() valid_token, valid_name x.split(,) user_info_list[valid_token] valid_name return user_info_list app.route(/task, methods[POST]) def task(): token request.json.get(token) if not token: dic {success: False, data: token失效} return jsonify(dic) user_info_list getUser() if token not in user_info_list: dic {success: False, data: token失效} return jsonify(dic) # 加入队列并返回tid tid str(uuid.uuid4()) # 任务入参 {tid - token} task_data {tid:tid,token:token} redisClient.lpush(TASK_QUEUE, pickle.dumps(task_data)) dic {success: True, tid:tid,message: 成功} return jsonify(dic) def work(): while True: queue_name, task_str redisClient.brpop([TASK_QUEUE]) if not task_str: time.sleep(2) continue # task_data json.loads(task_str) #支持加载字节、字符串 task_data pickle.loads(task_str) #加载字节 user_info_list getUser() tid task_data.get(tid) token task_data.get(token) name user_info_list.get(token) redisClient.hset(RESULT_QUEUE,tid,pickle.dumps(name)) app.route(/query, methods[POST]) def query(): tid request.json.get(tid) value redisClient.hget(RESULT_QUEUE,tid) if not value: dic {success: False, data: 数据还在跑} return jsonify(dic) dic {success: True, data: pickle.loads(value)} return jsonify(dic) if __name__ __main__: # 本地调试 redisClient.delete(TASK_QUEUE) redisClient.delete(RESULT_QUEUE) threading.Thread(targetwork,args()).start() #启动任务执行线程 app.run(debugTrue, host0.0.0.0, port5001) # 生产debugFALSE