Почему мои каналы RabbitMQ продолжают закрываться?

Я отлаживаю код Java, который использует Apache POI для извлечения данных из документов Microsoft Office. Иногда он сталкивается с большим документом, и POI падает, когда заканчивается память. В этот момент он пытается опубликовать ошибку в RabbitMQ, так что другие компоненты могут знать, что этот шаг не прошел и предпримет соответствующие действия. Однако, когда он пытается опубликовать в очередь, он получает com.rabbitmq.client.AlreadyClosedException (clean connection shutdown; reason: Attempt to use closed channel).

Здесь код обработчика ошибок:

try {
    //Extraction and indexing code
}
catch(Throwable t) {
    // Something went wrong! We'll publish the error and then move on with
    // our lives
    System.out.println("Error received when indexing message: ");
    t.printStackTrace();
    System.out.println();
    String error = PrintExc.format(t);
    message.put("error", error);

    if(mime == null) {
        mime = "application/vnd.unknown";
    }

    message.put("mime", mime);
    publish("IndexFailure", "", MessageProperties.PERSISTENT_BASIC, message);
}

Для полноты, здесь метод публикации:

private void publish(String exch, String route, 
    AMQP.BasicProperties props, Map<String, Object> message) throws Exception{
    chan.basicPublish(exch, route, props, 
        JSONValue.toJSONString(message).getBytes());  
}

Я не могу найти код в блоке try, который закрывает канал RabbitMQ. Существуют ли какие-либо обстоятельства, при которых канал может быть закрыт неявно?

EDIT: я должен отметить, что AlreadyClosedException вызывается вызовом basicPublish внутри публикации.

Ответы

Ответ 1

Канал AMQP закрывается по ошибке канала. Две распространенные вещи, которые могут вызвать ошибку канала:

  • Попытка опубликовать сообщение для обмена, который не существует
  • Попытка опубликовать сообщение с немедленным набором флагов, который не имеет очереди с активным набором пользователей

Я бы посмотрел на настройку ShutdownListener на канале, который вы пытаетесь опубликовать, используя addShutdownListener(), чтобы поймать событие выключения и посмотреть при чем это вызвало.

Ответ 2

Другая причина в моем случае заключалась в том, что по ошибке я дважды признал сообщение. Это приводит к ошибкам RabbitMQ в журнале, подобным этому после второго подтверждения.

=ERROR REPORT==== 11-Dec-2012::09:48:29 ===
connection <0.6792.0>, channel 1 - error:
{amqp_error,precondition_failed,"unknown delivery tag 1",'basic.ack'}

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

Ответ 3

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

Еще одна возможная причина для получения закрытого канала - это когда издатели и потребители обращаются к каналу/очереди с разными объявлениями/настройками очереди

Издатель

channel.queueDeclare("task_queue", durable, false, false, null);

Рабочий

channel.queueDeclare("task_queue", false, false, false, null);

С сайта RabbitMQ

RabbitMQ doesn't allow you to redefine an existing queue with different parameters and will return an error to any program that tries to do that

Ответ 4

У меня также была эта проблема. Причиной моего случая было то, что сначала я построил очередь с durable = false, и в файле журнала у меня появилось это сообщение об ошибке, когда я переключил durable на true:

"неэквивалентный arg 'durable' для очереди 'logsQueue' в vhost '/': получил" true ", но текущий -" false".

Затем я изменил имя очереди, и это сработало для меня. Я предположил, что сервер RabbitMQ хранит запись построенных очередей где-то, и он не может изменить статус от долговременного до недолговечного и наоборот.

Снова я сделал durable = false для новой очереди, и на этот раз я получил эту ошибку

"неэквивалентный arg 'durable' для очереди 'logsQueue1' в vhost '/': получил" false ", но текущий -" true".

Мое предположение было правдой. Когда я перечислил очереди на сервере rabbitMQ, выполните следующие действия:

rabbitmqctl list_queues

Я видел обе очереди на сервере.

Подводя итог, 2 решения: 1. переименование имени очереди, которое не является хорошим решением 2. сброс rabbitMQ с помощью:

rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl start_app

Ответ 5

Для тех, кто интересуется, почему их потребительские каналы закрываются, проверьте, не пытаетесь ли вы подтвердить или отменить доставку более одного раза.

В журнале rabbitmq вы увидите такие сообщения:

операция basic.ack вызвала исключение канала precondition_failed: неизвестный тег доставки...