芹菜：无法通过烧瓶调用任务，在烧瓶外工作

Question

@更新：将任务从 shared_task 更改为 app.celeryd.celery.task 即可解决问题。 shared_task 是否有一些额外的设置才能正常工作？

完全重写还是一样的错误。我会尽量保持相对简短。新的项目结构更简洁，如下所示：

proj
|-- app
|   |-- controller
|   |   |-- __init__.py
|   |   +-- greeting_model.py
|   |-- model
|   |   |-- __init__.py
|   |   +-- dto
|   |       |-- __init__.py
|   |       +-- greeting_dto.py
|   |-- service
|   |   |-- __init__.py
|   |   +-- greeting_service.py
|   |-- tasks
|   |   |-- __init__.py
|   |   +-- greeting_tasks.py
|   |-- __init__.py
|   |-- celeryd.py
|   +-- flaskd.py
|-- test.py
|-- worker.py
+-- ws.py

我正在分别初始化 celery 和 flask 并提供 worker.py，它在客户端机器上应该是运行，而 ws.py（flask 网络服务）将运行另外一个。 Celery 初始化非常简单，并使用带有 RabbitMQ 代理的 rpc 后端。现在的 2 个队列是静态的，但稍后将从配置中填充这些队列。

from kombu import Queue
from celery import Celery


celery = Celery('LdapProvider',
                broker='amqp://admin:passwd@localhost:5672/dev1',
                backend='rpc',
                include=['app.tasks.greeting_tasks'])
celery.conf.task_queues = (
    Queue("q1", routing_key="c1.q1"),
    Queue("q2", routing_key="c2.q2"),
)

Worker.py（用于启动 celery worker - 对于这个问题过于简化）：

from app.celeryd import celery as celery
from celery.bin.worker import worker


if __name__ == '__main__':
    celeryd = worker(app=celery)

    options = {
        'broker': 'amqp://admin:passwd@localhost:5672/dev1',
        'queues': 'q1',
        'loglevel': 'info',
        'traceback': True
    }

    celeryd.run(**options)

我将省略 flask 初始化并跳转到调用 celery 任务的 greeting_service.py:

# greeting_service.py:
from app.celeryd import celery
from app.tasks.greeting_tasks import say_hello


class GreetingService(object):
    def say_hello(self, name: str) -> str:
        async_result = say_hello.apply_async((name,), queue='q1')
        return async_result.get()



# greeting_tasks.py
from celery import shared_task


@shared_task(bind=True)
def say_hello(self, name: str) -> str:
    return name.capitalize()

无论我怎么尝试，这个调用都通过 flask 失败了。我创建 test.py 只是为了测试芹菜是否有效：

from app.celeryd import celery
from app.tasks.greeting_tasks import say_hello


if __name__ == '__main__':
    async_result = say_hello.apply_async(('jackie',), queue='q1')
    print(async_result.get())

与 greeting_service.py 几乎相同，只是不是从 greeting_controller 调用，后者是一个 flask_restplus 命名空间。 test.py 导致的差异：

/home/pupsz/PycharmProjects/provider/venv37/bin/python /home/pupsz/PycharmProjects/provider/test.py
Jackie

Process finished with exit code 0

[2020-01-16 18:56:17,065: INFO/MainProcess] Received task: app.tasks.greeting_tasks.say_hello[bb45e271-563e-405b-8529-7335a3810976]  
[2020-01-16 18:56:17,076: INFO/ForkPoolWorker-2] Task app.tasks.greeting_tasks.say_hello[bb45e271-563e-405b-8529-7335a3810976] succeeded in 0.010257695998006966s: 'Jackie'

虽然通过 flask 我得到的只是已经显示的并且工作日志没有显示任何传入任务意味着通过 flask apply_async 没有将任务发送到 RabbitMQ:

File "/home/xyz/PycharmProjects/proj/app/service/greeting_service.py", line 8, in say_hello
return async_result.get()
NotImplementedError: No result backend is configured.
Please see the documentation for more information.

我在 django 中发现了一个类似的问题但没有答案，所以我有点卡住了，希望能得到一些指导。

Answer 1

解决方案：使 shared_task 按预期工作的解决方案已在此处得到解答：LINK 将 celerz 初始化修改为：

from kombu import Queue
from celery import Celery


celery = Celery('LdapProvider',
                broker='amqp://admin:passwd@localhost:5672/dev1',
                backend='rpc')
                # include=['app.tasks.greeting_tasks'])
celery.conf.task_queues = (
    Queue("q1", routing_key="c1.q1"),
    Queue("q2", routing_key="c2.q2"),
)
celery.set_default()

即使我要删除注释掉的包含行，工作人员也会成功获取 app.tasks.greeting_tasks 中定义的 shared_task:

[tasks]
  . app.tasks.greeting_tasks.say_hello

应用程序设置为 default_app() 后，即使使用 shared_task 也不会再抛出 NotImplementedError。至于原因……我不知道，这是 6 个小时的不同配置和谷歌搜索的反复试验。我发现官方文档在某些更复杂的情况下乏善可陈。

芹菜：无法通过烧瓶调用任务，在烧瓶外工作

Celery: unable to call task through flask, works outside of flask

celery

python-3.7