python - 如何从脚本/模块 __main__ 启动 Celery worker？

Question

我已经Celery在一个模块中定义了一个应用程序，现在我想从它的同一个模块启动工作程序__main__，即通过运行模块python -m而不是celery从命令行。我试过这个：

app = Celery('project', include=['project.tasks'])

# do all kind of project-specific configuration
# that should occur whenever this module is imported

if __name__ == '__main__':
    # log stuff about the configuration
    app.start(['worker', '-A', 'project.tasks'])

但现在 Celery 认为我正在运行没有参数的工人：

Usage: worker <command> [options] 

Show help screen and exit.

Options:
  -A APP, --app=APP     app instance to use (e.g. module.attr_name)
[snip]

使用消息是您从中获得的消息celery --help，就好像它没有收到命令一样。我也试过

app.worker_main(['-A', 'project.tasks'])

但这抱怨-A不被认可。

那么我该怎么做呢？或者，我如何将回调传递给工作人员以使其记录有关其配置的信息？

Answer 1

使用 app.worker_main 方法（v3.1.12）：

± cat start_celery.py
#!/usr/bin/python

from myapp import app


if __name__ == "__main__":
    argv = [
        'worker',
        '--loglevel=DEBUG',
    ]
    app.worker_main(argv)

Answer 2

根据Django-Celery 模块中的代码，您可以尝试以下操作：

from __future__ import absolute_import, unicode_literals

from celery import current_app
from celery.bin import worker


if __name__ == '__main__':
    app = current_app._get_current_object()

    worker = worker.worker(app=app)

    options = {
        'broker': 'amqp://guest:guest@localhost:5672//',
        'loglevel': 'INFO',
        'traceback': True,
    }

    worker.run(**options)

Answer 3

自从Celery 5改变了

现在的worker_main结果：

AttributeError: 'Celery' object has no attribute 'worker_main'

对于 Celery 5，请执行以下操作：

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

if __name__ == '__main__':
    worker = app.Worker(
        include=['project.tasks']
    )
    worker.start()

有关详细信息，请参见此处celery.apps.worker和celery.worker.WorkController.setup_defaults（希望将来能更好地记录它）。

Answer 4

worker_main在这里放回 celery 5.0.3： https ://github.com/celery/celery/pull/6481

这在 5.0.4 上对我有用：

self.app.worker_main(argv = ['worker', '--loglevel=info', '--concurrency={}'.format(os.environ['CELERY_CONCURRENCY']), '--without-gossip'])

Answer 5

我认为您只是缺少包装 args 以便 celery 可以读取它们，例如：

queue = Celery('blah', include=['blah'])
queue.start(argv=['celery', 'worker', '-l', 'info'])