如何调用 Celery shared_task？

Question

我正在尝试在我的应用程序（不是 Django）中使用 stream_framework但我在调用 stream_framework 共享任务时遇到问题。Celery 似乎找到了任务：

-------------- celery@M3800 v3.1.25 (Cipater)
---- **** ----- 
--- * ***  * -- Linux-4.15.0-34-generic-x86_64-with-Ubuntu-18.04-bionic
-- * - **** --- 
- ** ---------- [config]
- ** ---------- .> app:         task:0x7f8d22176dd8
- ** ---------- .> transport:   redis://localhost:6379/0
- ** ---------- .> results:     redis://localhost:6379/0
- *** --- * --- .> concurrency: 8 (prefork)
-- ******* ---- 
--- ***** ----- [queues]
 -------------- .> celery           exchange=celery(direct) key=celery


[tasks]
  . formshare.processes.feeds.tasks.test_shared_task
  . stream_framework.tasks.fanout_operation
  . stream_framework.tasks.fanout_operation_hi_priority
  . stream_framework.tasks.fanout_operation_low_priority
  . stream_framework.tasks.follow_many
  . stream_framework.tasks.unfollow_many

[2018-09-17 10:06:28,240: INFO/MainProcess] Connected to redis://localhost:6379/0
[2018-09-17 10:06:28,246: INFO/MainProcess] mingle: searching for neighbors
[2018-09-17 10:06:29,251: INFO/MainProcess] mingle: all alone

我运行芹菜：

celery -A formshare.processes.feeds.celery_app worker --loglevel=info

我的 celery_app 有：

from celery import Celery

celeryApp = Celery('task', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0', include='formshare.processes.feeds.tasks')

问题是 delay() 不运行共享任务。我还在我的应用程序中创建了一个共享任务，但是当我调用 delay() 时，该任务也没有被调用。我想我需要从我的应用程序中将它们注册为可调用的？我似乎没有在网上找到任何信息。

我也尝试自动发现任务，但我遇到了同样的问题：

celeryApp.autodiscover_tasks(['stream_framework', 'formshare.processes.feeds'],force=True)

任何想法都受到高度赞赏。

Answer 1

共享任务是用于在不同应用程序之间实际共享任务的特定事物（我认为主要是 Django 应用程序，但例如我在烧瓶中使用它们）。

我们遇到了同样的问题，为了让它发挥作用，我们设置了

 celery_app.set_default()

关于 celery 实例化

否则，另一种使事情正确的方法是通过应用程序本身实际调用任务，因此围绕这些行

from celery import current_app
.
.
.
current_app.tasks['my.tasks.to.exec'].delay(something)

这始终有效，因为它是共享任务，因此在导入时不会绑定到任何应用程序，在这种情况下，它属于配置为“current_app”的应用程序