已配置Backend但Celery分布式任务无法获取结果（DisabledBackend错误）

问题背景 #

我正在实现一个MLOps应用，将Celery Worker部署在Docker容器中，从本地环境发送任务消息。当前运行三个容器：

• multi_label_text_classification_celery_server_1：负责训练与推理
• multi_label_text_classification_redis_1：消息中间件与结果存储
• multi_label_text_classification_triton_server_1：模型推理服务

推理任务可以通过本地代码远程调用成功：

import pandas as pd
import json
from celery_app import predict  # 假设本地Celery实例在celery_app.py中

dataset = pd.read_json('data.json')
data = dataset.text.values.tolist()
j_string = json.dumps(data, ensure_ascii=False)

predict_task = predict.apply_async(
    args=(
        'audience_bert',
        1,
        100,
        j_string
    )
)
print(predict_task)

执行后得到任务ID：758d7455-af2d-494e-8ba9-f9e502a8727c

但尝试检查任务状态并获取结果时：

from celery.result import AsyncResult
from celery_app import app  # 本地Celery实例

result = AsyncResult(task_id, app=app)
print(result.state)
print(result.get())

出现DisabledBackend错误，尽管容器内Worker和本地Celery应用都已配置Redis作为结果后端：

Traceback (most recent call last):
  File "<input>", line 1, in <module>
  File "C:\Users\ychuang\AppData\Local\pypoetry\Cache\virtualenvs\celery-client-HqyYFMWr-py3.8\lib\site-packages\celery\result.py", line 478, in state
    return self._get_task_meta()['status']
  File "C:\Users\ychuang\AppData\Local\pypoetry\Cache\virtualenvs\celery-client-HqyYFMWr-py3.8\lib\site-packages\celery\result.py", line 417, in _get_task_meta
    return self._maybe_set_cache(self.backend.get_task_meta(self.id))
  File "C:\Users\ychuang\AppData\Local\pypoetry\Cache\virtualenvs\celery-client-HqyYFMWr-py3.8\lib\site-packages\celery\backends\base.py", line 609, in get_task_meta
    meta = self._get_task_meta_for(task_id)
AttributeError: 'DisabledBackend' object has no attribute '_get_task_meta_for'

我已经排查了常见的DisabledBackend原因（缺少后端配置），但本地和容器内都已配置Redis后端，恳请帮忙解决。

相关配置代码 #

docker-compose.yml #

version: "3"

services:
  celery_server:
    env_file: .env
    build:
      context: .
      dockerfile: Dockerfile
    volumes:
      - models:/model/torch_script
    environment:
      LEVEL: ${LEVEL}
    links:
      - redis
    depends_on:
      - redis

  redis:
    image: redis:latest
    hostname: redis
    ports:
      - 6379:6379

  triton_server:
    image: nvcr.io/nvidia/tritonserver:22.06-py3
    hostname: triton
    ports:
      - 8000:8000
      - 8001:8001
      - 8002:8002
    command: ["tritonserver", "--model-store=/models", "--model-control-mode=poll", "--repository-poll-secs=30"]
    volumes:
      - models:/models
    shm_size: 1g
    ulimits:
      memlock: -1
      stack: 67108864

volumes:
  models:

容器内Celery Worker代码 #

import json
from typing import Dict

from celery import Celery

from config.settings import MODEL_CKPT, LogDir
from utils.inference_helper import chunks
from utils.log_helper import create_logger
from worker.inference.bert_triton_inference import BertInferenceWorker
from worker.train.chinese_bert_classification import ChineseBertClassification

app = Celery(
    name='bert_celery',
    broker="redis://redis:6379/0",
    backend="redis://redis:6379/1"
)

app.conf.task_routes = {
    'app.*': {'queue': 'deep_model'},
}

app.conf.update(result_expires=1)
app.conf.update(task_track_started=True)


@app.task(bind=True, queue='deep_model', name='training')
def training(
        self,
        model_name,
        version,
        dataset,
        label_col,
        learning_rate=2e-5,
        epochs=50,
        batch_size=32,
        max_len=30,
        is_multi_label=1,
        ckpt=MODEL_CKPT.get('chinese-bert-wwm')

):
    dataset = json.loads(dataset)
    label_col = json.loads(label_col)

    task_worker = ChineseBertClassification(
        max_len=max_len,
        ckpt=ckpt,
        epochs=epochs,
        learning_rate=learning_rate,
        batch_size=batch_size,
        dataset=dataset,
        label_col=label_col,
        model_name=model_name,
        model_version=version,
        is_multi_label=is_multi_label
    )

    task_worker.init_model()
    results: Dict[str, str] = task_worker.run()
    return results


@app.task(bind=True, queue='deep_model', name='predict')
def predict(self, model_name, version, max_len, dataset):
    logger = create_logger(LogDir.inference)
    data = json.loads(dataset)

    output = []
    for idx, chunk in enumerate(chunks(data, 32)):
        logger.info(f" ==== batch: {idx} ==== ")
        infer_worker = BertInferenceWorker(
            dataset=chunk,
            model_name=model_name,
            model_version=version,
            url='triton:8000',
            backend='pytorch',
            max_len=max_len,
            chunk_size=len(chunk)
        )
        results = infer_worker.run()
        output.extend(results.tolist())

    assert len(output) == len(data)

    return json.dumps(output, ensure_ascii=False)

本地Celery客户端代码 #

from celery import Celery


app = Celery(
    name='bert_celery',
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1"
)

app.conf.task_routes = {
    'app.*': {'queue': 'deep_model'},
}


@app.task(bind=True, queue='deep_model', name='training')
def training(
        self,
        model_name,
        version,
        dataset,
        label_col,
        learning_rate,
        epochs,
        batch_size,
        max_len,
        is_multi_label
):
    pass


@app.task(bind=True, queue='deep_model', name='predict')
def predict(
        self,
        model_name,
        version,
        max_len,
        dataset
):
    pass

排查与解决方向 #

1. 绑定正确的Celery实例到AsyncResult
   本地获取结果时，必须确保AsyncResult关联的是已配置好后端的本地Celery app实例，不能使用默认未配置的实例。修改获取结果的代码：

   from celery.result import AsyncResult
   from your_local_celery_module import app  # 导入本地定义的Celery app
   
   result = AsyncResult(task_id, app=app)
   

2. 验证本地到Redis的连接
   本地环境执行redis-cli ping，检查是否能正常访问localhost:6379，确认Redis容器端口映射正常。

3. 检查后端配置一致性
   确认容器内Worker与本地客户端的Redis后端配置完全匹配：容器内用redis://redis:6379/1，本地用redis://localhost:6379/1是正确的（Redis容器端口已映射到本地6379）。

4. 核对Celery版本
   本地客户端与容器内Worker的Celery版本必须一致，版本不兼容可能导致后端通信异常。

5. 检查Redis内的任务结果
   进入Redis容器，执行redis-cli SELECT 1，再用KEYS "*758d7455-af2d-494e-8ba9-f9e502a8727c*"查看是否存在该任务的结果键，确认Worker是否成功将结果写入Redis。

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