Dead Letter Queue 재처리와 Retry

Dead Letter Queue와 Dead Letter Exchange

DLQ (Dead Letter Queue)

실패한 메시지가 저장되는 큐. 메시지가 큐에서 제대로 처리 되지 못할 경우 DLQ로 이동됨.

 

DLQ로 이동하는 경우

• NACK 처리나 거부: basic.reject 혹은 basic.nack
• TTL 만료
• Overflow: 큐에 설정된 최대 메시지 갯수를 초과하면 가장 오래된 메시지가 삭제되고 DLQ로 이동

DLX (Dead Letter Exchange)

실패한 메시지를 어디로 보낼지 라우팅하는 Exchange.

DLQ로 바로 보내는 것이 아니라 RabbitMQ는 Exchange를 통해 라우팅한다.

 

전체 구조

Producer
      ↓
Exchange
      ↓
OrderQueue
      ↓ (reject / nack / ttl)
DeadLetterExchange
      ↓
DeadLetterQueue

---

Dead Letter Queue를 활용한 실패 메시지 재처리

1. DLQ와 DLX 빈 선언

@Configuration
public class RabbitMQConfig {
	public static final String ORDER_COMPLETED_QUEUE = "order_completed_queue";
	public static final String ORDER_EXCHANGE = "order_completed_exchange";
	public static final String DLQ = "deadLetterQueue";
	public static final String DLX = "deadLetterExchange";

	@Bean
	public TopicExchange orderExchange() {
		return new TopicExchange(ORDER_EXCHANGE);
	}

	@Bean
	public TopicExchange deadLetterExchange() {
		return new TopicExchange(DLX);
	}

	// 메시지가 처리되지 못했을 경우 자동으로 Dead Letter Queue로 이동하도록 설정
	@Bean
	public Queue orderqueue() {
		// return new Queue(ORDER_COMPLETED_QUEUE, false);
		return QueueBuilder.durable(ORDER_COMPLETED_QUEUE)
			.withArgument("x-dead-letter-exchange", DLX) // Dead Letter Exchange 설정
			.withArgument("x-dead-letter-routing-key", DLQ) // Dead Letter Routing Key 설정
			.ttl(5000)
			.build();
	}

	@Bean
	public Queue deadLetterQueue() {
		return new Queue(DLQ);
	}

	@Bean
	public Binding orderComplededBinding() {
		return BindingBuilder.bind(orderqueue()).to(orderExchange()).with("order.completed.#"); //bindingKey를 넣는건데, Spring API에서는 routingKey 파라미터
	}
	// Binding도 routing key를 기준으로 정의되기 때문에, AMQP에는 “binding key”라는 독립된 개념이 없다.
	//우리가 흔히 말하는 “binding key”는 AMQP 공식 용어라기보다는 설명 편의를 위한 표현에 가깝다.


	@Bean
	public Binding deadLetterBinding() {
		return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange()).with(DLQ);
	}
}

 

2. Ack/Nack 처리를 개발 시점에 직접 핸들링 하도록 설정

SimpleRabbitListenerContainerFactory를 이용하여 AcknowledgeMode 모드를 Manual로 설정.

AcknowledgeMode.MANUAL을 통해 메시지의 처리 결과를 RabbitMQ에 명시적으로 전달하고
basicAck , basicNack , basicReject 를 사용하여 메시지 재시도 및 DLQ 처리를 유연하게 구현할 수 있다.

/**
 * RabbitMQ 설정을 수동으로 하는 클래스
 */
@EnableRabbit //RabbitMQ 리스너(@RabbitListener)을 스캔하여 동작하게 만드는 활성화 어노테이션(스프링부트는 자동)
@Configuration
public class RabbitMQManualConfig {
	@Bean
	public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
		SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
		factory.setConnectionFactory(connectionFactory);
		factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
		return factory;
	}
}

 

3. Producer 메시지 발행

@Component
@RequiredArgsConstructor
public class OrderProducer {

	private final RabbitTemplate rabbitTemplate;

	public void sendShipping(String message) {
		rabbitTemplate.convertAndSend(
			RabbitMQConfig.ORDER_EXCHANGE,
			"order.completed.shipping",
			message);
		System.out.println("[주문 완료. 배송 지시 메시지 생성 : " + message + "]");
	}
}

 

4. Consumer 에서 메시지 실패 건에 대해 3번의 재시도 후 DLQ 이동

@Component
public class OrderConsumer {

	private static final int MAX_RETRIES = 3; // 최대 재시도 횟수
	private int retryCount = 0; // 현재 재시도 횟수

	// containerFactory: 리스너 컨테이너의 설정(ACK 모드, 재시도, prefetch, concurrency 등)을 바꾸기 위해 지정
	// @Header("amqp_deliveryTag"): RabbitMQ가 메시지에 붙여주는 전달 식별자(ack/nack/reject에 필요)를 Spring이 꺼내서 주입해주는 것
	// Channel: RabbitMQ Java Client의 현재 연결/채널(프로토콜 세션) 객체로, basicAck/basicNack/basicReject 같은 저수준 AMQP 명령을 보내는 통로
	@RabbitListener(queues = RabbitMQConfig.ORDER_COMPLETED_QUEUE, containerFactory = "rabbitListenerContainerFactory")
	public void processOrder(String message, Channel channel, @Header("amqp_deliveryTag") long tag) {
		try {
			// 실패 유발
			if ("fail".equalsIgnoreCase(message)) {
				if (retryCount < MAX_RETRIES) {
					System.err.println("#### Fail & Retry: " + message + " (Retry Count: " + retryCount + ")");
					retryCount++;
					throw new RuntimeException(message);
				} else {
					System.err.println("#### 최대 횟수 초과, DLQ로 이동 시킴");
					retryCount = 0;
					// deliveryTag: 메시지 고유식별 태그, multiple: true면 deliveryTag 이하 메시지 전부, requeue: true면 다시 큐로, false면 DLQ or drop
					channel.basicNack(tag, false, false); // 메시지 거부, 재큐잉하지 않음 (DLQ로 이동)
					return;
				}
			}
			// 성공 처리
			System.out.println("# 성공: " + message);
			channel.basicAck(tag, false); // 메시지 수동 ACK
			retryCount = 0;
		} catch (Exception e) {
			System.err.println("# error 발생 : " + e.getMessage());
			try {
				// 실패 시 basicReject를 사용하여 메시지를 재처리 전송
				channel.basicReject(tag, true); //basicReject는 단일 메시지 거부 전용 (간단 버전), basicNack은 확장형 거부 (여러 메시지 가능)
			} catch (IOException ex) {
				System.err.println("# fail & reject message : " + ex.getMessage());
			}
		}
	}
}

 

5. DLQ 컨슈머에서 메시지 수정하여 큐잉

@Component
@RequiredArgsConstructor
public class OrderDLQConsumer {

	private final RabbitTemplate rabbitTemplate;

	@RabbitListener(queues = RabbitMQConfig.DLQ)
	public void process(String message) {
		System.out.println("DLQ Message Received: " + message);

		try {
			String fixedMessage = "success";

			rabbitTemplate.convertAndSend(RabbitMQConfig.ORDER_EXCHANGE,
				"order.completed.shipping",
				fixedMessage
			);
			System.out.println("DLQ Message Sent: " + fixedMessage);
		} catch (Exception e) {
			System.err.println("### [DLQ Consumer Error] " + e.getMessage());
		}
	}
}

 

channel.basicReject(deliveryTag, requeue);

channel.basicNack(deliveryTag, multiple, requeue); // DLQ로 메시지 이동

위와 같은 설정은 처리가 다소 복잡하고, 메서드 호출도 혼동되기 쉬우므로 이런 처리보다는

Spring AMQP에서 제공하는 RetryTemplate을 통해 좀더 명확하고 간단하게 기능을 구현할 수 있다.

 

---

[발전1] RetryTemplate을 통한 간편한 재처리 설정

Spring AMQP는 RetryTemplate 을 통해 재시도 로직을 지원한다.

이 경우 Spring AMQP에서 AcknowledgeMode가 AUTO로 기본 세팅되어 있기 때문에 별도의 SimpleRabbitListenerContainerFactory를 통하지 않고 자동으로 Ack/Nack 처리가 가능하다.

• 재시도 중 메시지가 성공적으로 처리되면 Spring AMQP가 자동으로 Ack를 전송
• 모든 재시도가 실패하면 Nack를 보내고 RabbitMQ가 메시지를 DLQ로 이동

※ build.gradle 에 retry dependency 추가 필요

RetryConfig 클래스에서 기본 설정 세팅하여 빈으로 선언

@Configuration
public class RetryConfig {

	@Bean
	public RetryTemplate retryTemplate() {
		RetryTemplate retryTemplate = new RetryTemplate();
		// 재시도 정책 설정
		SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
		retryPolicy.setMaxAttempts(3); // 최대 재시도 횟수 설정
		// 백오프 정책 설정
		FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
		backOffPolicy.setBackOffPeriod(1000L); // 재시도 간격 설정 (1초)
		retryTemplate.setRetryPolicy(retryPolicy);
		retryTemplate.setBackOffPolicy(backOffPolicy);

		return retryTemplate;
	}
}

 

Consumer에서 retryTemplate으로 재시도 및 DLQ 이관

retryTemplate의 retryCount를 확인하여 세번 이하라면 throw e를 던짐으로써 (channel.basicReject(tag, true)) , RetryTemplate이 예외를 감지해서 BackOffPolicy에 따라 1초 후 네번까지 재실행(동일한 로직 실행)

@Component
@RequiredArgsConstructor
public class OrderConsumer {

	private final RabbitTemplate rabbitTemplate;
	private final RetryTemplate retryTemplate;

	@RabbitListener(queues = RabbitMQConfig.ORDER_COMPLETED_QUEUE)
	public void consume(String message) {
		retryTemplate.execute(context -> {
			try {
				System.out.println("# 리시브 메시지: " + message + " | 시도 횟수: " + (context.getRetryCount() + 1));

				if ("fail".equals(message)) {
					throw new RuntimeException(message);
				}
				System.out.println("# 메시지 처리 성공: " + message);
			} catch (Exception e) {
				if (context.getRetryCount() + 1 >= 3) {
					rabbitTemplate.convertAndSend(
						RabbitMQConfig.ORDER_TOPIC_DLX,
						RabbitMQConfig.DEAD_LETTER_ROUTING_KEY,
						message);
				} else {
					throw e; // RetryTemplate이 재시도를 수행하도록 예외를 다시 던짐
				}
			}
			return null;
		});
	}
}

 

ACK/NACK를 수동으로 던질 필요가 없어 훨씬 깔끔하다!

---

[발전2] Application.yml 설정을 통한 RetryTemplate 속성 정의

application.yml 에 rabbitmq listener retry 설정 추가

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guestuser
    password: guestuser
    listener:
      simple:
        retry:
          enabled: true # 재시도 활성화
          initial-interval: 1000 # 첫 재시도 간격 1초
          max-attempts: 3 # 최대 재시도 횟수
          max-interval: 10000 # 최대 재시도 간격 10초
        default-requeue-rejected: false # 재시도 실패 시 자동으로 DLQ로 이동

 

Consumer 로직 변경

대부분의 설정을 xml에 해두었기 때문에, execption만 던지면 retry 한다. 설정 횟수를 넘으면 DLQ로 간다.

@Component
public class OrderConsumer {
	private int retryCount;

	@RabbitListener(queues = RabbitMQConfig.ORDER_COMPLETED_QUEUE)
	public void consume(String message) {
		System.out.println("Received message: " + message + "count: " + retryCount++);
		if ("fail".equals(message)) {
			throw new RuntimeException("Processing failed. Retry: " + message);
		}
		System.out.println("Message processed successfully: " + message);
	}
}

---

참고 자료 & 이미지 출처
RabbitMQ를 이용한 비동기 아키텍처 한방에 해결하기