RabbitMQ: как правильно выполнять тяжелые задачи?

Question

[topuserman]

Всем привет!
У нас на проекте отправляются задания на генерацию отчетов в очередь rabbitmq.
Саму очередь обслуживает php-консюмер (работает через php-amqplib).

Проблема в том, что генерация отчетов, может занимать 10 мин, некоторые даже по часу..

Консюмер работает следующим образом:
берет N-сообщений из очереди, и начинает сразу же генерировать отчеты.

В момент, когда консюмер занят генераций отчета, как я понял, rabbitmq присылает некоторые проверки на жизнеспособность консюмера — heartbeat . По умолчанию, если консюмер не отвечает 60 сек., то при следующей итерации вылетает ошибка, что потеряно соединение.

У меня два вопроса:

1. Чем чревато отключение - heartbeat в rabbitmq ? Почитал тут но все равно не понял.
2. Как принято обрабатывать тяжелые задачи, которые получает косюмер ? может их не стоит сразу в самом консюмере запускать ? если запускать отдельным процессом, то возникает сразу вопрос, что делать, если в процессе генерации отчета произошла ошибка ? сообщение же из очереди прочитано и отправлено ack.

нужны best practices))

Answer 1

[Developer]

Сам консюмер не должен обрабатывать задачи. Его основное назначение - разгрузить очередь и отправить задачи дальше, например в базу данных или на другие сервера через балансировщик. Лучше всего первый способ.
Далее некий воркер смотрит необработанные задачи в базе данных и неспеша их обрабатывает. Обработанные задачи помечает флагом.

Comments

[topuserman]

А для чего тогда нужен rabbit?)
Если таски можно таким же образом писать в бд, и воркер будет читать базу, запускать задачу и помечать.

[Developer]

topuserman, нет, нельзя. БД не будет справляться с огромным потоком данных. Поэтому очереди и придумали.
Быстро пишем в кролика, читаем быстро консюмером и потом распределяем по шардированной БД, либо по распределённым сервисам через тот же nginx, например.
Тут нужна практика и понимать детали задачи

Answer 2

[Сергей]

Была подобная проблема, сишный клиент в один поток не умел нормально в heartbeat. При отключении heartbeat'ов на уровне AMQP протокола были проблемы с "зависшими клиентами" на стороне RabbitMQ, когда приложение выходило непредвиденно. Настройка TCP keepalive помогла решить данную проблему.
Не знаю точно как работает современный php, но, если есть возможность держать отдельный поток для поддержания AMQP коннекта с включённым heartbeat'ом, то это будет лучше, чем через TCP keepalive. Отсылать ack обратно лучше всё же после завершения обработки задачи.

Answer 3

[Дмитрий]

Это выглядит как проблема библиотеки php-amqplib, php до сих пор не умеет в многопоточность?
Можно создавать процесс, который будет генерировать отчет.

Answer 4

[part_os]

А как вы поддерживаете соединение с БД, если в течении генерации отчёта нету запросов к ней? Мы делаем ping к БД, так же и к ребитту тем самым сообщая что косьюмер живой м работает.

Comments

[topuserman]

Никак)) бывает, что ловим ошибку mysql - что потеряно соединение.

А как вы пингуете реббит и мускул когда сам консюмер занят?

[part_os]

где идёт обработка мы вызываем callback функции в которой находится

$request->setHeartbeatCallback(
                function () {
                    $this->heartbeat();
                }
            );

protected function heartbeat()
    {
        if ($this->channel !== null) {
            $this->logger->info('RABBIT: heartbeat');
            $this->channel->wait(null, true);
        }
    }

https://github.com/php-amqplib/php-amqplib/blob/c0...

у доктрины есть публичный метод ping() найти с ходу не могу, поищите, можно так же там же вызывать.

возможно есть более красивые решения, наши проблемы это решает. консьюмеры не умирают по закрытым каналам.

Answer 5

[Антон Минаков]

Частичное решение https://github.com/php-enqueue/enqueue-dev/blob/ma...

Answer 6

[Елена Степанова]

php-amqplib тормозной, лучше расширение amqp из pecl