Запуск параллельных воркеров в Celery

Question

Необходим вызов 2-4 воркеров из существующего (главного) воркера с получением результатов. Схема работы: Основной воркер подготавливает задачи (долго) > 2-4 воркера (еще более долго, рандомное кол-во времени) > суммирование результатов вычисления в основном воркере с передачей результатов в бд. Программа откуда вызывается основной воркер (main.py) не может блокироваться ни на секунду. Все вычисления необходимы в воркерах celery.

Пример моего тестового кода:

# tasks.py

@app.task
def a(m):
    n = 0
    while n < m:
        n += 1
        time.sleep(1)
    return n * 2

@app.task
def b(m):
    n = 0
    while n < m:
        n += 1
        time.sleep(3)
    return n * 3

@app.task
def start_group():
    g = group([a.s(5), b.s(5)])
    r = g.apply_async()
    result = r.get()
    return print(result)

# main.py

start_group.delay()

Получаю такую ошибку:

[2021-04-24 23:24:19,732: ERROR/ForkPoolWorker-2] Task celery_test.start_group[1f483195-8910-4ece-8a37-8d1a2608775e] raised unexpected: RuntimeError('Never call result.get() within a task!\nSee http://docs.celeryq.org/en/latest/userguide/tasks.html#task-synchronous-subtasks\n')
Traceback (most recent call last):
  File "/Users/alexg/PycharmProjects/test/venv/lib/python3.9/site-packages/celery/app/trace.py", line 405, in trace_task
    R = retval = fun(*args, **kwargs)
  File "/Users/alexg/PycharmProjects/test/venv/lib/python3.9/site-packages/celery/app/trace.py", line 697, in __protected_call__
    return self.run(*args, **kwargs)
  File "/Users/alexg/PycharmProjects/test/celery_test.py", line 32, in start_group
    result = r.get()
  File "/Users/alexg/PycharmProjects/test/venv/lib/python3.9/site-packages/celery/result.py", line 676, in get
    return (self.join_native if self.supports_native_join else self.join)(
  File "/Users/alexg/PycharmProjects/test/venv/lib/python3.9/site-packages/celery/result.py", line 793, in join_native
    assert_will_not_block()
  File "/Users/alexg/PycharmProjects/test/venv/lib/python3.9/site-packages/celery/result.py", line 37, in assert_will_not_block
    raise RuntimeError(E_WOULDBLOCK)
RuntimeError: Never call result.get() within a task!
See http://docs.celeryq.org/en/latest/userguide/tasks.html#task-synchronous-subtasks

[2021-04-24 23:24:24,903: INFO/ForkPoolWorker-3] Task celery_test.a[3118cea7-3c0c-4984-97ab-8998ca0d587e] succeeded in 5.753455244000001s: 10
[2021-04-24 23:24:34,828: INFO/ForkPoolWorker-1] Task celery_test.b[acb326a7-5efe-497b-9003-e52e9bfc5028] succeeded in 15.400090888s: 15

Понял, что метод .get() явно не подходит. Нужна помощь, уже очень долго читаю документацию на английском. Принцип работы subtasks, callback, chord все еще до конца не понятен. На русском ни одной подробной документации на данную тему.

Answer 1

Вам нужно использовать chord для этой задачи:

@app.task
def combine(*results):
    print(results)
    return sum(*results)

# main.py
res = chord([a.s(5), b.s(5)])(combine.s())

print(res.get())

chord создает задачу, которая состоит из двух этапов

запуск на выполнение всех задач переданных ей в первом аргументе. Каждая задача запускается отдельно, если у вас много worker-ов, они будут выполнены паралельно
когда все задачи выполнятся, вызывается callback (в этом случае combine) и ему передает результаты всех задач как список.

БЛОГ НА HUSL

Запуск параллельных воркеров в Celery

Ответы (1 шт):