Flask API调用queue.get()无法获取队列值的最优实现方案咨询

• 你当前用的是Python进程内的内存队列queue.Queue，如果Flask以多worker模式部署（比如用Gunicorn、uWSGI开多个进程，或者开启debug模式的热重载功能），每个进程会单独初始化一个q_sample实例，队列数据是进程隔离的。你请求/try接口时往当前进程的队列写了数据，后续请求/oneschedule可能被调度到另一个空队列的进程，自然返回失败。
• 内存队列的数据会在服务重启后全部清空，仅适合临时测试用，不适合生产环境。
• 原/oneschedule接口失败时返回的是纯字符串，和成功时的JSON格式不统一，也可能导致调用方解析失败。

方案1：本地测试临时修复 #

如果仅在本地做功能测试，不需要多worker部署，直接调整Flask启动参数强制单进程单线程运行即可：
启动命令修改为：
flask run --no-reload --without-threads
如果是代码中直接启动，修改入口代码：

if __name__ == '__main__':
    app.run(debug=False, use_reloader=False, threaded=False)

同时统一接口返回格式，避免解析异常：

@app.route('/oneschedule',methods=['GET'])
def oneschedule():
    if not q_sample.empty():
        return jsonify({"code": 0, "schedule": q_sample.get(block=False)})
    else:
        return jsonify({"code": -1, "msg": "Failed"}), 200

方案2：生产环境可用实现 #

使用独立的第三方中间件做队列存储，推荐用Redis的list结构，天然支持跨进程共享、数据持久化，实现成本很低：
首先安装Redis依赖：
pip install redis
修改代码如下：

import requests
import json
import redis
from flask import Flask, jsonify

app = Flask(__name__)
# 初始化Redis连接，参数根据实际环境修改
redis_client = redis.Redis(
    host="127.0.0.1",
    port=6379,
    db=0,
    decode_responses=True
)
QUEUE_KEY = "schedule_task_queue"

@app.route('/try',methods=['GET'])
def getschedule():
    resp = requests.get("https://api.postalpincode.in/pincode/620012", verify=False)
    j_data = json.loads(resp.content)
    for n in j_data:
        post_offices = n['PostOffice']
        for item in post_offices:
            # 字典转JSON字符串写入Redis队列右侧
            redis_client.rpush(QUEUE_KEY, json.dumps(item))
    return jsonify({"size": redis_client.llen(QUEUE_KEY)})

@app.route('/oneschedule',methods=['GET'])
def oneschedule():
    # 从Redis队列左侧弹出一条数据，为空返回None
    item_str = redis_client.lpop(QUEUE_KEY)
    if item_str:
        return jsonify({"code": 0, "schedule": json.loads(item_str)})
    else:
        return jsonify({"code": -1, "msg": "Failed"}), 200

该方案不受Flask worker数量影响，所有进程共享同一个Redis队列，服务重启数据也不会丢失，可直接用于生产环境。

内容的提问来源于stack exchange，提问作者Manoj Selvam