RabbitMQ 메시지 패턴방식

RabbitMQ

• RabbitMQ는 오픈 소스 메시지 브로커 소프트웨어로서 AMQP(Advanced Message Queuing Protocol) 기반으로 동작합니다.
• RabbitMQ는 확장성, 신뢰성, 유연성 측면에서 높은 평가를 받으며, 메시지 큐의 대표적인 구현체 중 하나입니다. RabbitMQ의 주요 특징과 개념은 다음과 같습니다

---

주요 개념

1. Producer(프로듀서):
   • 메시지를 생성하고 큐(queue)에 보냅니다.
   • RabbitMQ에서 메시지는 교환기(exchange)를 통해 큐에 전달되기 때문에, Producer는 메시지를 교환기에 보냅니다.
2. Exchange(교환기):
   • 교환기는 Producer로부터 받은 메시지를 특정한 큐에 전달하는 역할을 합니다.
   • Exchange의 타입에는 direct, topic, fanout, headers가 있으며, 각 타입에 따라 메시지 라우팅 방식이 달라집니다.
     • Direct: 특정 라우팅 키를 기준으로 메시지를 전달.
     • Topic: 라우팅 키 패턴을 기반으로 메시지를 큐에 전달.
     • Fanout: 라우팅 키에 상관없이 모든 큐에 메시지를 브로드캐스팅.
     • Headers: 메시지의 헤더를 기반으로 라우팅.
3. Queue(큐):
   • 실제 메시지가 저장되는 장소입니다.
   • RabbitMQ는 여러 개의 큐를 생성할 수 있으며, 각 큐는 여러 소비자에 의해 처리될 수 있습니다.
4. Consumer(컨슈머):
   • 메시지를 큐에서 가져가 처리하는 역할을 합니다.
   • 소비자는 하나의 큐에 연결되며, 큐의 메시지를 가져가서 특정 작업을 수행합니다.
5. Binding(바인딩):
   • Exchange와 Queue 사이의 연결을 의미합니다. Binding을 통해 특정 큐가 특정 Exchange로부터 메시지를 받도록 설정할 수 있습니다.

---

RabbitMQ의 주요 기능 및 장점

1. 확장성:
   • RabbitMQ는 멀티 프로토콜 지원과 여러 교환기 타입, 다양한 라우팅 옵션을 통해 높은 확장성을 제공합니다.
2. 내구성:
   • 메시지를 디스크에 저장하여 시스템 장애 시에도 메시지가 유실되지 않도록 하는 내구성을 제공합니다.
3. 메시지 확인:
   • RabbitMQ는 메시지의 전송 상태를 확인하고, 필요 시 재전송하거나 소비되지 않은 메시지를 다시 처리하는 메커니즘을 제공합니다.
4. 플러그인 지원:
   • RabbitMQ는 다양한 플러그인을 지원하여 메시징 기능을 확장할 수 있습니다. 예를 들어, Federation, Shovel, 그리고 다양한 인증/보안 플러그인이 있습니다.
5. 모니터링 및 관리 도구:
   • RabbitMQ는 웹 기반의 관리 콘솔을 제공하여, 메시지의 흐름과 큐의 상태를 시각적으로 확인하고 모니터링할 수 있습니다.

---

RabbitMQ 동작 과정 요약

1. Producer가 메시지를 특정 Exchange에 발송.
2. Exchange가 메시지를 라우팅하여 특정 Queue에 전달.
3. Queue에서 대기 중인 메시지를 Consumer가 가져가서 처리.
4. 메시지가 처리되면 RabbitMQ는 메시지 삭제 혹은 상태 업데이트.

---

메시지 패턴

• 메시지 패턴은 메시지 큐(Message Queue)를 사용할 때, 메시지를 생산자(Producer)와 소비자(Consumer) 간에 어떻게 주고받을지에 대한 다양한 방법을 설명합니다.
• RabbitMQ에서는 다양한 메시지 패턴을 지원하며, 이는 애플리케이션의 요구사항과 아키텍처에 맞추어 메시지를 라우팅하고 처리하는데 사용됩니다. 대표적인 메시지 패턴은 다음과 같습니다:

1. 단일 큐 기반의 기본 패턴 (One-to-One 또는 Point-to-Point Pattern)

이 패턴은 Producer가 메시지를 특정 Queue에 전송하고, Consumer가 그 Queue에서 메시지를 읽는 가장 단순한 패턴입니다.

• 구조: Producer -> Queue -> Consumer
• 장점: 구현이 간단하며, 메시지의 손실 없이 안정적인 전송이 가능합니다.
• 사용 사례: 단일 작업 처리, 단일 소비자가 처리해야 하는 비동기 작업.

import pika

# RabbitMQ 연결 설정
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Queue 선언
channel.queue_declare(queue='hello')

# 메시지 보내기 (Producer)
def send_message():
    channel.basic_publish(exchange='', routing_key='hello', body='Hello RabbitMQ!')
    print("[Producer] Sent 'Hello RabbitMQ!'")

# 메시지 받기 (Consumer)
def receive_message():
    def callback(ch, method, properties, body):
        print(f"[Consumer] Received {body}")

    channel.basic_consume(queue='hello', on_message_callback=callback, auto_ack=True)
    print('[Consumer] Waiting for messages...')
    channel.start_consuming()

# Producer 실행
send_message()

# Consumer 실행
receive_message()

# 연결 종료
connection.close()

2. Fanout 패턴 (One-to-Many 또는 Publish/Subscribe Pattern)

Fanout 패턴에서는 Producer가 메시지를 Exchange에 보내면, Exchange는 이 메시지를 바인딩된 모든 Queue에 브로드캐스팅합니다. 즉, 모든 Queue가 동일한 메시지를 수신하게 됩니다. 이 패턴에서 Exchange 타입은 fanout으로 설정됩니다.

• 구조: Producer -> Fanout Exchange -> 여러 Queue -> 여러 Consumer
• 장점: 모든 Queue가 동일한 메시지를 받을 수 있어 여러 소비자가 동시에 동일한 작업을 처리할 때 유용합니다.
• 사용 사례: 로깅, 이벤트 브로드캐스트, 알림 전송.

# Producer 코드
def producer_fanout():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    
    # Fanout Exchange 선언
    channel.exchange_declare(exchange='logs', exchange_type='fanout')

    # 메시지 발행
    message = "Broadcasting to all queues!"
    channel.basic_publish(exchange='logs', routing_key='', body=message)
    print(f"[Producer] Sent: {message}")
    
    connection.close()

# Consumer 코드
def consumer_fanout():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    
    # Fanout Exchange 연결
    channel.exchange_declare(exchange='logs', exchange_type='fanout')

    # 임시 Queue 선언 (Exchange와 바인딩)
    result = channel.queue_declare(queue='', exclusive=True)
    queue_name = result.method.queue
    channel.queue_bind(exchange='logs', queue=queue_name)

    print('[Consumer] Waiting for messages...')
    
    def callback(ch, method, properties, body):
        print(f"[Consumer] Received: {body}")

    channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)
    channel.start_consuming()

3. Direct 패턴 (Selective Consumers 또는 라우팅 기반 Pattern)

Direct 패턴에서는 Producer가 메시지를 Exchange에 보내며, 이때 라우팅 키를 사용하여 메시지를 특정 Queue로 라우팅합니다. Exchange 타입은 direct로 설정됩니다. 라우팅 키는 메시지와 Queue 간의 1:1 매핑을 제공합니다.

• 구조: Producer -> Direct Exchange -> 특정 Queue (라우팅 키 매칭) -> Consumer
• 장점: 특정 메시지를 특정 큐로 전송할 수 있어, 메시지의 타겟팅이 가능.
• 사용 사례: 상태 변경 이벤트 처리, 특정 로그 레벨 필터링.

# Producer 코드
def producer_direct():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    
    # Direct Exchange 선언
    channel.exchange_declare(exchange='direct_logs', exchange_type='direct')

    # 특정 라우팅 키로 메시지 전송
    severity = 'info'  # 라우팅 키
    message = "This is an info log."
    channel.basic_publish(exchange='direct_logs', routing_key=severity, body=message)
    print(f"[Producer] Sent '{message}' with routing key '{severity}'")
    
    connection.close()

# Consumer 코드
def consumer_direct(severity):
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    
    # Direct Exchange 연결
    channel.exchange_declare(exchange='direct_logs', exchange_type='direct')

    # Queue 선언 및 바인딩
    result = channel.queue_declare(queue='', exclusive=True)
    queue_name = result.method.queue
    channel.queue_bind(exchange='direct_logs', queue=queue_name, routing_key=severity)

    print(f'[Consumer] Waiting for messages with routing key: {severity}')
    
    def callback(ch, method, properties, body):
        print(f"[Consumer] Received: {body}")

    channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)
    channel.start_consuming()

4. Topic 패턴 (Pattern Matching 또는 다중 라우팅 패턴)

Topic 패턴은 Direct 패턴과 비슷하지만, 라우팅 키의 패턴을 기반으로 여러 Queue에 메시지를 라우팅할 수 있습니다. Exchange 타입은 topic으로 설정됩니다. 라우팅 키는 점(.)으로 구분되며, 와일드카드(* 또는 #)를 통해 복잡한 패턴 매칭을 지원합니다.

• 구조: Producer -> Topic Exchange -> 여러 Queue (패턴 매칭) -> 여러 Consumer
• 장점: 라우팅 키 패턴을 통해 다양한 형태의 라우팅이 가능.
• 사용 사례: 복잡한 이벤트 필터링, 다중 관심사 기반의 메시지 전송.

# Producer 코드
def producer_topic():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    
    # Topic Exchange 선언
    channel.exchange_declare(exchange='topic_logs', exchange_type='topic')

    # 특정 라우팅 키 패턴으로 메시지 전송
    routing_key = 'system.info'
    message = "System Info message."
    channel.basic_publish(exchange='topic_logs', routing_key=routing_key, body=message)
    print(f"[Producer] Sent '{message}' with routing key '{routing_key}'")
    
    connection.close()

# Consumer 코드
def consumer_topic(binding_key):
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    
    # Topic Exchange 연결
    channel.exchange_declare(exchange='topic_logs', exchange_type='topic')

    # Queue 선언 및 바인딩
    result = channel.queue_declare(queue='', exclusive=True)
    queue_name = result.method.queue
    channel.queue_bind(exchange='topic_logs', queue=queue_name, routing_key=binding_key)

    print(f'[Consumer] Waiting for messages with binding key: {binding_key}')
    
    def callback(ch, method, properties, body):
        print(f"[Consumer] Received: {body}")

    channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)
    channel.start_consuming()

5. Headers 패턴 (Message Header 기반 라우팅)

Headers 패턴에서는 메시지의 헤더 정보에 따라 라우팅이 결정됩니다. Exchange 타입은 headers로 설정되며, 바인딩 시 헤더의 조건을 설정할 수 있습니다. 조건에 따라 특정 Queue로 메시지가 전달됩니다.

• 구조: Producer -> Headers Exchange -> 여러 Queue (헤더 조건 매칭) -> 여러 Consumer
• 장점: 메시지의 헤더에 포함된 정보를 기반으로 유연한 라우팅 가능.
• 사용 사례: 복잡한 속성 기반의 필터링과 라우팅, 다중 속성 조건이 필요한 경우.

# Producer 코드
def producer_headers():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    
    # Headers Exchange 선언
    channel.exchange_declare(exchange='headers_logs', exchange_type='headers')

    # 특정 헤더를 가진 메시지 전송
    headers = {'type': 'error', 'format': 'json'}
    message = "Error in JSON format"
    channel.basic_publish(
        exchange='headers_logs', 
        routing_key='', 
        body=message,
        properties=pika.BasicProperties(headers=headers)
    )
    print(f"[Producer] Sent '{message}' with headers {headers}")
    
    connection.close()

# Consumer 코드
def consumer_headers(required_headers):
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    
    # Headers Exchange 연결
    channel.exchange_declare(exchange='headers_logs', exchange_type='headers')

    # Queue 선언 및 바인딩
    result = channel.queue_declare(queue='', exclusive=True)
    queue_name = result.method.queue
    channel.queue_bind(exchange='headers_logs', queue=queue_name, arguments=required_headers)

    print(f'[Consumer] Waiting for messages with headers: {required_headers}')
    
    def callback(ch, method, properties, body):
        print(f"[Consumer] Received: {body}")

    channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)
    channel.start_consuming()

6. RPC 패턴 (Remote Procedure Call Pattern)

RPC 패턴은 비동기 메시징에서 요청-응답 구조를 모방하는 패턴입니다. Producer가 특정 Queue에 요청 메시지를 보내면, Consumer가 요청을 처리한 후 결과를 응답 메시지 형태로 다시 특정 Queue에 전달합니다.

• 구조: Producer -> 요청 Queue -> Consumer -> 응답 Queue -> Producer
• 장점: 비동기 시스템에서도 요청-응답 방식으로 통신할 수 있음.
• 사용 사례: 원격 호출을 통한 데이터 처리 요청 및 응답.

RabbitMQ에서의 RPC 패턴은 비동기 메시지 큐를 통해 원격 호출을 구현하는 방식입니다. 일반적으로 RPC 패턴은 두 개의 큐를 사용합니다:

1. 요청 큐(Request Queue): 클라이언트가 서버에 요청을 보내는 큐입니다.
2. 응답 큐(Response Queue): 서버가 클라이언트에 응답을 보내는 큐입니다.

클라이언트는 요청을 보낼 때 응답을 받을 큐의 이름과 고유한 correlation_id를 설정하여, 서버로부터 받은 응답이 자신이 보낸 요청에 대한 것임을 확인할 수 있습니다.

아래 코드는 Python의 pika 라이브러리를 사용하여 RabbitMQ에서 RPC 패턴을 구현한 예제입니다.

RPC 서버 코드 (Server)

import pika

def fibonacci(n):
    """간단한 Fibonacci 함수 예제"""
    if n == 0:
        return 0
    elif n == 1:
        return 1
    else:
        return fibonacci(n - 1) + fibonacci(n - 2)

# RabbitMQ 연결 설정
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 요청 큐 선언
channel.queue_declare(queue='rpc_queue')

def on_request(ch, method, properties, body):
    n = int(body)
    print(f"[Server] Received request for fib({n})")

    # 피보나치 계산 수행
    response = fibonacci(n)

    # 응답 전송
    ch.basic_publish(
        exchange='',
        routing_key=properties.reply_to,
        properties=pika.BasicProperties(correlation_id=properties.correlation_id),
        body=str(response)
    )
    ch.basic_ack(delivery_tag=method.delivery_tag)
    print(f"[Server] Sent response: {response}")

# 요청 처리 콜백 설정
channel.basic_consume(queue='rpc_queue', on_message_callback=on_request)

print("[Server] Awaiting RPC requests...")
channel.start_consuming()

RPC 클라이언트 코드 (Client)

import pika
import uuid

class FibonacciRpcClient:
    def __init__(self):
        # RabbitMQ 연결 설정
        self.connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
        self.channel = self.connection.channel()

        # 응답 큐 선언 및 소비자 설정
        result = self.channel.queue_declare(queue='', exclusive=True)
        self.callback_queue = result.method.queue

        self.channel.basic_consume(
            queue=self.callback_queue,
            on_message_callback=self.on_response,
            auto_ack=True
        )
        self.response = None
        self.corr_id = None

    def on_response(self, ch, method, properties, body):
        if self.corr_id == properties.correlation_id:
            self.response = body

    def call(self, n):
        self.response = None
        self.corr_id = str(uuid.uuid4())

        # 요청 메시지 전송
        self.channel.basic_publish(
            exchange='',
            routing_key='rpc_queue',
            properties=pika.BasicProperties(
                reply_to=self.callback_queue,
                correlation_id=self.corr_id,
            ),
            body=str(n)
        )

        # 응답 대기
        while self.response is None:
            self.connection.process_data_events()
        return int(self.response)

# 클라이언트 인스턴스 생성
fibonacci_rpc = FibonacciRpcClient()

print("[Client] Requesting fib(30)")
response = fibonacci_rpc.call(30)
print(f"[Client] Got response: {response}")

코드 설명

1. 서버 코드 (Server):
   • 서버는 rpc_queue라는 이름의 큐를 선언하고, 이를 통해 클라이언트로부터 요청을 수신합니다.
   • 서버가 요청을 수신하면 on_request 함수가 호출되며, 메시지 본문(body)로 받은 숫자를 사용해 Fibonacci 수를 계산합니다.
   • 계산 결과는 클라이언트가 응답을 받을 큐(reply_to)와 고유한 식별자(correlation_id)를 사용해 클라이언트로 전송합니다.
2. 클라이언트 코드 (Client):
   • 클라이언트는 서버에 요청을 보내기 전에 임시 응답 큐를 선언합니다.
   • 요청 메시지를 보낼 때, reply_to에 응답 큐의 이름을 지정하고, correlation_id에 고유한 ID를 설정합니다.
   • 서버로부터 응답을 받을 때 correlation_id를 비교하여, 요청에 대한 응답임을 확인합니다.

7. Work Queue 패턴 (Task Distribution 또는 Job Queue Pattern)

Work Queue 패턴은 하나의 Queue에 여러 Consumer를 연결하여 작업을 분산 처리하는 패턴입니다. Producer는 작업을 Queue에 넣고, 여러 Consumer가 이 Queue에서 작업을 꺼내 병렬로 처리합니다.

• 구조: Producer -> Queue -> 여러 Consumer
• 장점: 다수의 소비자가 작업을 병렬로 처리하여, 부하 분산과 빠른 작업 완료가 가능.
• 사용 사례: 대규모 데이터 처리, 병렬 작업 분배.

Producer

import pika
import sys

def send_task(message):
    # RabbitMQ 연결
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()

    # Queue 선언
    channel.queue_declare(queue='task_queue', durable=True)

    # 메시지 전송
    channel.basic_publish(
        exchange='',
        routing_key='task_queue',
        body=message,
        properties=pika.BasicProperties(
            delivery_mode=2,  # 메시지를 지속적으로 저장하기 위한 설정
        )
    )

    print(f"[Producer] Sent: {message}")
    connection.close()

# 메시지 입력
message = ' '.join(sys.argv[1:]) or "Hello World!"
send_task(message)

Consumer

import pika
import time

def callback(ch, method, properties, body):
    print(f"[Consumer] Received: {body.decode()}")
    
    # 작업 처리 (간단하게 메시지의 점(.) 개수만큼 대기 시간 설정)
    time_to_process = body.decode().count('.')
    time.sleep(time_to_process)
    
    print(f"[Consumer] Done processing {body.decode()}")
    
    # 메시지 확인(Acknowledgment)
    ch.basic_ack(delivery_tag=method.delivery_tag)

def start_consumer():
    # RabbitMQ 연결
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()

    # Queue 선언 (Producer와 동일한 설정)
    channel.queue_declare(queue='task_queue', durable=True)

    # 메시지 소비 설정
    channel.basic_qos(prefetch_count=1)  # 각 Consumer가 한 번에 하나의 메시지만 처리
    channel.basic_consume(queue='task_queue', on_message_callback=callback)

    print("[Consumer] Waiting for messages...")
    channel.start_consuming()

start_consumer()

코드 설명

1. Producer 코드:
   • RabbitMQ 서버와 연결을 설정하고, task_queue라는 Queue를 선언합니다. 이 Queue는 메시지를 디스크에 저장하기 위해 durable 옵션을 사용합니다.
   • 메시지를 전송할 때, delivery_mode=2로 설정하여 메시지가 지속적으로 저장되도록 합니다.
   • 기본적으로 메시지의 본문을 전달하고, 이를 작업으로 큐에 넣습니다.
2. Consumer 코드:
   • RabbitMQ 서버와 연결을 설정하고, task_queue라는 Queue를 선언합니다. 이 Queue 역시 지속성 옵션을 사용하여 정의합니다.
   • basic_qos를 통해 각 Consumer가 한 번에 하나의 메시지만 처리하도록 제한합니다. 이는 메시지 처리 부담을 분산하기 위해 중요한 설정입니다.
   • callback 함수는 메시지를 수신하면 실행되며, 메시지 본문에 포함된 점(.)의 개수만큼 작업 시간을 설정하여 지연을 시뮬레이션합니다.
   • 메시지 처리가 완료되면 basic_ack을 호출하여 RabbitMQ에게 메시지 처리가 완료되었음을 알립니다.

실행 방법

1. RabbitMQ 서버가 실행 중인지 확인합니다.
2. 각 Consumer를 여러 개의 터미널에서 실행하여, 작업 분배를 확인할 수 있습니다. (각 터미널에서 python consumer.py와 같이 실행)
3. 별도의 터미널에서 Producer를 실행하여, 메시지를 큐에 넣습니다. (예: python producer.py Hello....)

동작 설명

• 여러 개의 Consumer가 동시에 실행 중일 때, Producer가 메시지를 생성하면 각 메시지가 여러 Consumer 사이에 균등하게 분배됩니다.
• 메시지를 수신한 Consumer는 각기 다른 속도로 메시지를 처리하며, 기본적으로 한 번에 하나의 메시지만 수신하여 처리하도록 설정되어 있습니다.

---

장애 복구 전략

• RabbitMQ와 같은 메시지 큐 시스템에서는 장애 발생 시 메시지의 유실이나 서비스 중단을 최소화하고, 빠른 복구를 보장하기 위한 다양한 복구 전략이 중요합니다.
• 장애는 하드웨어, 네트워크, 소프트웨어, 운영 상의 문제 등 다양한 원인으로 발생할 수 있으며, 이러한 장애에 대비하기 위해 RabbitMQ는 여러 가지 복구 전략을 제공합니다.

RabbitMQ 장애 복구 전략

1. Durable Queue 및 Persistent Message:
   • Durable Queue: Queue를 Durable로 선언하면, RabbitMQ 서버가 재시작되더라도 Queue 자체가 삭제되지 않고 유지됩니다.
   • Persistent Message: 메시지를 전송할 때, 메시지의 속성을 persistent로 설정하여 메시지가 디스크에 저장되도록 할 수 있습니다. 이렇게 설정하면 서버가 갑작스럽게 종료되더라도 메시지가 손실되지 않습니다.
2. Mirrored Queue (High Availability):
   • RabbitMQ에서는 Mirrored Queue를 설정할 수 있습니다. 이는 특정 Queue를 클러스터 내의 다른 노드에 복제하여, 노드 장애 시에도 메시지가 손실되지 않도록 하는 기능입니다.
   • Mirrored Queue를 사용하면 메시지가 동시에 여러 노드에 복제되어, 장애 시에도 복제된 노드가 즉시 대체하여 메시지를 제공할 수 있습니다.
   • 복제본의 수, 동기화 설정 등을 조정하여 가용성과 성능을 균형있게 설정할 수 있습니다.
3. Clustering:
   • RabbitMQ는 여러 노드를 묶어서 클러스터링을 지원합니다. 클러스터는 장애가 발생해도 서비스가 지속되도록 노드 간의 상호 연결성을 유지하며, 각 노드가 서로의 상태를 동기화합니다.
   • 클러스터링은 노드가 추가되거나 제거될 때 동적으로 확장/축소가 가능하여, 장애 발생 시 특정 노드를 대체할 수 있습니다.
4. Federation 및 Shovel Plugin:
   • Federation: RabbitMQ의 Federation은 지리적으로 분산된 여러 RabbitMQ 인스턴스 간에 메시지를 전송할 수 있는 기능을 제공합니다. 이를 통해 장애 시 다른 데이터 센터나 서버로 메시지를 전송하여 가용성을 높일 수 있습니다.
   • Shovel: Shovel 플러그인은 RabbitMQ 인스턴스 간에 메시지를 자동으로 이동시켜주는 기능을 제공하며, 장애나 메시지 전달 실패 시에도 메시지를 지속적으로 재시도할 수 있도록 도와줍니다.
5. Message Acknowledgment (Ack/Nack) 및 Dead-Letter Queue:
   • RabbitMQ는 메시지를 소비자가 처리했을 때 Ack(인증)를 받습니다. 만약 Ack를 받지 못하면, RabbitMQ는 메시지를 다시 큐에 넣거나 Dead-Letter Queue로 전송하여 재처리를 시도할 수 있습니다.
   • Dead-Letter Queue는 특정 메시지가 반복적인 처리 실패나 TTL(Timeout)에 도달했을 때, 해당 메시지를 별도의 Queue에 보내어 추가적인 조치를 할 수 있도록 하는 전략입니다.
6. 백업 및 복구 전략:
   • RabbitMQ 설정, Queue 정의, 사용자 정보 등을 정기적으로 백업하여, RabbitMQ가 완전히 복구 불가능한 상태가 되더라도 데이터를 복원할 수 있도록 합니다.
   • 백업 주기와 복구 시뮬레이션을 정기적으로 시행하여 실제 장애 발생 시 복구 계획이 제대로 작동하는지 점검하는 것이 좋습니다.
7. 자동 Failover:
   • 클러스터링과 함께 RabbitMQ는 자동으로 장애를 감지하고, 장애 발생 시 즉시 대체 노드로 전환하는 Failover 기능을 제공합니다.
   • 클러스터 내에서 리더 노드를 재선출하여 서비스 지속성을 유지하는 방식으로, 소비자와 생산자 측에서 가용성이 유지됩니다.
8. 모니터링 및 알림 시스템:
   • RabbitMQ는 내장 모니터링 도구와 Prometheus와 같은 외부 모니터링 시스템과 통합하여, 노드의 상태, 큐의 길이, 메시지 처리 속도 등을 실시간으로 모니터링할 수 있습니다.
   • 모니터링 시스템은 장애 발생 시 알림을 보내어 관리자나 DevOps 팀이 신속하게 조치를 취할 수 있도록 돕습니다.

---

요약

RabbitMQ는 분산 시스템에서 애플리케이션 간의 통신을 효율적으로 설계할 수 있도록 돕고, 메시지 손실 없이 안정적으로 데이터를 전달할 수 있는 강력한 메시징 시스템입니다.