1

好吧，伙计们，这让我发疯了。我已经为此工作了一整天，但无法让它工作！
我的芹菜项目结构是这样的： 

# celery.py

from celery.schedules import crontab
from celery import Celery

celery = Celery('scheduler.celery',
                include=['scheduler.tasks'])

celery.config_from_object('celeryconfig')

和：

# tasks.py

from scheduler.celery import celery

@celery.task
def test():
    do_something()

和：

# celeryconfig.py

from celery.schedules import crontab

CELERYBEAT_SCHEDULE = {
    'test-cron': {
        'task': 'tasks.test',
         'schedule': crontab(minute='*/1'),
    },
}
# CELERY_IMPORTS = ('tasks', )
BROKER_URL = 'redis://localhost:6379/0'

所有文件都在projects/scheduler/文件夹下。
当我启动celeryd服务时，我可以看到它正在运行并连接到我的代理，但是当我启动celerybeat服务时，我可以在日志中看到消息：Received unregistered task of type 'tasks.test'.
如果我取消注释CELERY_IMPORTS常量（如 SO 中的许多答案中所建议的那样），celeryd服务甚至都不会启动！实际上，它输出OK但使用ps ef | grep celery我可以看到它没有运行。

我的守护进程 conf 文件如下所示： 

# Name of nodes to start
CELERYD_NODES="w1"

# Where to chdir at start.
CELERYD_CHDIR="/home/me/projects/scheduler/"

# Extra arguments to celeryd
CELERYD_OPTS="--time-limit=300 --concurrency=4"

# %n will be replaced with the nodename.
CELERYD_LOG_FILE="/var/log/celery/%n.log"
CELERYD_PID_FILE="/var/run/celery/%n.pid"

# Workers should run as an unprivileged user.
CELERYD_USER="celery"
CELERYD_GROUP="celery"

# Extra arguments to celerybeat
CELERYBEAT_OPTS="--schedule=/var/run/celerybeat-schedule"

任何帮助是极大的赞赏。

4

1 回答 1

2

如果您没有scheduler模块并从模块根目录运行 celery：

runner.py：

from celery_test import celery

if __name__ == '__main__':
    celery.start()

重命名celery.py为celery_test.py：

from celery import Celery

celery = Celery('scheduler.celery',
                include=['tasks'])

celery.config_from_object('celeryconfig')

celeryconfig.py：

from celery.schedules import crontab

CELERYBEAT_SCHEDULE = {
    'test-cron': {
        'task': 'tasks.test',
         'schedule': crontab(minute='*/1'),
    },
}

BROKER_URL = 'redis://localhost:6379/0'

tasks.py，固定导入：

from celery_test import celery

@celery.task
def test():
    do_something()

有你必须小心导入和添加runner.py，因为它导入celery_test然后tasks在其中celery_test再次导入。

如果您有scheduler模块并从模块根目录运行 celery：

__init__.py空的。

runner.py：

from scheduler.celery_test import celery

if __name__ == '__main__':
    celery.start()

重命名celery.py为celery_test.py：

from celery import Celery

celery = Celery('scheduler.celery',
                include=['scheduler.tasks'])

celery.config_from_object('celeryconfig')

celeryconfig.py，固定默认任务名称：

from celery.schedules import crontab

CELERYBEAT_SCHEDULE = {
    'test-cron': {
        'task': 'scheduler.tasks.test',
         'schedule': crontab(minute='*/1'),
    },
}

BROKER_URL = 'redis://localhost:6379/0'

tasks.py：

from scheduler.celery_test import celery

@celery.task
def test():
    do_something()
于 2013-06-17T17:01:41.310 回答