python - 芹菜弦,没有执行回调函数
问题描述
from flask import Flask
from celery import Celery
from datetime import timedelta
from celery import group, chord, chain
def make_celery(app):
celery = Celery(
app.import_name,
backend=app.config['CELERY_RESULT_BACKEND'],
broker=app.config['CELERY_BROKER_URL']
)
celery.conf.update(app.config)
class ContextTask(celery.Task):
def __call__(self, *args, **kwargs):
with app.app_context():
return self.run(*args, **kwargs)
celery.Task = ContextTask
return celery
app = Flask(__name__)
app.config.update(
CELERY_BROKER_URL='redis://localhost:6379',
CELERY_RESULT_BACKEND='redis://localhost:6379',
)
celery = make_celery(app)
@celery.task()
def add_together(a, b):
print("starting task adding " + str(a) + " & " + str(b))
return a + b
@celery.task()
def finish_callback(results):
print('Group task finished')
return 0
@app.route('/group_task')
def task_group():
short_tasks = [add_together.delay(10, 1),
add_together.delay(10, 3),
add_together.delay(10, 4),
add_together.delay(10, 5),
add_together.delay(10, 6)
]
callback = finish_callback.s()
chord(short_tasks, callback)
return 'Executing Parallel Task'
我有一个 Flask 应用程序,其中有一些具有子任务的后台操作。我正在尝试用芹菜解决这个问题。在上面的代码片段中,chord 没有调用回调方法。所有的任务都成功执行了,但是没有执行回调方法。谁能告诉我上面的代码有什么问题?
解决方案
推荐阅读
- python - gcloud crontab 每次调用两次调用 appengine
- css - 在悬停时将“display:none”更改为“display:block”不起作用
- regex - 如何使用 mksquashfs -regex?
- jupyter-lab - Jupyterlab 将单元格拆分为 2 列
- node.js - Fresh Laravel Homestead 8.0.1:npm run dev 编译失败,npm ci 安装失败。跨环境的东西?
- c++ - 日期字符串到纪元秒 (UTC)
- c++ - 安装 GLFW 时遇到问题:无法解析的外部符号
- database - 从不同的数据库调用时如何获取包的日志?
- sql - 使用 WITH INSERT INTO
- variables - Ansible 中嵌套变量的语法错误