FastAPI集成Redis RQ与OpenCV时任务执行失败求助

架构与流程 #

• server.py：FastAPI接口定义，接收请求后调用redis_processes模块将任务加入队列
• redis_processes.py：管理Redis连接，创建RQ队列，负责将任务分发到对应队列
• worker_tasks.py：包含任务执行逻辑，由后台Redis RQ Worker进程处理

流程：FastAPI接口接收输入 → 调用redis_processes将任务加入Redis队列 → 后台Worker执行worker_tasks.py中的任务逻辑

错误现象 #

只要在worker_tasks.py中导入cv2（OpenCV），Worker执行任务时就会触发错误：Work-horse terminated unexpectedly; waitpid returned 6 (signal 6)，且该错误不会在终端直接显示。移除cv2导入后，任务可正常执行。

核心代码片段 #

server.py 片段 #

from fastapi import FastAPI, UploadFile, File, Query, HTTPException, Security, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from typing import List, Optional
from models import FeedbackTask
import redis_processes

app = FastAPI()
security = HTTPBearer()

# 端点与认证细节已省略
@app.post("/processfeedback/")
async def process_feedback(...): # 参数与处理逻辑省略
    return redis_processes.process_feedback(feedback_task)

redis_processes.py 片段 #

import redis
from rq import Queue
from redis_settings import workers_settings

import threading
import time
from datetime import datetime

from models import FeedbackTask
import worker_tasks as worker_tasks

# 建立Redis连接
redis_conn = redis.Redis(host='localhost', port=6379, db=0)

ai_task_processor_q         = Queue(name= workers_settings[0]['q'],connection=redis_conn)
initial_db_writer_q         = Queue(name= workers_settings[1]['q'],connection=redis_conn)
ai_task_result_writer_q     = Queue(name= workers_settings[2]['q'],connection=redis_conn)

def process_feedback(feedback_process_task: FeedbackTask):
    now = datetime.now()
    timestamp = int(time.mktime(now.timetuple()))
    feedback_process_task.collection_id = f"{feedback_process_task.user_id}_{timestamp}"

    job_db_w = initial_db_writer_q.enqueue(worker_tasks.add_raw_collection_to_db, feedback_process_task, result_ttl=-1)
    job_ai   = ai_task_processor_q.enqueue(worker_tasks.process_feedback, feedback_process_task, result_ttl=-1)

    return {"AI Feedback Process Job Status": job_ai.get_status(),
            "Raw Document Collection Write Status": job_db_w.get_status()}

worker_tasks.py 片段 #

import cv2  # 此导入引发问题
from models import FeedbackTask

def process_feedback(task: FeedbackTask) -> FeedbackTask:
    # 任务处理逻辑省略

排查方向与解决建议 #

问题根源分析 #

Signal 6对应SIGABRT，通常是进程自身触发的终止信号，常见原因包括：

• OpenCV的多线程/进程兼容性问题：Redis RQ Worker采用fork机制创建子进程执行任务，而OpenCV的某些组件（如GUI模块、初始化逻辑）在fork环境下可能出现资源冲突或初始化失败
• 环境依赖不匹配：Worker环境与主服务环境的OpenCV版本、依赖库（如libopencv、numpy）版本不一致，导致导入时崩溃
• 权限或资源限制：Worker进程没有足够权限访问OpenCV依赖的系统资源，或资源不足触发崩溃

具体排查步骤 #

1. 启用Worker详细日志：启动Worker时添加--verbose参数，或配置RQ的日志级别为DEBUG，查看是否有隐藏的错误输出，命令示例：rq worker --verbose ai_task_processor_q
2. 单独测试OpenCV导入：创建独立脚本仅导入cv2并执行简单操作（如cv2.imread），在Worker运行的环境中直接运行，确认是否能正常执行
3. 检查环境依赖一致性：对比主服务与Worker环境的Python版本、cv2版本、numpy版本，确保完全一致，可通过pip list命令查看
4. 禁用OpenCV的GUI模块：如果任务不需要GUI功能，导入cv2前设置环境变量禁用相关模块，示例：

   import os
   os.environ['OPENCV_IO_ENABLE_JASPER'] = '0'
   os.environ['QT_QPA_PLATFORM'] = 'offscreen'
   import cv2
   

5. 使用预初始化的Worker：修改RQ Worker的启动方式，避免fork子进程，或在Worker启动时提前导入cv2，而非在任务函数中导入，示例：
   • 创建worker_bootstrap.py，提前导入cv2：

     import cv2
     from rq import Worker, Queue, Connection
     import redis
     
     redis_conn = redis.Redis(host='localhost', port=6379, db=0)
     queues = ['ai_task_processor_q', 'initial_db_writer_q', 'ai_task_result_writer_q']
     
     if __name__ == '__main__':
         with Connection(redis_conn):
             worker = Worker(queues)
             worker.work()
     

   • 用此脚本启动Worker：python worker_bootstrap.py

解决建议 #

• 优先确保Worker与主服务环境依赖完全一致，推荐使用虚拟环境隔离
• 若任务无需OpenCV的GUI功能，强制禁用相关模块以避免初始化冲突
• 采用Worker预初始化方式，提前加载cv2，避免在任务执行时动态导入引发的fork问题

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