如何在后台任务中访问Redis连接？基于flask-base扩展RQ任务

我来帮你搞定这个问题！在flask-base这类用工厂模式搭建的Flask项目里，要让RQ的Worker和视图共用Redis连接，核心思路是把Redis连接和RQ队列实例做成可复用的共享模块，避免两边重复创建连接，同时适配项目的结构。下面是具体的实现步骤：

1. 创建共享的RQ配置模块 #

首先在app目录下新建一个tasks.py文件，专门用来管理Redis连接、RQ队列实例，以及定义后台任务。这样Worker和视图都能从这里导入统一的实例：

import os
from redis import Redis
from rq import Queue

# 读取Redis配置（兼容本地开发和Heroku等生产环境）
# flask-base本身就用环境变量管理配置，这里保持一致
redis_url = os.getenv('REDISTOGO_URL', 'redis://localhost:6379/0')
# 创建全局Redis连接实例
redis_conn = Redis.from_url(redis_url)
# 创建RQ队列实例（可以指定队列名称，比如'email'、'processing'，默认用'default'）
task_queue = Queue(connection=redis_conn)

2. 改造Worker启动脚本 #

不管你原来的run_worker()是放在单独的脚本里，还是集成到manage.py的命令中，现在都要导入上面的共享连接和队列，而不是在Worker里重新创建：

比如项目根目录的run_worker.py可以改成这样：

from rq import Worker, Connection
from app.tasks import redis_conn, task_queue

def run_worker():
    # 使用共享的Redis连接初始化RQ
    with Connection(redis_conn):
        # 监听指定队列（这里用我们定义的task_queue）
        worker = Worker([task_queue])
        worker.work()

if __name__ == '__main__':
    run_worker()

如果是用Flask的CLI命令（比如在manage.py里），可以加一个自定义命令：

# 在manage.py里添加
from flask_script import Manager
from app.tasks import redis_conn, task_queue
from rq import Worker, Connection

manager = Manager(create_app)

@manager.command
def run_worker():
    with Connection(redis_conn):
        worker = Worker([task_queue])
        worker.work()

if __name__ == '__main__':
    manager.run()

这样启动Worker的时候，直接运行python manage.py run_worker或者python run_worker.py就行，用的是和视图一致的Redis连接。

3. 在视图中调用RQ队列 #

现在你可以在任意视图里导入共享的task_queue，直接把后台任务丢进去执行。比如：

首先在tasks.py里定义你的后台任务（注意：如果任务需要访问Flask上下文，比如数据库查询，要手动推送上下文）：

# 继续在app/tasks.py里添加
def process_user_task(user_id):
    # 因为RQ Worker运行在独立进程，需要手动初始化Flask上下文
    from app import create_app
    app = create_app()
    with app.app_context():
        from app.models import User
        user = User.query.get(user_id)
        # 这里写你的耗时逻辑：比如生成报表、发送邮件、处理上传文件等
        print(f"Processing user: {user.username}")
        return f"Completed processing for {user.username}"

然后在视图里调用这个任务：

# 比如在app/main/views.py里
from flask import redirect, url_for, flash
from app.tasks import task_queue, process_user_task
from app.models import User

@main.route('/trigger-task/<int:user_id>')
def trigger_task(user_id):
    # 将任务加入RQ队列
    job = task_queue.enqueue(process_user_task, user_id)
    flash(f"后台任务已启动，任务ID：{job.id}")
    return redirect(url_for('main.user_profile', user_id=user_id))

几个关键注意事项 #

• 上下文问题：RQ Worker是独立于Flask应用的进程，所以任务里如果要使用数据库、Flask配置等，必须手动创建并推送Flask上下文，不然会出现"应用未初始化"的错误。
• 环境变量：确保生产环境（比如Heroku）的Redis URL环境变量正确配置，本地开发可以用默认的redis://localhost:6379/0。
• 任务独立性：后台任务里的逻辑要尽量独立，避免依赖视图里的临时变量，最好通过参数传递必要的数据（比如用户ID，而不是直接传递User对象）。

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