RabbitMQ 리스너의 에러 처리
RabbitMQ는 기본적으로 리스너에서 에러가 발생하면 해당 메시지를 다시 큐에 집어넣고 재발송을 하게 된다. 이러한 처리 방식의 문제점은 일시적인 에러가 아니라 당장 해결할 수 없는 에러(ex. DB 접근 차단)가 발생할 경우 무한히 재시도를 하게 된다는 것이다.
재시도 차단
가장 단순한 해결 방법은, 리스너에서 에러가 발생해도 다시 큐에 집어넣지 않는 것이다. 이 부분은 리스너에 Requeue 옵션을 false로 하여 제어할 수 있다.
@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
ConnectionFactory connectionFactory) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
factory.setDefaultRequeueRejected(false);
return factory;
}하지만 이런 방식으로 처리하면 에러가 발생하면 메시지는 유실되고, 재시도를 할 수 없게 된다.
DLX, DLQ를 활용한 에러 처리
DLX는 Dead Letter Exchange, DLQ는 Dead Letter Queue의 줄임말으로 실패한 메시지를 처리하기 위한 Exchange와 Queue를 의미한다. 특별하게 느껴지지만 크게 다르지 않다. 그저 특정 Queue에서 메시지를 처리하지 못하고 에러가 발생할 경우 DLX로 전송하여 처리하게 된다.
DLX로 실패한 메시지 전송
메시지를 처리하는 중 에러가 발생했을 때 해당 메시지를 DLX를 보내기 위해서는 먼저 Queue의 'x-dead-letter-exchange'를 DLX로 지정해야 한다. 그럼 실패한 메시지(Dead Letter)를 해당 Exchange로 전송하게 된다.
@Bean
public Queue hubQueue() {
return QueueBuilder.durable(hubProperties.queue())
// hubQueue의 DLX 기능 활성화
.withArgument("x-dead-letter-exchange", DLX)
.withArgument("x-dead-letter-routing-key", RETRY_ROUTING_KEY)
.build();
}그리고 AmqpRejectAndDontRequeueException을 발생시키거나, 메시지를 수동으로 승인/거부 처리할 수 있도록 한 뒤 메시지를 거절하면 DLX로 전송되게 된다.
@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
ConnectionFactory connectionFactory) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
// 메시지 승인/거절 수동으로 처리하도록 설정
factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
return factory;
}// 메시지를 큐로 재전송하지 않고 거절
channel.basicNack(tag, false, false);DLX로 전송된 실패한 메시지는 DLX와 바인딩된 DLQ로 전송되어 보관된다.
재시도 로직 구현
하지만 실패한 메시지를 바로 DLX로 보내면 재시도하면 해결될 수 있는 일시적인 에러일 경우, 메시지를 바로 처리하지 못하고 비효율적으로 처리하게 된다. 따라서 재시도 횟수를 제한하여 무한히 재시도되지 않도록 하고, 재시도 횟수를 초과하면 DLQ로 전송되어 보관되도록 구현하려고 한다.
RetryQueue
DLX와 바인딩된 큐에 DLQ뿐만 아니라 RetryQueue를 함께 바인딩하고, 'x-dead-letter-exchange'를 작업 Queue가 바인딩된 Exchange로 지정한다. 그리고 실패한 메시지를 전송할 때 사용할 'x-dead-letter-routing-key'를 RetryQueue에 바인딩된 라우팅키로 설정한다. 그리고 ttl을 설정하면 지정된 시간이 지나면 RetryQueue에 있는 메시지를 실패한 것으로 간주하고 'x-dead-letter-exchange'에 지정된 Exchange로 전송하게 된다.
@Bean
public Queue retryQueue() {
return QueueBuilder.durable(RETRY)
.withArgument("x-message-ttl", 3000)
.withArgument("x-dead-letter-exchange", hubProperties.exchange())
.withArgument("x-dead-letter-routing-key", hubProperties.routingKey())
.build();
}그러면 실패한 메시지는 우선 RetryQueue로 전송되고, 일정 시간 대기 후 다시 해당 Queue로 전송되어 재시도를 할 수 있게 된다.
재시도 횟수 제한
이제 재시도 횟수를 제한하려면 해당 메시지가 몇 번째 재시도 중인지 알아야 한다. RabbitMQ는 'x-dead-letter-exchange' 설정을 통해 실패한 메시지를 전송할 경우, 메시지의 헤더에 'xDeath'를 추가하고 해당 헤더에 리스트 형태로 몇번째 재시도 중인지 기록한다.
// 메시지의 x-death 헤더에서 현재까지의 재시도 횟수를 추출
private int getRetryCount(Message message) {
MessageProperties properties = message.getMessageProperties();
Map<String, Object> headers = properties.getHeaders();
Object xDeathHeader = headers.get("x-death");
if (xDeathHeader instanceof List<?> deathList) {
if (!deathList.isEmpty()) {
Object lastDeathEntry = deathList.getLast();
if (lastDeathEntry instanceof Map<?, ?> deathMap) {
Object countObject = deathMap.get("count");
if (countObject instanceof Number) {
return ((Number) countObject).intValue();
}
}
}
}
// x-death 헤더가 없거나 파싱에 실패하면 초기 상태(재시도 횟수 0)로 간주
return 0;
}이제 재시도 횟수를 읽어와 지정한 재시도 횟수를 초과하면 재시도를 중단하고 실패처리하도록 할 수 있다.
Troubleshooting
Routing Key 변경 문제
하지만 문제가 발생했다. RetryQueue를 통해 다시 돌아온 메시지의 라우팅키가 변경된다는 것이다. # 패턴을 활용하여 여러 라우팅키를 받고 있었던 경우 라우팅키가 변경되면 특정 라우팅키에 해당하는 로직을 실행할 수 없게 된다.
이 문제를 해결하기 위해 xDeath 헤더를 이용했다. xDeath 헤더에는 재시도 횟수 뿐만 아니라 다양한 정보를 기록하고 있다.
xDeath 헤더의 내용을 잘보면 리스트 형태로, 최신순으로 배열되어 있다. 그리고 각 시점마다의 라우팅키를 기록하고 있는 것을 확인할 수 있다. 따라서 xDeath 헤더의 리스트의 마지막 요소의 routing-keys를 확인하면 해당 메시지가 최초로 가지고 있던 라우팅키를 확인할 수 있다.
/**
* 메시지의 Original Routing Key를 반환
* - x-death 헤더가 존재하면, 리스트의 마지막 항목(가장 오래된 이벤트)에서 추출
*/
private String getOriginalRoutingKey(Message message) {
MessageProperties properties = message.getMessageProperties();
Map<String, Object> headers = properties.getHeaders();
Object xDeathHeader = headers.get("x-death");
// x-death 헤더가 존재하고 List 형태인 경우
if (xDeathHeader instanceof List<?> deathList) {
if (!deathList.isEmpty()) {
// 리스트의 마지막 항목(가장 오래된/최초 이벤트)을 참조
Object firstDeathEntry = deathList.getLast();
if (firstDeathEntry instanceof Map<?, ?> deathMap) {
// RabbitMQ가 기록한 라우팅 키 리스트
Object routingKeys = deathMap.get("routing-keys");
if (routingKeys instanceof List<?> routingKeyList) {
if (!routingKeyList.isEmpty()) {
// 리스트의 첫 번째 키를 반환
return routingKeyList.getFirst().toString();
}
}
}
}
}
// x-death 헤더가 없거나 파싱에 실패하면, 현재 메시지가 받은 키를 반환
return properties.getReceivedRoutingKey();
}DLQ로 최종 실패 메시지 전송
여기까지 구현하면 모든 문제가 끝난 줄 알았지만, 이제 최종적으로 재시도 횟수를 초과했을 때 이 메세지를 어떻게 처리할 것인지를 구현해야 한다. 재시도 횟수를 초과했을 때 그냥 메시지를 거절하게 되면, 'x-dead-letter-routing-key'에 설정된 값에 따라 다시 RetryQueue로 전송하게 된다. 그럼 또 다시 무한히 재시도하는 문제가 반복되는 것이다.
따라서 재시도 횟수를 초과할 경우, 해당 메시지는 승인 처리하여 폐기하고, rabbitTemplate을 활용하여 DLQ로 실패한 메시지를 전송한다.
if (retryCount >= MAX_RETRIES) {
rabbitTemplate.convertAndSend(RabbitConfig.DLX, RabbitConfig.DLQ_ROUTING_KEY, message);
channel.basicAck(tag, false);
log.error("재시도 횟수 제한 초과, 메시지 처리 실패", e);
} else {
channel.basicNack(tag, false, false);
log.warn("처리 실패, {}번째 재시도", retryCount + 1, e);
}이러면 재시도 횟수만큼 RetryQueue로 보내 일정 시간이 지난 후 재시도하고, 재시도 횟수를 초과할 경우 해당 메시지를 폐기하여 재시도 루프를 끊고, 실패한 메시지는 최종적으로 DLQ로 보내 추후 실패 원인을 분석할 수 있다.
