Маршрутизация мертвых букв в приложении Spring Boot

Question

Я прочитал несколько ответов по этой теме, и кажется, что конфигурация, которая у меня есть, должна работать правильно, но по какой-то причине это не так.

Вот конфиг:

@Bean Queue intakeQueue(String name) { return new Queue(name, true); } 

@Bean Exchange dlx(String name) { return new DirectExchange(name); } 

@Bean Queue dlq(String name) { return new Queue(name, false, false, true); } 

@Bean 
Binding dlb(Exchange dlx, Queue dlq, Queue reply) { 
    return BindingBuilder.bind(dlq).to(dlx).with(reply.getName()).noargs(); 
} 

@Bean 
Queue replyQueue(String name, Exchange dlx) { 
    Map<String, Object> args = new HashMap<>(); 
    args.put("x-dead-letter-exchange", dlx.getName()); 
    args.put("x-dead-letter-routing-key", name); 
    return new Queue(name, true, false, false, args); 
} 

RabbitMQ UI показывает, что очередь имеет ответ DLX и DLK атрибуты.

Я отправлять сообщения как

this.rabbit.convertSendAndReceive(intakeQueue, obj, message -> { 
    message.getMessageProperties().setPriority(10); 
    return message; 
}); 

обработчика сообщений бросает AmqpRejectAndDontRequeueException сразу после того, как он получает сообщение. Это сделано специально для тестирования. Я начал с советов по повторной попытке, но поскольку он не дал никаких результатов, я упростил тестовый пример.

public Object handleMessage(Object obj) throws IOException { 
    throw new AmqpRejectAndDontRequeueException("Testing retries!"); 
} 

Есть два вопроса, которые я вижу сейчас:

1. сообщения никогда не показывает в DLQ после ARADRE была брошена. Это происходит, если я публикую в DLQ прямо с handleMessage.
2. convertSendAndReceive не получает обратно ничего (исключение может быть?) И ждет, пока не произойдет тайм-аут, который составляет 5 минут в моем случае. Это может быть предназначено, но для вызовов в стиле RPC это довольно странно.

Я пропустил или неправильно сконфигурировал что-то?

Answer 1

Это не RPC в этом смысле; выброс исключения в слушателе не будет передаваться обратно отправителю.

Вы отклоняете поставку на intakeQueue, и сообщение будет перенаправлено на его DLX/DLQ (если сконфигурировано).

Для вашего тестового примера вы должны посмотреть , что DLQ (если таковой имеется), а не DLQ очереди ответа.

RabbitMQ не знает об отношениях в очереди. Вы не показываете контейнер или конфигурацию RabbitTemplate, но по умолчанию используется прямой ответ на маршрутизацию (которая использует «специальную» внутреннюю очередь).

Вы можете настроить шаблон кролика для использования фиксированной очереди ответа, но тогда вы должны предоставить контейнер-получатель ответа as described in the documentation.

Если отправитель имеет время ожидания при получении ответа, шаблон выдает ARADRE, поэтому в этом случае ответ будет мертвым.

Чтобы проверить это, спать в вашем слушателе дольше, чем время ожидания, затем ответить; вы должны увидеть ответ в DLQ очереди ответа.

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

В настоящее время нет никакой логики для повторного выброса такого исключения, однако вам придется делать это в своем коде (обнаружите, что ответ является исключением и повторно выбрасывает его).

Конечно, тип исключения должен быть Serializable.

Вы также можете использовать Spring Remoting over AMQP, который будет заботиться о распространении исключений.