记录在 Flask 项目中使用 Redis 实现任务队列的配置与示例，这里选用 RQ（Redis Queue）作为任务处理库。

一、安装依赖

pip install redis rq

需要确保 Redis 服务已启动（默认端口 6379）。

二、项目结构示例

myapp/
├── app.py
├── tasks.py

三、任务定义（tasks.py）

import time

def background_task(n):
    print(f"开始任务 {n}")
    time.sleep(n)
    print(f"任务 {n} 完成")
    return f"任务 {n} 完成"

四、Flask 主程序（app.py）

from flask import Flask, request, jsonify
from redis import Redis
from rq import Queue
from tasks import background_task

app = Flask(__name__)
redis_conn = Redis()
task_queue = Queue(connection=redis_conn)

@app.route('/add-task', methods=['POST'])
def add_task():
    seconds = int(request.form.get('seconds', 5))
    job = task_queue.enqueue(background_task, seconds)
    return jsonify({"job_id": job.get_id()})

@app.route('/task-status/<job_id>')
def task_status(job_id):
    job = task_queue.fetch_job(job_id)
    if job:
        return jsonify({"status": job.get_status(), "result": job.result})
    return jsonify({"error": "任务不存在"}), 404

五、启动 Worker

在项目根目录执行：

rq worker

Worker 会监听队列并执行任务。

六、运行流程

  1. 启动 Redis 服务
  2. 启动 Flask 应用
  3. 启动 RQ Worker
  4. 调用 /add-task 添加任务
  5. 通过 /task-status/<job_id> 查询任务状态

七、小结

  • RQ 使用简单，适合中小型异步任务场景
  • Redis 提供持久化与多 worker 支持
  • 适合延迟执行、后台计算、批量任务等需求