Flask框架内容基础3 -- 使用redis实现异步任务队列

前面所了解的所有请求都是同步的，那么当面临异步请求时，应该怎么做？

调用者：携带参数发送请求

API：接收请求并生成一个任务ID，接下来：返回给调用者+放到任务队列中

worker：等待redis队列（List），一旦接收到任务，就执行并将结果返回到结果队列（Hash）

调用者：等待n秒后，携带任务ID再次发送请求，获取结果

API：接收任务ID，根据ID去结果队列（Hash）中获取 

一、调用者调用逻辑代码：

import json
import uuid


import redis
from flask import Flask, jsonify, send_file, request

app = Flask(__name__)


@app.route('/task', methods=['POST'])
def task():
    task_str = request.json.get("order_string")
    # 生成任务id
    tid = str(uuid.uuid4())
    # 将任务放入任务队列中
    info_dict = {'tid': tid, 'data': task_str}
    # 连接redis
    REDIS_CONN_PARAMS = {
        "host": '127.0.0.1',
        'password': '123456',
        'port': 6379,
        'encoding': 'utf-8'
    }
    conn = redis.Redis(**REDIS_CONN_PARAMS)
    conn.lpush("spider_task_list",json.dumps(info_dict))
    return jsonify({'status': True, 'data': tid, 'message': "正在处理，预计等待1分钟"})


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080, debug=True)

二、工作者对任务队列任务进行处理 

import json
import uuid

import redis
from flask import Flask, jsonify, send_file, request


def get_task():
    REDIS_CONN_PARAMS = {
        "host": '127.0.0.1',
        'password': '123456',
        'port': 6379,
        'encoding': 'utf-8'
    }
    conn = redis.Redis(**REDIS_CONN_PARAMS)
    data = conn.brpop("spider_task_list", timeout=10)
    return json.loads(data[1].decode('utf-8'))


def set_result(tid, res_info):
    # 将处理结果放入结果队列
    REDIS_CONN_PARAMS = {
        "host": '127.0.0.1',
        'password': '123456',
        'port': 6379,
        'encoding': 'utf-8'
    }
    conn = redis.Redis(**REDIS_CONN_PARAMS)
    conn.hset("spider_result_list", tid, res_info)


def run():
    while True:
        # 1.去redis中获取任务
        # 连接redis
        task_info = get_task()
        if not task_info:
            continue
        # todo 处理任务
        res_info = ''
        tid = task_info['tid']
        set_result(tid, res_info)

三、调用者再次调用api获取处理好的结果：

@app.route('/result', methods=['GET'])
def result():
    # 根据任务id去结果队列中寻找
    tid = request.args.get('tid')
    if not tid:
        return jsonify({'status': False, 'error': '参数错误'})
    REDIS_CONN_PARAMS = {
        "host": '127.0.0.1',
        'password': '123456',
        'port': 6379,
        'encoding': 'utf-8'
    }
    conn = redis.Redis(**REDIS_CONN_PARAMS)
    data = conn.hget("spider_result_list", tid)

    if not data:
        return jsonify({"status": True, 'data': "", 'message': "未完成，请继续等待"})
    data_str = data.decode('utf-8')
    return jsonify({'status': True, 'data': data_str})

优化:上述在连接redis时遇到了与连接数据库时一样的问题，所以我们可以比葫芦画瓢，创建redis连接池。

REDIS_POOL = redis.ConnectionPool(host='127.0.0.1', password='123456', port=6379, encoding='utf-8', max_connections=100)