0

Имеется схема RPC в которой приложение на Java (будем называть его TaskManager'ом) отправляет задания в очередь RabbitMQ (очередь A) к приложениям на PHP (назовем их Worker'ами). После выполнения задания воркер отправляет сообщение с результатом работы обратно к таскменеджеру (очередь B).

Callback Worker'а примерно следующий:

$this->callback = function ($req) {
            $n = $req->body;
            $this->correlation_id = $req->get('correlation_id');

            $msg = new AMQPMessage(
                (string)$this->task($n),
                array('delivery_mode' => 2, 'correlation_id' => $this->correlation_id)
            );

            $req->delivery_info['channel']->basic_publish($msg, '', $req->get('reply_to'));
            $req->delivery_info['channel']->basic_ack($req->delivery_info['delivery_tag']);
        };

У меня возникает следующая проблема:

Запускаем таскменеджер, запускаем около тысячи воркеров. Затем перезагружаем сервер RabbitMQ и видим что некоторые ответы на посланные задачи таскменеджером приходят по два раза.

Есть подозрение что это происходит из за того, что в момент перезагрузки RabbitMQ воркер успевает выполнить функцию basic_publish() в очередь B, а basic_ack() происходит уже при оборванном соединении.

Так как очереди durable, то при поднятии RabbitMQ видит ответ от воркера в очереди к таскменеджеру (в очереди B) и так же видит что воркер не ответил на одно сообщение в очереди A: восстанавливает его и опять отдает воркеру. В итоге таскменеджер получает два одинаковых ответа на одно свое задание.

Пока что это единственное объяснение.

Как решить данную проблему? Есть ли еще варианты её появления?


TEMP:

$this->callback = function ($req) {
            $n = $req->body;
            $this->correlation_id = $req->get('correlation_id');

            $msg = new AMQPMessage(
                (string)$this->task($n),
                array('delivery_mode' => 2, 'correlation_id' => $this->correlation_id)
            );
            $req->delivery_info['channel']->tx_select();    
            $req->delivery_info['channel']->basic_publish($msg, '', $req->get('reply_to'));
            $req->delivery_info['channel']->basic_ack($req->delivery_info['delivery_tag']);
            $req->delivery_info['channel']->tx_commit();
        };

Несколько вопросов:

Если в момент падения RabbitMQ вызвать любую из функций(tx_select, publish, ack или commit) то выпадет исключение о разрыве коннекции?

Как определить что транзакция прошла успешно? (Не будет исключений? tx_commit() возвращает NULL)

В каких случаях вызывать tx_rollback()? В момент получении ошибки в бизнес логике, то есть ошибки не связанной с подключением к rabbit?

1 ответ 1

1

Всё верно, проблема из-за того что брокер взял сообщение с очереди, выполнил publish, но не успел выполнить ack.

Можно поменять местами publish и ack, тогда часть сообщений в этой ситуации будет теряться.

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

Если есть возможность обеспечить чистое завершение процессов, то можно обрабатывать сигнал завершения процесса отключая скрипт после завершения обработки callback.

3
  • На сколько я понимаю, транзакции сработают только если нужно выполнить допустим два publish'а, но в случае publish и ack они не сработают, так как ack не бросает исключений и не возвращает никакого результата (потому что функция асинхронная).
    – alexsm
    30 авг 2017 в 9:47
  • Если верить документации: rabbitmq.com/semantics.html , оба метода, и ack, и publish, поддерживают транзакционность.
    – Tatikoma
    30 авг 2017 в 9:50
  • Дополнил пост блоком TEMP, посмотрите пожалуйста.
    – alexsm
    30 авг 2017 в 11:43

Ваш ответ

Нажимая на кнопку «Отправить ответ», вы соглашаетесь с нашими условиями использования и подтверждаете, что прочитали и поняли наши политику конфиденциальности и нормы поведения.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.