带有一组链条的芹菜弦，链条中有例外

Question

我正在使用 celery 用一组链条来运行和弦。当组中的所有任务（链...）成功完成时，就会触发和弦回调，并且事情会按照我的预期工作。但是，当组中的任务失败时，在这种情况下，我不希望调用 chord 回调，chord_unlock 会无限循环。如果组中的一个链发生故障，如何避免 chord_unlock 循环？

这是我的代码：

@app.task
def test1():
    logging.info("test1")
    raise Exception()

@app.task
def test2():
    logging.info("test2")

@app.task
def test3():
    logging.info("test3")

@app.task
def test4():
    logging.info("test4")

@app.task
def cb(id):
    logging.info("cb")

def test():
    chains = [chain(test1.si(), test2.si()), chain(test3.si(), test4.si())]
    chord(group(chains))(cb.si()) 

还有日志：

[2018-09-09 15:15:12,933: INFO/MainProcess] Received task: projecttasks.tasks.test[e332ee64-84b3-4f3f-bb84-de83fe03b758]
[2018-09-09 15:15:12,973: INFO/MainProcess] Received task: projecttasks.tasks.test1[5a8191fd-a9c6-430f-bb46-61a111766776]
[2018-09-09 15:15:12,977: INFO/ForkPoolWorker-1] test1
[2018-09-09 15:15:12,986: INFO/MainProcess] Received task: projecttasks.tasks.test3[22c2ac6d-1bca-41f8-b617-2d7ee5b20d4d]
[2018-09-09 15:15:13,006: INFO/MainProcess] Received task: celery.chord_unlock[872672ca-a627-4fe9-b81b-07b4bbc2c2c2]  ETA:[2018-09-09 15:15:13.978298+00:00]
[2018-09-09 15:15:13,015: ERROR/ForkPoolWorker-1] Task projecttasks.tasks.test1[5a8191fd-a9c6-430f-bb46-61a111766776] raised unexpected: Exception()
Traceback (most recent call last):
   File "/usr/local/lib/python2.7/dist-packages/celery/app/trace.py", line 382, in trace_task
     R = retval = fun(*args, **kwargs)
   File "/usr/local/lib/python2.7/dist-packages/celery/app/trace.py", line 641, in __protected_call__
     return self.run(*args, **kwargs)
   File "/project/projecttasks/tasks.py", line 185, in test1
     raise Exception()
Exception
[2018-09-09 15:15:13,017: INFO/ForkPoolWorker-2] Task projecttasks.tasks.test[e332ee64-84b3-4f3f-bb84-de83fe03b758] succeeded in 0.0738198000472s: None
[2018-09-09 15:15:13,021: INFO/ForkPoolWorker-2] test3
[2018-09-09 15:15:13,029: INFO/MainProcess] Received task: projecttasks.tasks.test4[6f791b45-0c9f-4b0c-984d-387429b39fad]
[2018-09-09 15:15:13,034: INFO/ForkPoolWorker-1] test4
[2018-09-09 15:15:13,042: INFO/ForkPoolWorker-2] Task projecttasks.tasks.test3[22c2ac6d-1bca-41f8-b617-2d7ee5b20d4d] succeeded in 0.0202670998406s: None
[2018-09-09 15:15:13,045: INFO/ForkPoolWorker-1] Task projecttasks.tasks.test4[6f791b45-0c9f-4b0c-984d-387429b39fad] succeeded in 0.0111124999821s: None
[2018-09-09 15:15:14,789: INFO/MainProcess] Received task: celery.chord_unlock[872672ca-a627-4fe9-b81b-07b4bbc2c2c2]  ETA:[2018-09-09 15:15:15.785804+00:00]
2018-09-09 15:15:14,790: INFO/ForkPoolWorker-2] Task celery.chord_unlock[872672ca-a627-4fe9-b81b-07b4bbc2c2c2] retry: Retry in 1s
[2018-09-09 15:15:15,953: INFO/ForkPoolWorker-2] Task celery.chord_unlock[872672ca-a627-4fe9-b81b-07b4bbc2c2c2] retry: Retry in 1s
[2018-09-09 15:15:15,958: INFO/MainProcess] Received task: celery.chord_unlock[872672ca-a627-4fe9-b81b-07b4bbc2c2c2]  ETA:[2018-09-09 15:15:16.952066+00:00]
[2018-09-09 15:15:18,792: INFO/ForkPoolWorker-2] Task celery.chord_unlock[872672ca-a627-4fe9-b81b-07b4bbc2c2c2] retry: Retry in 1s
2018-09-09 15:15:18,795: INFO/MainProcess] Received task: celery.chord_unlock[872672ca-a627-4fe9-b81b-07b4bbc2c2c2]  ETA:[2018-09-09 15:15:19.790674+00:00]
[2018-09-09 15:15:20,726: INFO/ForkPoolWorker-2] Task celery.chord_unlock[872672ca-a627-4fe9-b81b-07b4bbc2c2c2] retry: Retry in 1s
[2018-09-09 15:15:20,728: INFO/MainProcess] Received task: celery.chord_unlock[872672ca-a627-4fe9-b81b-07b4bbc2c2c2]  ETA:[2018-09-09 15:15:21.724806+00:00]
[2018-09-09 15:15:22,797: INFO/ForkPoolWorker-2] Task celery.chord_unlock[872672ca-a627-4fe9-b81b-07b4bbc2c2c2] retry: Retry in 1s
[2018-09-09 15:15:22,799: INFO/MainProcess] Received task: celery.chord_unlock[872672ca-a627-4fe9-b81b-07b4bbc2c2c2]  ETA:[2018-09-09 15:15:23.795726+00:00]
[2018-09-09 15:15:24,801: INFO/ForkPoolWorker-2] Task celery.chord_unlock[872672ca-a627-4fe9-b81b-07b4bbc2c2c2] retry: Retry in 1s
[2018-09-09 15:15:24,802: INFO/MainProcess] Received task: celery.chord_unlock[872672ca-a627-4fe9-b81b-07b4bbc2c2c2]  ETA:[2018-09-09 15:15:25.799817+00:00]
2018-09-09 15:15:25,952: INFO/ForkPoolWorker-2] Task celery.chord_unlock[872672ca-a627-4fe9-b81b-07b4bbc2c2c2] retry: Retry in 1s
[2018-09-09 15:15:25,953: INFO/MainProcess] Received task: celery.chord_unlock[872672ca-a627-4fe9-b81b-07b4bbc2c2c2]  ETA:[2018-09-09 15:15:26.951535+00:00]
[2018-09-09 15:15:28,809: INFO/ForkPoolWorker-2] Task celery.chord_unlock[872672ca-a627-4fe9-b81b-07b4bbc2c2c2] retry: Retry in 1s
[2018-09-09 15:15:28,814: INFO/MainProcess] Received task: celery.chord_unlock[872672ca-a627-4fe9-b81b-07b4bbc2c2c2]  ETA:[2018-09-09 15:15:29.807464+00:00]
[2018-09-09 15:15:30,723: INFO/ForkPoolWorker-2] Task celery.chord_unlock[872672ca-a627-4fe9-b81b-07b4bbc2c2c2] retry: Retry in 1s

Answer 1

我遇到了类似的问题。我调用了一组链，在所有任务完成后（不管结果如何），我想调用回调函数。使用带有错误处理 (on_error) 的和弦和回调函数似乎是完美的解决方案。不幸的是，一切都适用于任务列表，但不适用于链列表。丑陋的钩子是一个包装任务。

@celery_app.task
def task_one():
    return 'OKIDOKI'

@celery_app.task
def task_two(str):
    return f'{str} YOUPI'

@celery_app.task
def task_three(str):
    return f'{str} MAKAPAKA'

@celery_app.task
def task_exception(str):
    raise KeyError(f'{str} Ups')

@celery_app.task(ignore_result=True)
def task_wrapper(*args, **kwargs):
    if 'job' in kwargs:
        kwargs['job'].apply()

@celery_app.task(ignore_result=True)
def callback_task(*args, **kwargs):
    return (args, kwargs, 'Yeah')

def test():
    chains = []

    tasks = [
        task_one.s(),
        task_two.s(),
        task_exception.s(),
        task_three.s(),
    ]
    chains.append(task_wrapper.s(job=chain(*tasks)))

    tasks = [
        task_one.s(),
        task_two.s(),
        task_three.s(),
    ]
    chains.append(task_wrapper.s(job=chain(*tasks)))

    chord(chains, callback_task.s()).apply_async()