我可以对芹菜任务对象进行本地回调吗

Question

我安装了一个小演示，它旨在调用Celery上一个耗时的函数。我想执行

import itertools
import time
import logging
from celery.result import AsyncResult
from myproj.tasks.time_consuming_thing import time_consuming_thing
log: logging.Logger = logging.getLogger()

def log_the_result(result):
    print("Result: %r" % result)
def main():
    for i in itertools.count(0):
        log.info("About to schedule a task: #%i", i)
        result: AsyncResult = time_consuming_thing.delay()
        result.then(callback=log_the_result)
        time.sleep(10)

if __name__ == "__main__":
    logging.basicConfig()
    logging.getLogger("").setLevel(logging.INFO)
    main()

实际上似乎什么都没有发生：

我可以看到工人正在返回一个价值，但这个价值似乎永远不会传递给消费者。永远不会调用回调函数。

我该怎么做才能用结果的返回值调用回调函数？

Answer 1

为了使用then功能，您必须使用aio、线程或gevent。

对于gevent，你可以使用这样的东西(从上面的github线程复制粘贴(：

import gevent.monkey
gevent.monkey.patch_all()
import itertools
import time
import logging
from celery.result import AsyncResult
from myproj.tasks.time_consuming_thing import time_consuming_thing
log: logging.Logger = logging.getLogger()

def log_the_result(result):
    print("Result: %r" % result)
def main():
    for i in itertools.count(0):
        log.info("About to schedule a task: #%i", i)
        result: AsyncResult = time_consuming_thing.delay()
        result.then(callback=log_the_result)
        time.sleep(10)

if __name__ == "__main__":
    logging.basicConfig()
    logging.getLogger("").setLevel(logging.INFO)
    main()