Как отправить сообщение непосредственно в очередь на парковку, предотвратить перезапуск и выйти из потока программы? - PullRequest
Новая
       75

Как отправить сообщение непосредственно в очередь на парковку, предотвратить перезапуск и выйти из потока программы?

0 голосов
/ 14 февраля 2020

В настоящее время у меня есть 4 очереди:

  • test-queue
  • test-queue-short-term-dead-letter
  • test-queue-long- term-dead-letter
  • test-queue-parking-lot

Когда сообщение поступает в test-queue, я проверяю, имеет ли сообщение правильный формат , Если это не Я хочу отправить сообщение непосредственно в очередь на стоянку .

Я не могу использовать AmqpRejectAndDontRequeue(), потому что он автоматически отправит сообщение на настроенный DLQ ( тест-очередь кратковременная мертвая буква).

Использование RabbitTemplate.convertAndSend() с другим исключением, таким как BadRequestException, не работает. Сообщение отправляется в очередь на стоянку, как и ожидалось, однако такое же сообщение останется в test-queue

. Использование RabbitTemplate.convertAndSend() само по себе не будет работать, так как программа продолжает выполнение.

Все очереди связаны с одним прямым обменом, каждый с уникальными ключами маршрутизации. test-queue настроен со следующими аргументами:

  • x-dead-letter-exchange: ""
  • x-dead-letter-routing-key: <shortTermDeadLetterKey>

Получатель: 

  @RabbitListener(queues = "test-queue")
  public void receiveMessage(byte[] person) {
    String personString = new String(person);

    if (!personString.matches(desiredRegex)) {
      rabbitTemplate.convertAndSend("test-exchange", "test-queue-parking-lot",
          "invalid person");
      log.info("Invalid person");
    }
    ...some other code which I dont want to run as the message has arrived in the incorrect format
}

1 Ответ

0 голосов
/ 14 февраля 2020

Проблема была решена путем ручного подтверждения сообщения и возврата из метода. 

  @RabbitListener(queues = "test-queue")
  public void receiveMessage(byte[] person, Channel channel,
  @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception) {
    String personString = new String(person);

    if (!personString.matches(desiredRegex)) {
      rabbitTemplate.convertAndSend("test-exchange", "test-queue-parking-lot",
          "invalid person");
      log.info("Invalid person");
      channel.basicAck(tag, false);
      return;
    }
    ...some other code which I dont want to run as the message has arrived in the incorrect format
}
...