сельдерей вложенные аккорды не вызывают обратный вызов в правильном порядке - PullRequest
Новая
       6

сельдерей вложенные аккорды не вызывают обратный вызов в правильном порядке

0 голосов
/ 25 июня 2018

Я пытаюсь вызвать метод, который создает аккорд, и использовать этот аккорд в другом аккорде, который я строю в другом методе.Но я не могу заставить его вызвать обратный вызов в правильном порядке. 

#celery obj
app = Celery('tasks', backend=BACKEND, broker=BROKER)

Вот простые методы, которые я использую для проверки этого: 

@app.task
def first_task(args):
    return 'first_task'


@app.task
def first_body(args):
    return 'first_body'


@app.task
def second_body(args):
    return 'second_body'


@app.task
def third_body(args):
    return 'third_body'

Вот как ясоздайте мои аккорды: 

@app.task
def simple_chord(args):
    c = chord(group([first_task.s(args)]), body=first_body.s())
    return c()

# this works as expected 
# first_task -> first_body -> second_body -> third_body
def test_chords():
    c = chord(group([chord(group([chord(group([first_task.s('foo')]),
                                        body=first_body.s())]),
                           body=second_body.s())]),
              body=third_body.s())

    c.delay()

# this does not work as expected
# first_task -> second_body -> first_body -> third_body
def test_chords_2():
    c = chord(group([chord(group([simple_chord.s('foo')]),
                           body=second_body.s())]),
              body=third_body.s())

    c.delay()

Когда я строю аккорды в одном месте, как в test_chords, он делает то, что от меня ожидают (first_task -> first_body -> second_body -> third_body).Однако я хочу получить аккорд, который я строю в методе simple_chord, и использовать его в другом аккорде, как в test_chords_2, но он не ожидает исполнения first_body перед выполнением second_body.Вместо этого он выполняется в следующем порядке: first_task -> second_body -> first_body -> third_body

Что я вижу в рабочей консоли: 

[2018-06-25 16:55:21,466: INFO/MainProcess] Received task: tasks.simple_chord[c78eb259-e1e1-4b4d-b053-247afce4d536]  
[2018-06-25 16:55:21,660: INFO/MainProcess] Received task: tasks.first_task[fca138d2-c672-4299-a2ed-1bb5cacb66da]  
[2018-06-25 16:55:21,685: INFO/ForkPoolWorker-1] Task tasks.simple_chord[c78eb259-e1e1-4b4d-b053-247afce4d536] succeeded in 0.19740361299773213s: <AsyncResult: c4db6bbf-d150-44db-a34e-99ea6bd71012>
[2018-06-25 16:55:21,688: INFO/MainProcess] Received task: tasks.second_body[afc32047-3239-4118-bb2c-c116f9a241bf]  
[2018-06-25 16:55:23,340: INFO/ForkPoolWorker-1] Task tasks.first_task[fca138d2-c672-4299-a2ed-1bb5cacb66da] succeeded in 1.639255696994951s: 'first_task'
[2018-06-25 16:55:23,344: INFO/MainProcess] Received task: tasks.first_body[c4db6bbf-d150-44db-a34e-99ea6bd71012]  
[2018-06-25 16:55:24,099: INFO/ForkPoolWorker-1] Task tasks.second_body[afc32047-3239-4118-bb2c-c116f9a241bf] succeeded in 0.7374479610007256s: 'second_body'
[2018-06-25 16:55:24,103: INFO/MainProcess] Received task: tasks.third_body[9a90c2da-cd03-4f55-9675-9194acd7d42f]  
[2018-06-25 16:55:24,656: INFO/ForkPoolWorker-1] Task tasks.first_body[c4db6bbf-d150-44db-a34e-99ea6bd71012] succeeded in 0.5298690330091631s: 'first_body'
[2018-06-25 16:55:25,247: INFO/ForkPoolWorker-1] Task tasks.third_body[9a90c2da-cd03-4f55-9675-9194acd7d42f] succeeded in 0.5770523219980532s: 'third_body'

Мой вопрос: как лучше всегосделать что-то подобное?

1 Ответ

0 голосов
/ 26 июня 2018

Решил проблему, вызвав simple_chord обычно (не как подпись) в test_chords_2.Я заставил его работать асинхронно без необходимости: 

def test_chords_2():
c = chord(group([chord(group([simple_chord('foo')]),
                       body=second_body.s())]),
          body=third_body.s())

print(c)
c.delay()
...