Python RabbitMQ 개념 정리

1️⃣ 기본 연결

RabbitMQ는 강력한 오픈소스 메시지 브로커로, AMQP 프로토콜을 기반으로 안정적인 메시지 큐잉 시스템을 제공한다. Python에서는 pika 라이브러리를 통해 쉽게 연결하고 사용할 수 있다.

import pika
from typing import Optional, Callable, Dict, Any
import json
import logging
import time
from contextlib import contextmanager

class RabbitMQClient:
    """RabbitMQ 서버와의 연결을 관리하는 클라이언트 클래스"""
    
    def __init__(
        self,
        host: str = 'localhost',
        port: int = 5672,
        username: str = 'guest',
        password: str = 'guest',
        virtual_host: str = '/',
        connection_attempts: int = 3,
        retry_delay: int = 5,
        heartbeat: int = 600
    ):
        """
        RabbitMQ 클라이언트 초기화
        
        Args:
            host: RabbitMQ 서버 호스트
            port: RabbitMQ 서버 포트
            username: 인증 사용자 이름
            password: 인증 비밀번호
            virtual_host: 가상 호스트
            connection_attempts: 연결 시도 횟수
            retry_delay: 재시도 간격(초)
            heartbeat: 하트비트 간격(초)
        """
        self.logger = logging.getLogger(__name__)
        
        # 인증 정보 설정
        self.credentials = pika.PlainCredentials(username, password)
        
        # 연결 파라미터 설정
        self.parameters = pika.ConnectionParameters(
            host=host,
            port=port,
            virtual_host=virtual_host,
            credentials=self.credentials,
            connection_attempts=connection_attempts,
            retry_delay=retry_delay,
            heartbeat=heartbeat,
            socket_timeout=10
        )
        
        self.connection = None
        self.channel = None
    
    def create_connection(self):
        """
        RabbitMQ 서버와의 새 연결 생성
        
        Returns:
            BlockingConnection: RabbitMQ 연결 객체
        """
        try:
            self.logger.info("RabbitMQ 서버에 연결 중...")
            connection = pika.BlockingConnection(self.parameters)
            self.logger.info("RabbitMQ 서버에 연결됨")
            self.connection = connection
            return connection
        except pika.exceptions.AMQPConnectionError as e:
            self.logger.error(f"RabbitMQ 연결 실패: {e}")
            raise
    
    def create_channel(self):
        """
        RabbitMQ 채널 생성
        
        Returns:
            BlockingChannel: RabbitMQ 채널 객체
        """
        if not self.connection or self.connection.is_closed:
            self.create_connection()
            
        self.channel = self.connection.channel()
        return self.channel
    
    def close(self):
        """연결 및 채널 닫기"""
        if self.channel and self.channel.is_open:
            self.logger.info("RabbitMQ 채널 닫는 중...")
            self.channel.close()
            
        if self.connection and self.connection.is_open:
            self.logger.info("RabbitMQ 연결 닫는 중...")
            self.connection.close()
    
    @contextmanager
    def connection_context(self):
        """
        컨텍스트 매니저를 통한 RabbitMQ 연결 관리
        
        사용 예:
        ```
        with client.connection_context() as channel:
            channel.queue_declare(queue='hello')
            channel.basic_publish(exchange='', routing_key='hello', body='Hello World!')
        ```
        """
        connection = None
        channel = None
        try:
            connection = self.create_connection()
            channel = connection.channel()
            yield channel
        finally:
            if channel and channel.is_open:
                channel.close()
            if connection and connection.is_open:
                connection.close()
    
    def check_queue_exists(self, queue_name: str) -> bool:
        """
        큐가 존재하는지 확인
        
        Args:
            queue_name: 확인할 큐 이름
            
        Returns:
            bool: 큐 존재 여부
        """
        with self.connection_context() as channel:
            try:
                channel.queue_declare(queue=queue_name, passive=True)
                return True
            except pika.exceptions.ChannelClosedByBroker:
                return False

✅ 특징:

유연한 연결 설정 및 파라미터 구성
인증 정보 관리 및 보안 연결
재시도 로직을 통한 내결함성 제공
컨텍스트 매니저를 통한 자원 관리
채널 생성 및 관리
로깅을 통한 연결 상태 추적
가상 호스트 지원
타입 힌팅을 통한 코드 가독성 향상

2️⃣ 기본 메시징

RabbitMQ에서 메시지 주고받기는 프로듀서와 컨슈머라는 두 핵심 컴포넌트를 통해 이루어진다. 프로듀서는 메시지를 발행하고, 컨슈머는 메시지를 소비한다.

class RabbitMQProducer:
    """RabbitMQ 메시지 발행을 담당하는 프로듀서 클래스"""
    
    def __init__(self, client: RabbitMQClient):
        """
        RabbitMQ 프로듀서 초기화
        
        Args:
            client: RabbitMQ 클라이언트 인스턴스
        """
        self.client = client
        self.logger = logging.getLogger(__name__)
    
    def publish(
        self,
        queue_name: str,
        message: Dict[str, Any],
        durable: bool = True,
        exchange: str = '',
        routing_key: str = None,
        properties: pika.BasicProperties = None,
        mandatory: bool = False
    ):
        """
        메시지 발행
        
        Args:
            queue_name: 큐 이름
            message: 발행할 메시지 (딕셔너리)
            durable: 큐 지속성 여부
            exchange: 교환기 이름
            routing_key: 라우팅 키 (지정하지 않으면 queue_name 사용)
            properties: 메시지 속성
            mandatory: 메시지 전달 보장 여부
        
        Returns:
            bool: 메시지 발행 성공 여부
        """
        if routing_key is None:
            routing_key = queue_name
            
        if properties is None:
            # 기본 속성 설정 - 메시지 지속성 보장
            properties = pika.BasicProperties(
                delivery_mode=2,  # 메시지 지속성
                content_type='application/json',
                timestamp=int(time.time()),
                message_id=f"{int(time.time())}-{id(message)}"
            )
        
        try:
            with self.client.connection_context() as channel:
                # 큐 선언
                channel.queue_declare(queue=queue_name, durable=durable)
                
                # 메시지 발행
                channel.basic_publish(
                    exchange=exchange,
                    routing_key=routing_key,
                    body=json.dumps(message),
                    properties=properties,
                    mandatory=mandatory
                )
                
                self.logger.info(f"메시지 발행 성공: queue={queue_name}, routing_key={routing_key}")
                return True
                
        except Exception as e:
            self.logger.error(f"메시지 발행 실패: {e}")
            return False
    
    def publish_batch(
        self,
        queue_name: str,
        messages: list,
        batch_size: int = 100,
        durable: bool = True
    ):
        """
        배치 메시지 발행
        
        Args:
            queue_name: 큐 이름
            messages: 발행할 메시지 목록
            batch_size: 배치 크기
            durable: 큐 지속성 여부
            
        Returns:
            dict: 성공/실패 메시지 수
        """
        with self.client.connection_context() as channel:
            # 큐 선언
            channel.queue_declare(queue=queue_name, durable=durable)
            
            success_count = 0
            failure_count = 0
            
            for i in range(0, len(messages), batch_size):
                batch = messages[i:i+batch_size]
                
                for message in batch:
                    try:
                        # 메시지 발행
                        channel.basic_publish(
                            exchange='',
                            routing_key=queue_name,
                            body=json.dumps(message),
                            properties=pika.BasicProperties(
                                delivery_mode=2,  # 메시지 지속성
                                content_type='application/json'
                            )
                        )
                        success_count += 1
                    except Exception as e:
                        self.logger.error(f"배치 메시지 발행 실패: {e}")
                        failure_count += 1
            
            return {
                'total': len(messages),
                'success': success_count,
                'failure': failure_count
            }

class RabbitMQConsumer:
    """RabbitMQ 메시지 소비를 담당하는 컨슈머 클래스"""
    
    def __init__(self, client: RabbitMQClient):
        """
        RabbitMQ 컨슈머 초기화
        
        Args:
            client: RabbitMQ 클라이언트 인스턴스
        """
        self.client = client
        self.logger = logging.getLogger(__name__)
        self.should_stop = False
    
    def consume(
        self,
        queue_name: str,
        callback: Callable,
        durable: bool = True,
        prefetch_count: int = 1,
        auto_ack: bool = False
    ):
        """
        메시지 소비
        
        Args:
            queue_name: 소비할 큐 이름
            callback: 메시지 처리 콜백 함수
            durable: 큐 지속성 여부
            prefetch_count: 한 번에 가져올 메시지 수
            auto_ack: 자동 확인 여부
        """
        def wrapped_callback(ch, method, properties, body):
            """
            콜백 래퍼 함수
            
            Args:
                ch: 채널
                method: 전달 메서드
                properties: 메시지 속성
                body: 메시지 본문
            """
            try:
                message = json.loads(body)
                self.logger.info(f"메시지 수신: {queue_name}")
                
                # 사용자 정의 콜백 실행
                result = callback(ch, method, properties, message)
                
                # 자동 확인이 아닌 경우 메시지 확인
                if not auto_ack:
                    ch.basic_ack(delivery_tag=method.delivery_tag)
                
                return result
                
            except json.JSONDecodeError as e:
                self.logger.error(f"메시지 형식 오류: {e}")
                # 메시지 형식 오류 시 거부 (재큐잉 없음)
                if not auto_ack:
                    ch.basic_reject(delivery_tag=method.delivery_tag, requeue=False)
            except Exception as e:
                self.logger.error(f"메시지 처리 오류: {e}")
                # 기타 오류 시 재큐잉
                if not auto_ack:
                    ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
        
        # 연결 및 채널 생성
        connection = self.client.create_connection()
        channel = connection.channel()
        
        try:
            # 큐 선언
            channel.queue_declare(queue=queue_name, durable=durable)
            
            # QoS 설정
            channel.basic_qos(prefetch_count=prefetch_count)
            
            # 콜백 설정
            channel.basic_consume(
                queue=queue_name,
                on_message_callback=wrapped_callback,
                auto_ack=auto_ack
            )
            
            self.logger.info(f"큐 {queue_name}에서 메시지 대기 중...")
            
            # 메시지 소비 시작
            while not self.should_stop:
                connection.process_data_events(time_limit=1)  # 이벤트 처리
                
        except KeyboardInterrupt:
            self.logger.info("컨슈머 중지 요청")
            self.should_stop = True
        except Exception as e:
            self.logger.error(f"컨슈머 오류: {e}")
        finally:
            self.logger.info("컨슈머 종료")
            if channel and channel.is_open:
                channel.close()
            if connection and connection.is_open:
                connection.close()
    
    def stop(self):
        """컨슈머 중지"""
        self.should_stop = True

# 사용 예제
def message_handler(ch, method, properties, message):
    """
    메시지 처리 핸들러 함수
    
    Args:
        ch: 채널
        method: 전달 메서드
        properties: 메시지 속성
        message: 디코딩된 메시지 (dict)
    """
    print(f"메시지 처리: {message}")
    # 메시지 처리 로직...
    return True

# 예제 실행
if __name__ == "__main__":
    # 로깅 설정
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    
    # 클라이언트 생성
    client = RabbitMQClient(
        host='localhost',
        port=5672,
        username='guest',
        password='guest'
    )
    
    # 프로듀서 생성 및 메시지 발행
    producer = RabbitMQProducer(client)
    producer.publish(
        queue_name='example_queue',
        message={'data': 'Hello RabbitMQ!', 'timestamp': time.time()}
    )
    
    # 컨슈머 생성 및 메시지 소비
    consumer = RabbitMQConsumer(client)
    try:
        consumer.consume(
            queue_name='example_queue',
            callback=message_handler,
            prefetch_count=10
        )
    except KeyboardInterrupt:
        consumer.stop()

✅ 특징:

강력한 메시지 발행 및 소비 기능
컨텍스트 매니저를 통한 자원 관리
배치 처리 지원으로 성능 최적화
메시지 형식 검증 및 오류 처리
QoS 설정을 통한 부하 관리
메시지 확인 및 거부 메커니즘
재시도 로직과 내결함성
로깅을 통한 메시지 상태 추적
타입 힌팅으로 코드 가독성 및 안정성 향상
컨슈머 우아한 종료 지원

3️⃣ 고급 메시징 패턴

RabbitMQ는 단순한 점대점 메시징 이상의 다양한 고급 메시징 패턴을 제공한다. 교환기(Exchange)는 메시지 라우팅의 핵심 구성 요소로, 여러 유형을 통해 복잡한 메시징 요구사항을 구현할 수 있다.

class RabbitMQExchange:
    """RabbitMQ 교환기를 관리하는 클래스"""
    
    def __init__(self, client: RabbitMQClient):
        """
        RabbitMQ 교환기 관리자 초기화
        
        Args:
            client: RabbitMQ 클라이언트 인스턴스
        """
        self.client = client
        self.logger = logging.getLogger(__name__)
    
    def setup_fanout(self, exchange_name: str, durable: bool = True):
        """
        팬아웃 교환기 설정 - 모든 큐에 메시지 브로드캐스트
        
        Args:
            exchange_name: 교환기 이름
            durable: 교환기 지속성 여부
            
        Returns:
            pika.channel.Channel: 설정된 채널
        """
        with self.client.connection_context() as channel:
            channel.exchange_declare(
                exchange=exchange_name,
                exchange_type='fanout',
                durable=durable
            )
            self.logger.info(f"팬아웃 교환기 '{exchange_name}' 설정 완료")
            return channel
    
    def setup_direct(self, exchange_name: str, durable: bool = True):
        """
        다이렉트 교환기 설정 - 라우팅 키 기반 메시지 전달
        
        Args:
            exchange_name: 교환기 이름
            durable: 교환기 지속성 여부
            
        Returns:
            pika.channel.Channel: 설정된 채널
        """
        with self.client.connection_context() as channel:
            channel.exchange_declare(
                exchange=exchange_name,
                exchange_type='direct',
                durable=durable
            )
            self.logger.info(f"다이렉트 교환기 '{exchange_name}' 설정 완료")
            return channel
    
    def setup_topic(self, exchange_name: str, durable: bool = True):
        """
        토픽 교환기 설정 - 패턴 매칭 기반 메시지 전달
        
        Args:
            exchange_name: 교환기 이름
            durable: 교환기 지속성 여부
            
        Returns:
            pika.channel.Channel: 설정된 채널
        """
        with self.client.connection_context() as channel:
            channel.exchange_declare(
                exchange=exchange_name,
                exchange_type='topic',
                durable=durable
            )
            self.logger.info(f"토픽 교환기 '{exchange_name}' 설정 완료")
            return channel
    
    def setup_headers(self, exchange_name: str, durable: bool = True):
        """
        헤더 교환기 설정 - 헤더 속성 기반 메시지 전달
        
        Args:
            exchange_name: 교환기 이름
            durable: 교환기 지속성 여부
            
        Returns:
            pika.channel.Channel: 설정된 채널
        """
        with self.client.connection_context() as channel:
            channel.exchange_declare(
                exchange=exchange_name,
                exchange_type='headers',
                durable=durable
            )
            self.logger.info(f"헤더 교환기 '{exchange_name}' 설정 완료")
            return channel
    
    def bind_queue_to_exchange(
        self,
        queue_name: str,
        exchange_name: str,
        routing_key: str = '',
        headers: dict = None
    ):
        """
        큐를 교환기에 바인딩
        
        Args:
            queue_name: 큐 이름
            exchange_name: 교환기 이름
            routing_key: 라우팅 키
            headers: 헤더 속성 (헤더 교환기용)
            
        Returns:
            bool: 바인딩 성공 여부
        """
        try:
            with self.client.connection_context() as channel:
                # 큐 선언
                channel.queue_declare(queue=queue_name, durable=True)
                
                # 바인딩 설정
                if headers:
                    channel.queue_bind(
                        queue=queue_name,
                        exchange=exchange_name,
                        arguments=headers
                    )
                else:
                    channel.queue_bind(
                        queue=queue_name,
                        exchange=exchange_name,
                        routing_key=routing_key
                    )
                
                self.logger.info(f"큐 '{queue_name}'를 교환기 '{exchange_name}'에 바인딩 완료")
                return True
                
        except Exception as e:
            self.logger.error(f"바인딩 실패: {e}")
            return False
    
    def publish_to_exchange(
        self,
        exchange_name: str,
        routing_key: str,
        message: dict,
        headers: dict = None
    ):
        """
        교환기에 메시지 발행
        
        Args:
            exchange_name: 교환기 이름
            routing_key: 라우팅 키
            message: 발행할 메시지
            headers: 헤더 속성 (헤더 교환기용)
            
        Returns:
            bool: 발행 성공 여부
        """
        properties = pika.BasicProperties(
            delivery_mode=2,  # 메시지 지속성
            content_type='application/json',
            headers=headers
        )
        
        try:
            with self.client.connection_context() as channel:
                channel.basic_publish(
                    exchange=exchange_name,
                    routing_key=routing_key,
                    body=json.dumps(message),
                    properties=properties
                )
                
                self.logger.info(f"교환기 '{exchange_name}'에 메시지 발행 성공")
                return True
                
        except Exception as e:
            self.logger.error(f"교환기 메시지 발행 실패: {e}")
            return False

# 교환기 패턴 예제
class ExchangePatterns:
    """RabbitMQ 교환기 패턴 구현 예제"""
    
    def __init__(self, client: RabbitMQClient):
        self.client = client
        self.exchange_manager = RabbitMQExchange(client)
        self.logger = logging.getLogger(__name__)
    
    def setup_pub_sub(self, exchange_name: str):
        """
        발행-구독 패턴 설정 (팬아웃 교환기 사용)
        
        Args:
            exchange_name: 교환기 이름
            
        Returns:
            list: 생성된 큐 이름 목록
        """
        # 팬아웃 교환기 설정
        self.exchange_manager.setup_fanout(exchange_name)
        
        # 여러 큐 생성 및 바인딩
        queues = []
        for i in range(3):
            queue_name = f"{exchange_name}_queue_{i}"
            with self.client.connection_context() as channel:
                result = channel.queue_declare(queue=queue_name, exclusive=True)
                queue_name = result.method.queue
                queues.append(queue_name)
                
                # 큐를 교환기에 바인딩
                channel.queue_bind(
                    exchange=exchange_name,
                    queue=queue_name
                )
                
        self.logger.info(f"발행-구독 패턴 설정 완료: {queues}")
        return queues
    
    def setup_routing(self, exchange_name: str, routing_keys: list):
        """
        라우팅 패턴 설정 (다이렉트 교환기 사용)
        
        Args:
            exchange_name: 교환기 이름
            routing_keys: 라우팅 키 목록
            
        Returns:
            dict: 라우팅 키별 큐 이름 매핑
        """
        # 다이렉트 교환기 설정
        self.exchange_manager.setup_direct(exchange_name)
        
        # 라우팅 키별 큐 생성 및 바인딩
        routing_map = {}
        for key in routing_keys:
            queue_name = f"{exchange_name}_{key}"
            with self.client.connection_context() as channel:
                result = channel.queue_declare(queue=queue_name, exclusive=True)
                queue_name = result.method.queue
                routing_map[key] = queue_name
                
                # 큐를 교환기에 바인딩
                channel.queue_bind(
                    exchange=exchange_name,
                    queue=queue_name,
                    routing_key=key
                )
        
        self.logger.info(f"라우팅 패턴 설정 완료: {routing_map}")
        return routing_map
    
    def setup_topics(self, exchange_name: str, topic_patterns: list):
        """
        토픽 패턴 설정 (토픽 교환기 사용)
        
        Args:
            exchange_name: 교환기 이름
            topic_patterns: 토픽 패턴 목록
            
        Returns:
            dict: 토픽 패턴별 큐 이름 매핑
        """
        # 토픽 교환기 설정
        self.exchange_manager.setup_topic(exchange_name)
        
        # 토픽 패턴별 큐 생성 및 바인딩
        topic_map = {}
        for pattern in topic_patterns:
            queue_name = f"{exchange_name}_{pattern.replace('.', '_').replace('*', 'star').replace('#', 'hash')}"
            with self.client.connection_context() as channel:
                result = channel.queue_declare(queue=queue_name, exclusive=True)
                queue_name = result.method.queue
                topic_map[pattern] = queue_name
                
                # 큐를 교환기에 바인딩
                channel.queue_bind(
                    exchange=exchange_name,
                    queue=queue_name,
                    routing_key=pattern
                )
        
        self.logger.info(f"토픽 패턴 설정 완료: {topic_map}")
        return topic_map

✅ 특징:

다양한 교환기 유형 지원 (팬아웃, 다이렉트, 토픽, 헤더)
유연한 메시지 라우팅 구성
바인딩을 통한 메시지 필터링
발행-구독 패턴 구현
라우팅 패턴 구현
토픽 기반 패턴 매칭
헤더 기반 메시지 라우팅
컨텍스트 매니저를 통한 자원 관리
로깅을 통한 교환기 작업 추적

4️⃣ 메시지 처리와 확인

메시지 처리는 RabbitMQ 시스템의 핵심 부분으로, 안정적인 처리와 오류 복구 메커니즘이 필수적이다. 적절한 확인(ack)과 거부(nack) 처리를 통해 메시지가 손실되지 않도록 보장할 수 있다.

class MessageHandler:
    """RabbitMQ 메시지 처리 및 확인 관리 클래스"""
    
    def __init__(self):
        self.logger = logging.getLogger(__name__)
    
    @staticmethod
    def process_message(
        channel,
        method,
        properties,
        body
    ):
        """
        메시지 처리 기본 핸들러
        
        Args:
            channel: RabbitMQ 채널
            method: 전달 메서드
            properties: 메시지 속성
            body: 메시지 본문
        """
        try:
            # 메시지 파싱
            message = json.loads(body)
            print(f"Received message: {message}")
            
            # 메시지 처리 로직 (예시)
            # 실제 구현에서는 비즈니스 로직에 맞게 처리
            processing_time = 0.5  # 메시지 처리에 걸리는 시간 (예시)
            time.sleep(processing_time)
            
            # 메시지 처리 성공 시 확인(ack)
            channel.basic_ack(delivery_tag=method.delivery_tag)
            print(f"처리 완료 및 확인(ack): {message.get('id', 'unknown')}")
            
        except json.JSONDecodeError as e:
            # JSON 파싱 오류 - 메시지 형식이 잘못된 경우
            print(f"JSON 파싱 오류: {e}")
            # 형식 오류는 재시도해도 동일하므로 재큐잉하지 않음
            channel.basic_reject(
                delivery_tag=method.delivery_tag,
                requeue=False
            )
        except Exception as e:
            # 기타 처리 오류 - 일시적인 문제일 수 있으므로 재큐잉
            print(f"메시지 처리 오류: {e}")
            channel.basic_nack(
                delivery_tag=method.delivery_tag,
                requeue=True
            )
    
    @staticmethod
    def process_with_retry(
        channel,
        method,
        properties,
        body,
        max_retries: int = 3,
        retry_exchange: str = None,
        retry_queue: str = None,
        dead_letter_exchange: str = None
    ):
        """
        재시도 로직이 포함된 메시지 처리 핸들러
        
        Args:
            channel: RabbitMQ 채널
            method: 전달 메서드
            properties: 메시지 속성
            body: 메시지 본문
            max_retries: 최대 재시도 횟수
            retry_exchange: 재시도 교환기 이름
            retry_queue: 재시도 큐 이름
            dead_letter_exchange: 데드레터 교환기 이름
        """
        try:
            # 헤더에서 재시도 횟수 확인
            headers = properties.headers or {}
            retry_count = headers.get('x-retry-count', 0)
            
            # 메시지 파싱
            message = json.loads(body)
            print(f"처리 중 (시도 {retry_count+1}/{max_retries+1}): {message}")
            
            # 메시지 처리 로직 (예시)
            # 실제 구현에서는 비즈니스 로직에 맞게 처리
            if random.random() < 0.3:  # 30% 확률로 오류 발생 (예시)
                raise Exception("랜덤 처리 오류 (테스트용)")
                
            # 메시지 처리 성공 시 확인(ack)
            channel.basic_ack(delivery_tag=method.delivery_tag)
            print(f"처리 완료: {message.get('id', 'unknown')}")
            
        except Exception as e:
            print(f"처리 오류: {e}")
            
            # 최대 재시도 횟수 초과 여부 확인
            if retry_count >= max_retries:
                print(f"최대 재시도 횟수 초과: {message.get('id', 'unknown')}")
                
                # 데드레터 교환기가 설정된 경우 해당 교환기로 전송
                if dead_letter_exchange:
                    print(f"데드레터 큐로 이동: {message.get('id', 'unknown')}")
                    properties.headers = headers
                    properties.headers['x-error'] = str(e)
                    properties.headers['x-failed-at'] = time.time()
                    
                    channel.basic_publish(
                        exchange=dead_letter_exchange,
                        routing_key='dead_letter',
                        body=body,
                        properties=properties
                    )
                
                # 원본 메시지 확인(ack)
                channel.basic_ack(delivery_tag=method.delivery_tag)
                
            else:
                # 재시도 큐로 다시 전송
                if retry_exchange and retry_queue:
                    print(f"재시도 큐로 이동: {message.get('id', 'unknown')}")
                    
                    # 재시도 정보 업데이트
                    headers['x-retry-count'] = retry_count + 1
                    headers['x-original-exchange'] = properties.headers.get('x-original-exchange', '')
                    headers['x-original-routing-key'] = properties.headers.get('x-original-routing-key', method.routing_key)
                    headers['x-last-retry-at'] = time.time()
                    
                    new_properties = pika.BasicProperties(
                        delivery_mode=2,
                        headers=headers
                    )
                    
                    # 재시도 큐로 발행
                    channel.basic_publish(
                        exchange=retry_exchange,
                        routing_key=retry_queue,
                        body=body,
                        properties=new_properties
                    )
                    
                    # 원본 메시지 확인(ack)
                    channel.basic_ack(delivery_tag=method.delivery_tag)
                    
                else:
                    # 재시도 설정이 없는 경우 원본 큐에 재큐잉
                    print(f"원본 큐에 재큐잉: {message.get('id', 'unknown')}")
                    channel.basic_nack(
                        delivery_tag=method.delivery_tag,
                        requeue=True
                    )
    
    @staticmethod
    def setup_retry_mechanism(
        channel,
        queue_name: str,
        retry_delay: int = 30000,  # 재시도 지연 시간 (밀리초)
        max_retries: int = 3
    ):
        """
        재시도 메커니즘 설정
        
        Args:
            channel: RabbitMQ 채널
            queue_name: 원본 큐 이름
            retry_delay: 재시도 지연 시간 (밀리초)
            max_retries: 최대 재시도 횟수
            
        Returns:
            tuple: (재시도 큐 이름, 데드레터 큐 이름)
        """
        # 데드레터 교환기 및 큐 설정
        dlx_name = f"{queue_name}.dlx"
        dl_queue_name = f"{queue_name}.deadletter"
        
        channel.exchange_declare(
            exchange=dlx_name,
            exchange_type='direct',
            durable=True
        )
        
        channel.queue_declare(
            queue=dl_queue_name,
            durable=True
        )
        
        channel.queue_bind(
            queue=dl_queue_name,
            exchange=dlx_name,
            routing_key='dead_letter'
        )
        
        # 재시도 교환기 및 큐 설정
        retry_exchange = f"{queue_name}.retry"
        retry_queue = f"{queue_name}.retry"
        
        channel.exchange_declare(
            exchange=retry_exchange,
            exchange_type='direct',
            durable=True
        )
        
        # 재시도 큐 설정 (TTL 적용)
        channel.queue_declare(
            queue=retry_queue,
            durable=True,
            arguments={
                'x-dead-letter-exchange': '',  # 기본 교환기로 설정
                'x-dead-letter-routing-key': queue_name,  # 원본 큐로 다시 라우팅
                'x-message-ttl': retry_delay  # 재시도 지연 시간
            }
        )
        
        channel.queue_bind(
            queue=retry_queue,
            exchange=retry_exchange,
            routing_key=retry_queue
        )
        
        # 원본 큐에 데드레터 교환기 설정
        channel.queue_declare(
            queue=queue_name,
            durable=True,
            arguments={
                'x-dead-letter-exchange': dlx_name,
                'x-dead-letter-routing-key': 'dead_letter'
            }
        )
        
        print(f"재시도 메커니즘 설정 완료: {queue_name}")
        return retry_queue, dl_queue_name

✅ 특징:

오류 처리 및 복구 메커니즘
재시도 로직 구현
메시지 응답 (ack, nack, reject)
데드레터 큐 설정 및 관리
지연 재시도 (TTL 활용)
진행 상태 추적 (헤더 사용)
최대 재시도 횟수 설정
오류 정보 기록
메시지 처리 상태 로깅

5️⃣ 모니터링과 관리

RabbitMQ는 강력한 모니터링 기능과 관리 도구를 제공하여 메시징 인프라의 상태를 추적하고 최적화할 수 있다. Python에서는 관리 API를 통해 이러한 기능에 프로그래매틱하게 접근할 수 있다.

import requests
import base64
import urllib.parse
import time
from typing import Dict, List, Any, Optional

class RabbitMQMonitor:
    """RabbitMQ 모니터링 및 관리를 위한 클래스"""
    
    def __init__(
        self,
        host: str = 'localhost',
        port: int = 15672,
        username: str = 'guest',
        password: str = 'guest',
        protocol: str = 'http'
    ):
        """
        RabbitMQ 모니터 초기화
        
        Args:
            host: RabbitMQ 관리 인터페이스 호스트
            port: RabbitMQ 관리 인터페이스 포트 (기본 15672)
            username: 관리자 사용자 이름
            password: 관리자 비밀번호
            protocol: 통신 프로토콜 (http 또는 https)
        """
        self.base_url = f"{protocol}://{host}:{port}/api"
        self.auth = (username, password)
        self.logger = logging.getLogger(__name__)
    
    def _request(
        self,
        method: str,
        endpoint: str,
        data: dict = None,
        params: dict = None
    ) -> dict:
        """
        RabbitMQ API 요청 메서드
        
        Args:
            method: HTTP 메서드 (GET, POST, PUT, DELETE)
            endpoint: API 엔드포인트
            data: 요청 데이터 (있는 경우)
            params: URL 파라미터 (있는 경우)
            
        Returns:
            dict: API 응답
        """
        url = f"{self.base_url}/{endpoint}"
        
        try:
            response = requests.request(
                method=method,
                url=url,
                auth=self.auth,
                json=data,
                params=params
            )
            
            response.raise_for_status()
            
            if response.content:
                return response.json()
            return {}
            
        except requests.exceptions.HTTPError as e:
            self.logger.error(f"API 요청 오류 ({method} {url}): {e}")
            if response.content:
                self.logger.error(f"응답: {response.text}")
            raise
        except Exception as e:
            self.logger.error(f"API 요청 중 예외 발생: {e}")
            raise
    
    def get_overview(self) -> dict:
        """
        RabbitMQ 서버 개요 정보 조회
        
        Returns:
            dict: 서버 개요 정보
        """
        return self._request('GET', 'overview')
    
    def get_nodes(self) -> List[dict]:
        """
        RabbitMQ 노드 목록 조회
        
        Returns:
            List[dict]: 노드 정보 목록
        """
        return self._request('GET', 'nodes')
    
    def get_queues(self, vhost: str = None) -> List[dict]:
        """
        큐 목록 조회
        
        Args:
            vhost: 가상 호스트 (지정하지 않으면 모든 가상 호스트)
            
        Returns:
            List[dict]: 큐 정보 목록
        """
        endpoint = 'queues'
        if vhost:
            endpoint = f"queues/{urllib.parse.quote_plus(vhost)}"
        return self._request('GET', endpoint)
    
    def get_queue(self, vhost: str, queue_name: str) -> dict:
        """
        특정 큐 정보 조회
        
        Args:
            vhost: 가상 호스트
            queue_name: 큐 이름
            
        Returns:
            dict: 큐 정보
        """
        endpoint = f"queues/{urllib.parse.quote_plus(vhost)}/{urllib.parse.quote_plus(queue_name)}"
        return self._request('GET', endpoint)
    
    def get_exchanges(self, vhost: str = None) -> List[dict]:
        """
        교환기 목록 조회
        
        Args:
            vhost: 가상 호스트 (지정하지 않으면 모든 가상 호스트)
            
        Returns:
            List[dict]: 교환기 정보 목록
        """
        endpoint = 'exchanges'
        if vhost:
            endpoint = f"exchanges/{urllib.parse.quote_plus(vhost)}"
        return self._request('GET', endpoint)
    
    def get_bindings(self, vhost: str = None) -> List[dict]:
        """
        바인딩 목록 조회
        
        Args:
            vhost: 가상 호스트 (지정하지 않으면 모든 가상 호스트)
            
        Returns:
            List[dict]: 바인딩 정보 목록
        """
        endpoint = 'bindings'
        if vhost:
            endpoint = f"bindings/{urllib.parse.quote_plus(vhost)}"
        return self._request('GET', endpoint)
    
    def get_connections(self) -> List[dict]:
        """
        연결 목록 조회
        
        Returns:
            List[dict]: 연결 정보 목록
        """
        return self._request('GET', 'connections')
    
    def get_channels(self) -> List[dict]:
        """
        채널 목록 조회
        
        Returns:
            List[dict]: 채널 정보 목록
        """
        return self._request('GET', 'channels')
    
    def purge_queue(self, vhost: str, queue_name: str) -> bool:
        """
        큐 내용 제거
        
        Args:
            vhost: 가상 호스트
            queue_name: 큐 이름
            
        Returns:
            bool: 성공 여부
        """
        try:
            endpoint = f"queues/{urllib.parse.quote_plus(vhost)}/{urllib.parse.quote_plus(queue_name)}/contents"
            self._request('DELETE', endpoint)
            self.logger.info(f"큐 '{queue_name}' 내용 제거 완료")
            return True
        except Exception as e:
            self.logger.error(f"큐 내용 제거 실패: {e}")
            return False
    
    def create_user(self, username: str, password: str, tags: str = "management") -> bool:
        """
        사용자 생성
        
        Args:
            username: 사용자 이름
            password: 비밀번호
            tags: 사용자 태그 (쉼표로 구분)
            
        Returns:
            bool: 성공 여부
        """
        try:
            data = {
                "password": password,
                "tags": tags
            }
            self._request('PUT', f"users/{urllib.parse.quote_plus(username)}", data)
            self.logger.info(f"사용자 '{username}' 생성 완료")
            return True
        except Exception as e:
            self.logger.error(f"사용자 생성 실패: {e}")
            return False
    
    def set_permissions(
        self,
        vhost: str,
        username: str,
        configure: str = ".*",
        write: str = ".*",
        read: str = ".*"
    ) -> bool:
        """
        사용자 권한 설정
        
        Args:
            vhost: 가상 호스트
            username: 사용자 이름
            configure: 설정 권한 정규식
            write: 쓰기 권한 정규식
            read: 읽기 권한 정규식
            
        Returns:
            bool: 성공 여부
        """
        try:
            data = {
                "configure": configure,
                "write": write,
                "read": read
            }
            endpoint = f"permissions/{urllib.parse.quote_plus(vhost)}/{urllib.parse.quote_plus(username)}"
            self._request('PUT', endpoint, data)
            self.logger.info(f"사용자 '{username}'의 '{vhost}' 권한 설정 완료")
            return True
        except Exception as e:
            self.logger.error(f"권한 설정 실패: {e}")
            return False
    
    def get_queue_stats(self, vhost: str, queue_name: str) -> dict:
        """
        큐 통계 정보 조회
        
        Args:
            vhost: 가상 호스트
            queue_name: 큐 이름
            
        Returns:
            dict: 큐 통계 정보
        """
        try:
            queue_info = self.get_queue(vhost, queue_name)
            
            stats = {
                "name": queue_info.get("name"),
                "messages": queue_info.get("messages", 0),
                "messages_ready": queue_info.get("messages_ready", 0),
                "messages_unacknowledged": queue_info.get("messages_unacknowledged", 0),
                "consumers": queue_info.get("consumers", 0),
                "message_stats": queue_info.get("message_stats", {}),
                "memory": queue_info.get("memory", 0),
                "state": queue_info.get("state", "unknown"),
                "node": queue_info.get("node", "unknown")
            }
            
            return stats
        except Exception as e:
            self.logger.error(f"큐 통계 조회 실패: {e}")
            return {}
    
    def monitor_queues(
        self,
        vhost: str = '/',
        interval: int = 5,
        limit: int = 10,
        output: bool = True
    ):
        """
        큐 모니터링 실행
        
        Args:
            vhost: 가상 호스트
            interval: 모니터링 간격(초)
            limit: 모니터링 횟수 (0=무제한)
            output: 출력 여부
        """
        count = 0
        try:
            while limit == 0 or count < limit:
                queues = self.get_queues(vhost)
                
                if output:
                    print(f"\n===== 큐 모니터링 ({time.strftime('%Y-%m-%d %H:%M:%S')}) =====")
                    print(f"{'이름':30} {'메시지':10} {'준비':10} {'미확인':10} {'소비자':5}")
                    print("-" * 70)
                
                for q in queues:
                    name = q.get("name", "")
                    messages = q.get("messages", 0)
                    messages_ready = q.get("messages_ready", 0)
                    messages_unack = q.get("messages_unacknowledged", 0)
                    consumers = q.get("consumers", 0)
                    
                    if output:
                        print(f"{name[:30]:30} {messages:10} {messages_ready:10} {messages_unack:10} {consumers:5}")
                
                count += 1
                if limit == 0 or count < limit:
                    time.sleep(interval)
                    
        except KeyboardInterrupt:
            print("\n모니터링 중단됨")
        except Exception as e:
            self.logger.error(f"모니터링 오류: {e}")

# 사용 예제
if __name__ == "__main__":
    # 로깅 설정
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    
    # RabbitMQ 모니터 생성
    monitor = RabbitMQMonitor(
        host='localhost',
        port=15672,
        username='guest',
        password='guest'
    )
    
    # 서버 개요 정보 조회
    overview = monitor.get_overview()
    print(f"RabbitMQ 버전: {overview.get('rabbitmq_version')}")
    print(f"Erlang 버전: {overview.get('erlang_version')}")
    print(f"노드: {overview.get('node')}")
    
    # 큐 목록 조회
    queues = monitor.get_queues('/')
    print(f"\n큐 목록 ({len(queues)}개):")
    for q in queues:
        print(f"- {q.get('name')}: {q.get('messages')} 메시지")
    
    # 큐 모니터링 (5초 간격, 3회)
    print("\n큐 모니터링 시작...")
    monitor.monitor_queues(interval=5, limit=3)

✅ 특징:

RabbitMQ 관리 API 활용
서버 상태 모니터링
큐와 교환기 정보 조회
리소스 사용량 추적
메시지 통계 분석
사용자 관리 및 권한 설정
실시간 모니터링 기능
대시보드 정보 프로그래매틱 접근
REST API 기반 통합
모니터링 자동화

주요 팁

✅ 모범 사례:

연결 관리: 장기 실행 애플리케이션은 연결 풀링과 자동 재연결 메커니즘을 구현하여 안정성을 높인다.
메시지 지속성: 중요 메시지는 지속성을 설정하고 큐도 내구성 있게 구성하여 서버 재시작 시에도 데이터 손실을 방지한다.
QoS 설정: prefetch_count 값을 적절히 조정하여 워커 간 부하 균형을 유지하고 메모리 사용량을 관리한다.
교환기 선택: 메시징 패턴에 맞는 적절한 교환기 유형을 선택하여 라우팅 효율성을 극대화한다.
재시도 메커니즘: 데드레터 큐와 TTL을 활용한 지연 재시도 전략으로 일시적 오류를 우아하게 처리한다.
모니터링: 큐 길이, 메시지 처리율, 메모리 사용량을 정기적으로 모니터링하여 병목 현상을 감지한다.
성능 최적화: 배치 처리, 메시지 압축, 채널 재사용으로 처리량을 최적화한다.
보안 설정: 강력한 암호와 TLS 연결로 메시지 전송을 보호하고, VHOST와 사용자 권한을 세분화하여 접근을 제한한다.
클러스터링: 고가용성을 위해 RabbitMQ 클러스터를 구성하고 적절한 정책을 설정한다.
백업 전략: 정기적인 구성 백업과 메시지 스냅샷으로 재해 복구 대비책을 마련한다.
로깅 전략: 메시지 처리 과정에서 충분한 컨텍스트를 로깅하여 문제 해결과 감사를 용이하게 한다.

KR_RabbitMQ - somaz94/python-study GitHub Wiki

Python RabbitMQ 개념 정리

1️⃣ 기본 연결

2️⃣ 기본 메시징

3️⃣ 고급 메시징 패턴

4️⃣ 메시지 처리와 확인

5️⃣ 모니터링과 관리

주요 팁

⚠️ GitHub.com Fallback ⚠️

KR_RabbitMQ - somaz94/python-study GitHub Wiki

Python RabbitMQ 개념 정리

1️⃣ 기본 연결

2️⃣ 기본 메시징

3️⃣ 고급 메시징 패턴

4️⃣ 메시지 처리와 확인

5️⃣ 모니터링과 관리

주요 팁

⚠️ **GitHub.com Fallback** ⚠️

⚠️ GitHub.com Fallback ⚠️