RabbitMQ Dead-Letter-Queue 적용기
회사 업무를 하면서 이메일 자동발송, 상품 이미지 리사이징 등 부가 로직을 RabbitMQ 메세지큐에 태워서 처리하였는데, Queue에서 소비할 때 문제가 생겨 소비가 되지 않으면 무한정 시도하며 다음 큐 메세지가 소비가 되지 않는 문제가 있었다. 이를 개선하기 위해 Dead-Letter-Queue를 적용해보게 되었다.
Dead Letter Queue
DLQ(Dead Letter Queue)
DLQ(Dead Letter Queue)는 소프트웨어 시스템에서 오류로 인해 처리할 수 없는 메시지를 임시로 저장하는 특수한 유형의 메시지 대기열이다.
📫 DLQ = 실패한 메시지의 임시 저장소 = 배달 못한 편지 대기열
RabbitMQ 메시지큐 시스템은 소스 대기열의 메시지가 처리되지 못할 경우, 해당 Queue에 다시 넣고 성공하거나 보존 기간이 만료될 때까지 지속적으로 처리를 시도한다.
DLQ를 활용했을 때의 장점
- 통신 비용 절감
- 실패한 메시지가 만료될 때까지 처리를 시도하는 대신, 정해진 횟수의 재시도 후에 DLQ로 이동하여 통신 비용을 줄일 수 있다.
- 문제 해결 개선
- 오류의 원인을 식별하는 데 집중할 수 있다. (모니터링과 가시성)
- 시스템 안정성과 내결함성 향상
- 데이터 정합성
- 실패 후 재처리를 통하여 데이터의 정합성을 지킬 수 있다.
이로 인해 DLQ는 데이터 무결성, 내결함성, 안정성이 중요한 미션 크리티컬 시스템에 특히 유용하다.
(e.g. 금융, 물류, 의료 시스템)
하지만 로직에 따라 DLQ를 사용하지 말아야 할 때도 있다.
- Dead Letter Queue를 사용하면 안 되는 케이스
- 연결 실패에 대한 DLQ
- 순서보장이 필요한 경우(FIFO 큐)
🚗 Parking Lot Queue
정확히는, Dead-Letter Queue는 실패한 메시지를 “재시도, 재처리”하기 위한 대기열이다.
메시지를 그냥 소비할 수 없고 수동으로 처리해야 하거나 n번의 재시도 허용 횟수를 초과한 메시지를 “추가 처리”하기 위해 Parking Lot Queue(주차장 대기열) 개념이 있다.
Parking Lot Queue에서는 허용 횟수를 초과한 DLQ의 메시지를 전달 받아 추가 처리한다.
DLX, DLQ, Work Queue
Dead-Letter-Queue 역시 다른 RabbitMQ와 같이 Exchange/Queue 방식으로 전송된다.
DLQ의 교환기를 DLX라고 한다.
DLX(Dead-Letter-Exchange)
일반적인 교환기와 같다. 특정 메시지를 다른 Queue가 처리하지 못할 경우 해당 Exchange로 전송한다.
하나의 Queue로만 메시지를 처리하려면 Fanout Type, 각각의 Queue로 처리하려면 Direct Type으로 생성하면 된다.
DLQ(Dead-Letter-Queue)
DLX와 연결할 Queue이다. DLQ에서 전송받을 Exchange를 바인딩해준다.
Work Queue
비즈니스 로직을 처리할 Exchange와 Queue를 생성하고, 해당 큐에 x-dead-letter-exchange, x-dead-letter-routing-key arguments를 설정해준다. 메시지 처리를 실패할 경우 해당 DLX로 전달된다.
자동 생성 시 소스에 DLX, DLQ 정보와 arguments 설정을 추가해준다.
arguments 설정 추가
// Queue Bean 생성할 때(RabbitMQQueueConfig.class)
@Bean public Queue ~~~Queue() {
Map<String, Object> arguments = new HashMap<>();
arguments.put("x-dead-letter-exchange", "${dlx_name}");
arguments.put("x-dead-letter-routing-key", "${dlq_routing_key}");
return new Queue(${queue_name}, true, false, false, arguments);
}
// Consumer에서 큐 바인딩할 때(~~Consumer.class)
@RabbitListener(bindings = @QueueBinding(
value = @Queue(
value = "${queue_name}",
arguments = {
@Argument(name = "x-dead-letter-exchange", value = "${dlx_name}"),
@Argument(name = "x-dead-letter-routing-key", value = "${dlq_routing_key}")
}
),
exchange = @Exchange(value = "${exchange_name}"),
key = "${routing_key}"
))
public void ~~~Consumer(Message message) {
...
}
⚠️ 자동 생성 시 x-dead-letter-exchange , x-dead-letter-routing-key arguments가 설정이 되지 않았거나, RabbitMQ 콘솔에 동일한 이름의 arguments 설정이 다른 Queue가 존재할 경우 PRECONDITION_FAILED 에러 발생
만약, 수동으로 RabbitMQ 콘솔에서 DLX, DLQ를 생성하고 Work Queue에 DLQ를 설정하고 싶다면
DLX, DLQ를 생성한 뒤
Work Queue를 생성할 때 아래와 같이 arguments를 기입해주면 된다.
DLQ queuing (실패 메시지 처리)
WorkQueue의 메시지가 Consume하는 과정 중 실패하면 arguments에 설정된 DLX로 전달되도록 처리한다.
실패를 처리하는 방법은 다음과 같다.
- AmqpRejectAndDontRequeueException 예외 던지기
- Manual Rejection (수동 거부 처리)
위 두 가지 방법은 한 프로젝트 내에서 동시 사용 불가능
manual reject를 구현하려면 application.yml을 수정하기 때문에 전체 config에 영향이 있기 때문이다.
AmqpRejectAndDontRequeueException 예외 처리
- Spring AMQP에서 제공하는 특수한 예외 클래스
- RabbitMQ 메세지 소비 중 발생하는 예외를 처리할 때 사용되며, 메시지가 자동으로 거부되고 큐에 다시 들어가지 않도록 한다.
- AmqpRejectAndDontRequeueException 예외를 던지면 해당 큐에 arguments로 설정된 Dead Letter Exchange(DLX)로 전달된다.
Manual Rejection (수동 거부)
- Message와 Channel을 사용해 직접 manual reject process를 구현하는 방법
- Spring 설정 파일에서 메시지를 수동으로 승인/거부할 수 있도록 설정
- rabbitmq: # 생략 listener: direct: acknowledge-mode: manual # 수동 설정 simple: acknowledge-mode: manual # 수동 설정
- channel: Spring 서버와 RabbitMQ 사이에서 메시지가 이동하는 터널
- channel 사용하여 실패 시 메시지를 직접 거부, 승인
- acknowledge-mode: manual 설정으로 메시지 승인 또한 수동으로 해야 하므로 비효율적이다.
@RabbitListener(/*생략*/)
public void handler(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
try {
// consume 로직
} catch (Exception e) {
// 예외발생 시 메시지를 직접 거부한다.
channel.basicReject(tag, false);
}
// 예외가 발생하지 않는다면 메시지를 수동으로 승인한다.
channel.basicAck(tag, false);
}
Retry (재시도 로직)
근본적으로 잘못된 메시지는 재처리 과정이 불필요하지만, 외부 환경에 영향을 받은 경우에는 재처리 과정이 유의미하다. (Custom Error Handling으로 Exception에 따라 재처리 여부를 결정할 수도 있다.)
Consumer에서 예외 발생 시 DLX로 Queuing된 메시지를 n번 재시도 후 처리하는 로직을 추가한다.
재시도 로직을 추가하는 방법은 크게 두 가지가 있다.
- Spring boot 설정파일에 재시도 처리 작성 (Spring에서 제공하는 기능)
- 재시도 로직 직접 작성
설정 파일로 재시도 처리
- 설정 파일에 재시도 정보 설정
rabbitmq:
host: 118.67.129.5
port: 5672
username: admin
password: 1234
# 재처리 설정
listener:
simple:
retry:
enabled: true # 재시도
initial-interval: 3s # 최초 메시지 처리 실패 후 재시도까지의 인터벌
max-interval: 10s # 최대 재시도 인터벌
max-attempts: 5 # 최대 재시도 횟수
multiplier: 2 # 이전 interval * multiplier = 다음 interval- 위의 예시와 같이 설정하면, 다음과 같은 매커니즘으로 재시도 처리됨
- 1st ➡️ 3s ➡️ 2nd ➡️ 6s ➡️ 3th ➡️ 10s ➡️ 4th ➡️ 10s ➡️ 5th
- 설정 파일로 처리하는 재시도 방법에는 큰 단점이 존재한다.
- Spring은 메시지 A가 실패하여 재시도 처리 중일 때, 메시지 A가 성공적으로 소비되거나 재시도 처리 기간이 종료될 때까지 다음 메시지 B의 처리를 시작하지 않는다.
- 만약 1분 간격으로 재시도 3회를 하고, 영구적인 오류를 가지고 있다면 다음 메시지 처리 과정을 3분 동안 차단하게 된다.
직접 재시도 로직 구현 (retry_count 체크)
- 헤더 정보에 retry 횟수를 추가하여 직접 재시도 로직 구현
/*
* DLQ 재시도 로직
*/
public void processFailedMessagesRequeue(QueueEnums.ProcessType queueTypeEnum, Message message) {
Map<String, Object> headers = message.getMessageProperties().getHeaders();
if (headers.isEmpty()) {
return;
}
// header 필요 값 추출
Integer retriesCount = (Integer) headers.get(RETRY_COUNT_HEADER); // 현재 재시도 횟수
if (retriesCount == null) {
retriesCount = 0;
}
String exchange = message.getMessageProperties().getHeaders().get(EXCHANGE_HEADER).toString(); // 실패한 exchange
List<Map<String, Object>> xDeathList = (List<Map<String, Object>>) headers.get(X_DEATH_HEADER);
String routingKey = xDeathList.stream() // 실패한 routingKey
.filter(xDeathEntity -> xDeathEntity.containsKey(ROUTING_KEY))
.map(key -> (List<?>) key.get(ROUTING_KEY))
.filter(routingKeyList -> !routingKeyList.isEmpty())
.map(routingKeyList -> routingKeyList.get(0).toString())
.findAny()
.orElse("");
// 해당 queue type 인터벌 타임
int intervalMillis = queueTypeEnum.getIntervalSecond() * 1000;
try {
Thread.sleep(intervalMillis);
log.info("Retrying message for the {} time", ++retriesCount);
// 재시도 횟수 정보 추가
message.getMessageProperties().getHeaders().put(RETRY_COUNT_HEADER, retriesCount);
// 재시도
rabbitTemplate.send(exchange, routingKey, message);
} catch (Exception e) {
log.info("Fail retrying message at the {} time", ++retriesCount);
// 재시도 횟수 정보 추가
message.getMessageProperties().getHeaders().put(RETRY_COUNT_HEADER, retriesCount);
}
}- 재시도 횟수 초과 시 해당 메시지 소비 처리: deadLetterQueue 테이블에 insert
- deadLetterQueue 정보
- process_type_enum : 큐 처리 타입 enum
- queue_id : 해당 큐 테이블 PK
- json_data : DLQ 정보(reason, queue name, exchange name)
- deadLetterQueue 정보
Processing (후처리)
- deat_letter_queue 테이블에 DLQ 정보 insert
- Slack 통지
Dead Letter Queue 흐름
참고