跨服务器调用Celery任务报错：未注册任务'tasks.test'如何解决？

问题背景 #

两台服务器部署架构：

• 服务器1：Django应用 + Redis（作为Celery Broker）
• 服务器2：Celery Worker

Celery服务器（服务器2）的task.py代码：

from celery import Celery

app = Celery('tasks')
app.conf.broker_url = 'redis://localhost:6379/0'

@app.task(bind=True)
def test():
    print('Testing')

Django服务器（服务器1）调用任务的代码：

from celery import Celery
celery = Celery()
celery.conf.broker_url = 'redis://localhost:6379/0'
celery.send_task('tasks.test')

启动Celery Worker的命令：

celery -A tasks worker --loglevel=INFO

调用时触发错误：

收到类型为'tasks.test'的未注册任务。消息已被忽略并丢弃。
您是否记得导入包含该任务的模块？或者您是否使用了相对导入？

解决方法 #

1. 修正Redis连接地址 #

如果Redis部署在Django服务器（服务器1），Celery服务器（服务器2）的broker_url不能写localhost，必须替换为Django服务器的公网IP或内网地址，否则Worker无法连接到正确的Broker，导致任务无法被正确注册和接收。

修改服务器2的task.py：

from celery import Celery

# 替换为Django服务器的实际IP/域名
app = Celery('tasks', broker='redis://<Django服务器IP>:6379/0')

@app.task(bind=True)
def test():
    print('Testing')

2. 确保Worker正确加载任务模块 #

启动Worker时，确认task.py在当前工作目录，或者通过绝对路径指定模块：

# 替换为task.py所在的绝对路径
celery -A /full/path/to/tasks worker --loglevel=INFO

启动后查看Worker日志，确认是否有tasks.test任务被注册的日志信息，若没有则说明Worker未找到任务模块，需调整路径。

3. 统一任务命名与显式注册 #

给任务指定明确的名称，避免命名匹配问题：

from celery import Celery

app = Celery('tasks', broker='redis://<Django服务器IP>:6379/0')

# 显式指定任务名称，确保和调用端一致
@app.task(bind=True, name='tasks.test')
def test():
    print('Testing')

# 或者让Celery自动发现任务模块
app.autodiscover_tasks(['tasks'])

4. 更可靠的调用方式：共享任务模块 #

将task.py作为公共代码同步到Django服务器和Celery服务器，Django端直接导入任务调用，无需使用send_task：

在Django项目中添加task.py（和服务器2的代码一致），然后调用：

from tasks import test
# 使用delay()或apply_async()触发任务
test.delay()

这种方式依赖Celery自动管理任务名称，能有效避免名称匹配错误。

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