Docker部署FastAPI+Redis+RQ时，Worker无法定位任务模块

我通过docker-compose部署了FASTAPI_SERVER、REDIS_SERVER和WORKER_SERVER，希望借助Redis和RQ为FastAPI实现后台任务功能。

FastAPI端任务入队代码 #

@app.get("/api/greet")
async def greet():
    try:
        redis_connection = Redis(host=host, port=port, password=password)
        queue = Queue(name=report_queue, connection=redis_connection)
        logger.debug(f"queue created successfully! {queue.connection}")

        job = queue.enqueue("app.tasks.generate_report_task", {"name": "test"})
        logger.debug(f"job created successfully! {job.id}")
    except Exception as e:
        logger.error(f"queue creation failed: {e}")
        raise Exception(f"queue creation failed: {e}")
    return {"message": "Hello World"}

Worker服务目录结构 #

──> ls           
app  Dockerfile  requirements.txt  scripts  supervisord.conf  venv

──> cd app && tree
├── config.py
├── database.py
├── __init__.py
├── model.py
├── reports
│	├── generation.py
│	├── __init__.py
│	└── utils.py
├── tasks.py
└── templates
    └── ppt_template.pptx

FastAPI入队成功日志 #

fastapi-4          | 2024-01-26 08:57:53.463 | DEBUG    | app.main:greet:176 - queue created successfully! Redis<ConnectionPool<Connection<host=redis,port=6379,db=0>>>
fastapi-4          | 2024-01-26 08:58:00.584 | DEBUG    | app.main:greet:179 - job created successfully! 71dc6cb5-29b6-4d30-b078-5b315bd8ea44

Worker执行报错日志 #

worker_server      | 2024-01-26 08:58:01,153 DEBG 'worker-0' stderr output:
worker_server      | 08:58:01 [Job 71dc6cb5-29b6-4d30-b078-5b315bd8ea44]: exception raised while executing (app.tasks.generate_report_task)
worker_server      | Traceback (most recent call last):
worker_server      |   File "/home/appuser/venv/lib/python3.10/site-packages/rq/utils.py", line 118, in import_attribute
worker_server      |     attribute_owner = getattr(module, attribute_owner_name)
worker_server      | AttributeError: module 'app' has no attribute 'tasks'
worker_server      | 
worker_server      | During handling of the above exception, another exception occurred:
worker_server      | 
worker_server      | Traceback (most recent call last):
worker_server      |   File "/home/appuser/venv/lib/python3.10/site-packages/rq/worker.py", line 1428, in perform_job
worker_server      |     rv = job.perform()
worker_server      |   File "/home/appuser/venv/lib/python3.10/site-packages/rq/job.py", line 1278, in perform
worker_server      |     self._result = self._execute()
worker_server      |   File "/home/appuser/venv/lib/python3.10/site-packages/rq/job.py", line 1315, in _execute
worker_server      |     result = self.func(*self.args, **self.kwargs)
worker_server      |   File "/home/appuser/venv/lib/python3.10/site-packages/rq/job.py", line 425, in func
worker_server      |     return import_attribute(self.func_name)
worker_server      |   File "/home/appuser/venv/lib/python3.10/site-packages/rq/utils.py", line 120, in import_attribute
worker_server      |     raise ValueError('Invalid attribute name: %s' % attribute_name)
worker_server      | ValueError: Invalid attribute name: generate_report_task

本地非Docker测试场景（可正常运行） #

入队代码enqueue_tasks.py #

from rq import Queue
from redis import Redis

if __name__ == "__main__":
    redis_conn = Redis()
    queue = Queue(name="report-generation", connection=redis_conn)

    job = queue.enqueue("worker.tasks.example_task", kwargs={"x": 1, "y": 2})
    print(f"Task enqueued with job ID: {job.id}")

任务代码worker/tasks.py #

import asyncio

async def example_task(**kwargs):
    print("started")
    await asyncio.sleep(5)
    return kwargs.get("x", 1) + kwargs.get("y", 2)

启动Worker命令 #

rq worker report-generation

问题原因及解决办法 #

1. 修复Worker容器的Python路径 #

RQ Worker找不到app.tasks模块，核心原因是Worker容器的Python解释器无法定位到app模块所在目录。

• 确认Worker启动时的工作目录是app文件夹的父目录（即执行ls能看到app的目录）。
• 如果用supervisord管理Worker，修改supervisord.conf中的命令，先切换到正确目录再启动Worker：

  [program:worker]
  command=cd /home/appuser && rq worker report-generation
  

• 或者在Worker的Dockerfile中设置工作目录：

  WORKDIR /home/appuser
  

2. 验证模块导入有效性 #

进入Worker容器，执行以下命令测试模块导入：

python -c "import app.tasks; print(app.tasks.generate_report_task)"

如果报错，说明路径仍有问题，可手动添加路径到sys.path：

import sys
sys.path.append("/path/to/app-parent-dir")
import app.tasks

3. 统一任务引用路径 #

确保FastAPI中enqueue的任务路径（app.tasks.generate_report_task）和Worker容器中的模块结构完全一致。如果Worker容器中的任务模块路径不同，调整FastAPI的引用路径为Worker能识别的格式。

4. 处理异步任务兼容问题 #

RQ原生不支持异步函数（async def），需做兼容处理：

• 方案一：用asyncio.run()包裹异步逻辑，修改tasks.py：

  import asyncio
  
  def generate_report_task(**kwargs):
      async def _inner_task():
          # 原异步任务逻辑
          pass
      return asyncio.run(_inner_task(**kwargs))
  

• 方案二：使用rq-async扩展来原生支持异步任务。

内容的提问来源于stack exchange，提问作者Ashish Kumar Jha