RabbitMQ - Dead Letter Exchange
📘 RabbitMQ Dead Letter Exchange & TTL 설정
RabbitMQ는 기본적으로 메시지가 예상치 못하게 처리될 수 없을 경우 다시 Queue로 보내는 Re-Queuing을 수행합니다.
하지만 계속 동일한 에러로 메세지를 처리할 수 없을 경우, 이 메세지는 계속 Queue에 담기고 에러도 계속 생기는 루프가 될 것입니다.
RabbitMQ는 기본적으로 Delayed Message 기능을 지원하지 않으므로 플러그인을 설치하고 RabbitMQ를 재시작 해줍니다.
플러그인 설치, 서비스 재시작은 관리자 권한 CMD를 열어서 RabbitMQ가 설치된 디렉토리의 sbin 폴더 내부에서 진행해야 합니다.
# RabbitMQ Delayed Message Exchange 플러그인 설치
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
# RabbitMQ 재시작
rabbitmq-service.bat stop
rabbitmq-service.bat startMessage가 Dead Letter가 되는 3가지 조건
- Requeue False로 설정되어 있는 Consumer가 reject/nack 응답을 통해서 거절한 Message.
- Requeue True로 설정되어 있는 Consumer가 reject/nack 응답을 통해서 거절한 Message는, Message가 존재했던 Queue에 다시 Requeue되고 DLX 기능은 동작하지 않습니다.
- Per-message TTL (Time to Leave)이 만료한 Message.
- Queue가 가득차서 버려진 Message.
Dead Letter가 된 Message의 "x-death" Header에는 Message가 Dead Letter가 된 이유 및 관련 정보가 저장되어 있습니다.
"x-death" Header에 저장되어 있는 주요 정보
- reason : Message가 Dead Letter가 된 이유.
- time : Message가 Dead Letter가 시간.
- count : 동일한 reason, 동일한 queue에서 Message가 Dead Letter가 된 횟수.
- queue : Message가 Dead Letter가 되기전에 존재했던 Queue.
- exchange : Dead Letter가 된 Message를 마지막으로 처리한 Exchange. 여러번 DLX에 의해서 처리된 Message의 경우 DLX 정보가 저장되어 있을 수 있습니다.
이를 방지하고자 특정 횟수 이상의 Error가 발생한 메세지를 Dead Letter Exchange로 보내,
적절한 Error Handling 과정을 거치도록 설계해야 합니다.
Dead Letter Exchange 란?
- Dead Letter Exchange(DLX)란 처리할 수 없는 메시지를 전달하는 Exchange를 의미합니다.
TTL (Time to Live) 란?
- 메시지가 특정 Queue에 머무를 수 있는 시간을 설정할 수 있는 값입니다.
- Ququq별로 TTL 시간 세팅값은 각 어플레케이션 환경별로 개발자가 세팅할 수 있습니다.
예를 들어,
계속 Error가 발생한 메시지 & 특정 시간 Queue에 머무른 메시지들을 Dead Letter Exchange에 쌓고 Admin에게 푸시 알림이 보내 Error Handling 과정을 거치도록 설계할 수 있습니다.
대충 Flow를 그려보고 해당 그림처럼 RabbitMQ의 Exchange와 Queue를 설정해보겠습니다.
Exchange (DLX) 생성
- Exchange 생성 방법은 저번 글에서 포스팅 했으니 단순히 x.pics, x.pics.dlx 라는 이름의 아무 옵션이 안달린 2개의 Exchange를 만들어 줍니다.
Queue (DLQ) 생성
- 정상적인 데이터를 받을 q.pics 큐와 DLQ인 q.pics.dlx 큐를 만드는데 큐를 만들기 전 Arguments 부분에 설정해야 할 것이 있습니다.
- Argument 부분에 거부되거나 버려진 메시지(Dead Letter)에 대한 처리를 Key:Value 형식으로
x-dead-letter-exchange : x.pics.dlx로 설정해줍니다. - 이렇게 설정하면 버려지거나 거부된 메시지는
x.pics.dlxExchange로 들어가게 됩니다. - 그리고 DQL인
q.pics.dlx에서x.pics.dlx와 바인딩 해주면서 Argument로x-dead-letter-routing-key : 설정할값으로 Routing Key를 이욯애 바인딩 합니다.
📘 Client
지난번 MQTT Client를 만들때 썼던 코드인데 아래 코드에서 Stomp Client의 Subscribe 옵션인 connectHeaderWithAutoConfirm 변수를 봅시다.
StompHeader에 x-queue-type, x-message-ttl 등 RabbitMQ에서 설정한 Queue의 옵션들과 매핑 & 적용할 수 있습니다.
예를 들어, RabbitMQ 웹 관리 콘솔에서 Queue에 x-message-ttl을 적용했으면 이 클라이언트 코드에서도 적용을 해줘야 합니다.
만약 웹 관리 콘솔에 설정된 값을 클라이언트에서 헤더로 안넣어주면 Stomp 소켓이 열리지 않게 됩니다.
import React, { useEffect, useState } from 'react';
import { Client, StompHeaders } from '@stomp/stompjs';
interface RabbitMqWebSocketHandlerState {
messages: string[];
subscribed: boolean;
client: Client;
}
const RabbitMqWebSocketHandler: React.FC<RabbitMqWebSocketHandlerState> = () => {
const [messages, setMessages] = useState<string[]>([]);
const [subscribed, setSubscribed] = useState(false);
const [client, setClient] = useState<Client>();
// Life Cycle Hooks
useEffect(() => {
subscribeToQueue();
return () => {
unSubscribeFromQueue();
};
}, []);
// 구독 함수
const subscribeToQueue = () => {
const client = new Client({
brokerURL: 'ws://localhost:15674/ws',
// RabbitMQ 관리 콘솔 인증 정보
connectHeaders: {
login: 'guest',
passcode: 'guest',
},
debug: (str: string) => {
console.log(str);
},
});
// Stomp Client Header - AutoConfirm, Message TTL 옵션 추가
const connectHeadersWithAutoConfirm: StompHeaders = {
...client.connectHeaders,
'x-queue-type': 'quorum',
'x-message-ttl': 200000,
autoConfirm: true,
};
// Quorum Queue Subscribe
client.onConnect = () => {
console.log('Socket Connected');
// 1번째 파라미터로 Queue 이름, 2번째는 콜백 함수
client.subscribe('q.frame', (frame) => {
const newMessage = `Test - Message: ${frame.body}`;
setMessages((prevMessages) => [...prevMessages, newMessage]);
},
{
id: 'Test-Subscribe',
...connectHeadersWithAutoConfirm,
});
setSubscribed(true);
};
// 오류 메시지의 세부 정보 출력
client.onStompError = (frame) => {
console.error('STOMP error', frame.headers['message']);
console.log('Error Details:', frame.body);
};
setClient(client);
client.activate();
};
// 구독 해제 함수, 버튼을 클릭하면 구독을 해제함
const unSubscribeFromQueue = () => {
if (client) {
client.unsubscribe('Test-Subscribe');
setClient(null);
setSubscribed(false);
}
};
return (
<div>
<h2>RabbitMQ Listener</h2>
<ul>
{messages.map((message, index) => (
<li key={index}>
<p>{message}</p>
</li>
))}
</ul>
{!subscribed ? (
<button onClick={subscribeToQueue}>Subscribe</button>
) : (
// 구독 중일 때 해지 버튼
<button onClick={unSubscribeFromQueue}>Unsubscribe</button>
)}
</div>
);
};
export default RabbitMqWebSocketHandler;