【web服务】耗时任务基于API与worker模式

• 场景：当我们爬虫或者获取数据需要很长的时间计算或者等待，这种情况情况下基于API与worker模式完成交付
• 1.API端:用flask框架

import json
import uuid
import redis
from flask import Flask, request, jsonify

app = Flask(__name__)
pool = redis.ConnectionPool(host='127.0.0.1', port=6379, password='123456',encoding='utf-8', max_connections=1000)

@app.route("/task",method=["POST"])
def task():
    # 构建参数
    total_dict = {}
    user_dict = request.json
    total_dict.update(user_dict)
    total_dict.update({"token":"1231312","sign":"dsfewfwef"})

    # 创建任务ID
    task_id = str(uuid.uuid4())

    # 放入redis
    conn = redis.Redis(connection_pool=pool)
    task_info = {"uid":task_id,"info":total_dict}
    conn.lpush("task_queue",json.dumps(task_info))

    return jsonify({"status":True,"data":task_id})

@app.route("/fetch",method=["GET"])
def fetch():
    # 通过url传递任务id
    task_id = request.args.get("task_id")
    conn = redis.Redis(connection_pool=pool)
    result = conn.hget("task_result",task_id)
    if not result:
        return jsonify({"status":False,"data":"任务还在执行中"})

    conn.hdel("task_result",task_id)
    result_string = result.decode('utf-8')
    return jsonify({"status":True,"data":result_string})

if __name__ == '__main__':
    app.run(host="127.0.0.1", port=9999)

• 2.客户端

import time

import requests

# 1.发送请求（创建任务）
res = requests.post(
    url="http://127.0.0.1:9999/task",
    json={
        "aid": 1,
        "bfc": True,
        "ctime": int(time.time())
    }
)
# {"status": True, "data": "fa5d4f0d-9d29-4008-8097-d1a7d5f9c2cd"}
res_dict = res.json()
task_id = res_dict['data']

# 2.查询结果
while True:
    time.sleep(5)
    res = requests.get(
        url="http://127.0.0.1:9999/fetch",
        params={
            "task_id": task_id
        }
    )
    # {"status": False, "data": "任务还在执行中"}
    # {"status": True, "data": result_string}
    res_dict = res.json()
    if not res_dict['status']:
        print(res_dict['data'])
        continue

    print("执行完毕", res_dict['data'])
    break

• 3.worker

import random
import time

import redis
import json

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

def run():
    while True:
        conn = redis.Redis(connection_pool=POOL)
        res = conn.brpop(["task_queue"], timeout=30)  # ("task_queue",b"{}")
        print("取任务：", res)
        if not res:
            continue
        task_dict = json.loads(res[-1].decode('utf-8'))
        task_id = task_dict['uid']

        # 根据任务信息去执行任务，耗时：5分钟、1小时
        time.sleep(random.randint(5, 10))

        # 结算结果放入结果队列
        conn = redis.Redis(connection_pool=POOL)
        conn.hset("task_result", task_id, "执行结果")
"""
redis = {
	"task_result":{
		"fa5d4f0d-9d29-4008-8097-d1a7d5f9c2cd":"执行结果",
		"fa5d4f0d-9d29-4008-8097-d1a7d5f9c3cd":"执行结果",
	}
}
"""


if __name__ == '__main__':
    run()

• 4.运行效果
  

• 5.此场景适用于计算密集型与长时间耗时任务，我们将API部署在远程服务器上，使用本地电脑当做worker，可以节约成本