Django项目中Celery共享任务导入失败及任务未注册问题求助

I've run into similar headaches when packaging shared Celery tasks into a reusable library for Django projects, so let's break down your problem and walk through actionable fixes:

Context of Your Issue #

You're using Celery 4.1.0, trying to trigger a task from futurelibrary.tasks via app.send_task() but hitting a NotRegistered('futurelibrary.tasks.taskfunction') error. The auto-discovery feature isn't working as expected, and you can't add imports to __init__.py because you need to support multiple Celery configurations.

Troubleshooting Steps & Solutions #

1. Turn futurelibrary into a Valid Python Module #

Right now, your futurelibrary folder only contains tasks.py—Python won't recognize it as an importable module without an __init__.py file. Even an empty one will resolve this:

myproject
├──futurelibrary
│   ├── __init__.py  # Add this empty file
│   └──tasks.py

2. Confirm Celery Can Locate the futurelibrary Module #

Celery needs the parent directory of futurelibrary in its Python path. Add a quick debug check in djangoapp2/celery.py to verify:

from __future__ import absolute_import
import sys
from celery import Celery
from settings import celeryconfig

# Debug: Print current Python path
print("Python Path:", sys.path)

app = Celery('my_app')
app.config_from_object(celeryconfig)
app.autodiscover_tasks(['futurelibrary'])

If myproject isn't in the path, you can add it explicitly (though this shouldn't be needed if you're running the worker from the project root):

sys.path.append('/absolute/path/to/myproject')

3. Ensure Tasks Use shared_task Correctly #

Double-check that your task in futurelibrary/tasks.py uses the @shared_task decorator (not bound to a specific Celery app):

from celery import shared_task

@shared_task
def taskfunction():
    # Your task logic here
    return "Task completed successfully"

Avoid @app.task here—it ties the task to a single Celery instance, which breaks your multi-config requirement.

4. Fix Conflicting Configurations in workerconf.py #

Your worker starts with celery worker --config=settings.workerconf—make sure this config doesn't override auto-discovery settings:

• Look for CELERY_IMPORTS or CELERY_INCLUDE in workerconf.py—these can interfere with autodiscover_tasks.
• Explicitly specify your Celery app when starting the worker to avoid loading a default instance:

  celery worker --app=djangoapp2.celery:app --config=settings.workerconf
  

5. Force Task Import via Config (Alternative to Auto-Discovery) #

If auto-discovery still fails, directly tell Celery to import your tasks in celeryconfig.py:

# celeryconfig.py
CELERY_IMPORTS = ['futurelibrary.tasks']

This bypasses auto-discovery and ensures the task module loads when Celery starts.

6. Validate the Task Name #

Sometimes the task name doesn't match what you're calling. Print the actual task name to confirm:

# In futurelibrary/tasks.py
print("Registered task name:", taskfunction.name)

If you set a custom name with @shared_task(name="custom_task_name"), use that name in send_task() instead of the default futurelibrary.tasks.taskfunction.

7. Check Worker Logs for Task Registration #

Start the worker with verbose logging to see if the task is being registered:

celery worker --app=djangoapp2.celery:app --config=settings.workerconf --loglevel=info

Look for a line like this in the logs:

[INFO/MainProcess] Registered tasks:
    ...
    'futurelibrary.tasks.taskfunction',
    ...

If the task isn't listed here, circle back to steps 1-4 to fix module import or configuration issues.

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