1
2
3
4
5
6
7

| 구분 | 항목                | 설명                              | 기술적 근거                       | 실무 효과             |
|------|---------------------|-----------------------------------|-----------------------------------|----------------------|
| 장점 | 표준화/호환성        | AMQP 기반, 다양한 언어·플랫폼 지원 | AMQP 구현, 오픈 API               | 다양한 환경에 도입    |
| 장점 | 유연한 라우팅        | 다양한 Exchange 타입               | Direct/Topic/Fanout/Headers 지원  | 복잡시나리오 구현     |
| 장점 | 장애 내구성/복구      | 메시지 내구성, 미러 큐             | Durable Queue/HA 기능              | 장애 시 데이터 보호   |
| 장점 | 관리 편의성          | 대시보드, 모니터링, 확장 도구       | Management Plugin, CLI 지원        | 운영 효율성 증대      |
| 장점 | 실시간 비동기 처리    | 소비/생산자 분리, 높은 처리율       | 비동기 기반 연결                   | 병렬·분산 처리 효율   |

1
2
3
4
5
6

| 구분 | 항목            | 설명                       | 해결책                                 | 대안 기술           |
|------|-----------------|----------------------------|----------------------------------------|---------------------|
| 단점 | 높은 지연        | 메시지 대량·순차 처리 시    | 노드 확장, 프리페칭 등 설정 최적화      | Kafka, Redis        |
| 단점 | 순차처리 한계    | Pub/Sub 대규모 분산에 취약  | Exchange·큐 구조 최적화                | Kafka               |
| 단점 | 클러스터 복잡도  | HA, 미러큐 설정 어렵고 복잡 | 자동화/모니터링 도구 활용               | AWS SQS, Kafka      |
| 단점 | 운영 지표 부족   | 심층 모니터링 한계          | 외부 모니터링 시스템 연동               | Prometheus          |

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# RabbitMQ 메시지 전송 예제 (Python, pika 라이브러리 사용)
import pika

def send_message():
    """
    RabbitMQ Producer 역할 구현 예시
    - 메시지 발행, 큐에 전송, 신뢰성 보장
    """
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='hello')
    channel.basic_publish(exchange='',
                         routing_key='hello',
                         body='Hello RabbitMQ!')
    connection.close()

1
2
3
4
5
6

| 분류 기준   | 유형           | 특징                | 적용 시나리오                   |
|-------------|---------------|---------------------|---------------------------------|
| Exchange 타입| Direct         | 라우팅 키 일치 시 전달 | 단순 큐 처리                    |
| Exchange 타입| Topic          | 라우팅 패턴 기반 처리  | 이벤트/로그 분류                |
| Exchange 타입| Fanout         | 모든 큐에 브로드캐스트 | 대량 메시지 알림                |
| Exchange 타입| Headers        | 헤더 기반 라우팅      | 동적 메시지 분류                |

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

# 주문 서비스(Producer)가 메시지 발행
import pika

def send_order_complete(order_id):
    """
    주문 완료 이벤트 송신
    - RabbitMQ 큐에 메시지 전달 (비동기, 신뢰성 보장)
    """
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='order_notify')
    channel.basic_publish(exchange='',
                         routing_key='order_notify',
                         body=f'Order Complete: {order_id}')
    connection.close()

# 알림 서비스(Consumer)가 메시지 수신
def receive_order_notify():
    """
    RabbitMQ 큐에서 메시지 소비
    - 소비자가 메시지 수신 후 알림 처리
    """
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='order_notify')

    def callback(ch, method, properties, body):
        print("알림 수신:", body.decode())

    channel.basic_consume(queue='order_notify',
                          on_message_callback=callback,
                          auto_ack=True)
    print('알림 서비스 대기중...')
    channel.start_consuming()

1
2
3
4
5
6
7
8
9

# RabbitMQ 미러 큐 설정 예시
# 장애 복구와 데이터 유실 방지 목적
policies:
  ha-all:
    pattern: ".*"
    definition:
      ha-mode: all
      ha-sync-mode: automatic
    priority: 0

1
2
3
4
5
6

| 항목         | 주요 위험 | 권장사항                           | 효과             |
|--------------|----------|-------------------------------------|------------------|
| 클러스터링   | 노드 장애| 미러 큐/HA 전략 구성                | 데이터 유실 방지 |
| 성능최적화   | 메시지 적체| Prefetch/Parallel Consumer 설정      | 처리율 향상      |
| 보안         | 무단 접근| TLS·인증·권한 분리                  | 침해 위험 최소화 |
| 모니터링     | 장애 지연| 외부 모니터링 시스템 연동            | 빠른 대응        |

1
2
3
4
5
6

| 구분     | 전략           | 설명                 | 권장 환경     |
|----------|----------------|----------------------|--------------|
| 스케일링 | 노드·미러 확장  | 클러스터로 노드 확장  | TPS 증가      |
| 큐 설계  | 큐 분리/분산    | 주요 이벤트별 분산    | 대량 이벤트   |
| ACK 전략 | msg ack 조정    | 소비자 수동/자동 확인 | 저지연 환경   |
| 연결 관리| 연결 재사용     | Channel Pool 운용     | 고부하 상황   |

1
2
3
4
5
6
7
8
9

| 카테고리 | Phase | 항목                   | 중요도 | 학습 목표                    | 실무 연관성 | 설명                          |
|----------|-------|------------------------|--------|-----------------------------|-------------|-------------------------------|
| 기초     | 1     | 개념/등장배경/필요성   | 필수   | RabbitMQ 이해 및 맥락파악     | 높음        | 기본 메시지 브로커 동작 구조   |
| 핵심     | 2     | 아키텍트/구성/원리     | 필수   | 시스템 구조 및 동작원리 습득  | 높음        | 네트워크 및 큐 라우팅 이해    |
| 분석     | 3     | 장/단점, 트레이드오프  | 필수   | 기술선정 합리성 증대          | 높음        | 실무 적용시 고려포인트 정리   |
| 구현     | 4     | 배포/설정/프레임워크   | 권장   | 실제 설치·운영 방법 숙달      | 중간        | CLI/플러그인/클러스터링 실습  |
| 실습     | 5     | 실무 예제/도입사례     | 권장   | 현장 적용 경험 확장           | 중간        | Python 등 코드로 적용         |
| 운영     | 6     | 보안/관측성/최적화     | 필수   | 운영 효율화, 장애 대응 강화   | 높음        | 대시보드·모니터링 실습        |
| 고급     | 7     | 트렌드/연계/도전과제   | 선택   | 미래 기술 리더십 확보         | 낮음        | 클라우드·AI 연동 등 전망 정리 |

1
2
3
4
5
6
7
8
9

| 카테고리 | 용어           | 정의                                     | 관련 개념          | 실무 활용          |
|----------|----------------|------------------------------------------|--------------------|--------------------|
| 핵심     | 메시지 브로커   | 시스템 간 메시지 중계, 데이터 흐름 관리   | MOM, AMQP          | 이벤트/데이터 처럼 |
| 핵심     | Exchange       | 메시지 라우팅, 다양한 분기 지원           | 라우팅 키, 타입     | 유연한 대기열 구성 |
| 핵심     | Queue          | 메시지 저장소, 소비 대기                  | Durable/Transient  | 장애 복구 핵심     |
| 구현     | Producer       | 메시지 발행 역할                         | API, 이벤트 소스    | 이벤트 소스 구현   |
| 구현     | Consumer       | 메시지 수신 및 처리 역할                  | ACK, 소비자 그룹    | 백엔드 서비스      |
| 운영     | 미러 큐        | 메시지 복제, 장애 복구 기능               | HA, 클러스터        | 데이터 유실 방지   |
| 운영     | Dead Letter    | 실패/에러 메시지 분리 저장                | 장애 탐지/복구      | 실시간 에러 추적   |

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

# RabbitMQ 연결 및 채널 관리 예시
import pika

# TCP 연결 생성 (비용이 높은 작업)
connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost')
)

# 채널 생성 (가벼운 논리적 연결)
channel = connection.channel()

# 채널 내에서 여러 작업 수행
channel.queue_declare(queue='task_queue', durable=True)
channel.basic_publish(
    exchange='',
    routing_key='task_queue', 
    body='Hello World!',
    properties=pika.BasicProperties(delivery_mode=2)  # 메시지 지속성
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# Direct Exchange - 정확한 키 매칭
channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
channel.basic_publish(
    exchange='direct_logs',
    routing_key='error',  # 라우팅 키 지정
    body='Error message'
)

# Topic Exchange - 패턴 매칭
channel.exchange_declare(exchange='topic_logs', exchange_type='topic')
channel.basic_publish(
    exchange='topic_logs', 
    routing_key='user.signup.email',  # 계층적 키
    body='User signup notification'
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

def callback(ch, method, properties, body):
    print(f"Processing: {body}")
    
    # 비즈니스 로직 처리
    try:
        process_message(body)
        # 처리 성공 시 ACK
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as e:
        # 처리 실패 시 NACK (재시도 또는 DLQ 이동)
        ch.basic_nack(
            delivery_tag=method.delivery_tag, 
            requeue=False
        )

channel.basic_consume(
    queue='task_queue',
    on_message_callback=callback,
    auto_ack=False  # 수동 ACK 모드
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

# Producer (작업 생성자)
import pika
import json

def send_task(task_data):
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    
    # 작업 큐 선언 (지속성 큐)
    channel.queue_declare(queue='task_queue', durable=True)
    
    # 작업 메시지 발송 (지속성 메시지)
    channel.basic_publish(
        exchange='',
        routing_key='task_queue',
        body=json.dumps(task_data),
        properties=pika.BasicProperties(
            delivery_mode=2,  # 메시지를 디스크에 저장
        )
    )
    connection.close()

# Consumer (작업 처리자)
def process_task(ch, method, properties, body):
    """
    이 함수는 RabbitMQ의 Work Queue 패턴에서 작업을 처리하는 핵심 부분
    - body: 처리할 작업 데이터 (JSON 형태)
    - method.delivery_tag: 메시지 확인응답을 위한 식별자
    """
    task_data = json.loads(body)
    print(f"Processing task: {task_data}")
    
    try:
        # 실제 비즈니스 로직 처리
        result = expensive_operation(task_data)
        
        # 작업 완료 후 확인응답 (ACK)
        ch.basic_ack(delivery_tag=method.delivery_tag)
        print(f"Task completed: {result}")
        
    except Exception as e:
        # 처리 실패 시 거부 (다른 컨슈머가 재처리)
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
        print(f"Task failed: {e}")

def start_worker():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    
    # QoS 설정: 한 번에 하나의 메시지만 처리
    channel.basic_qos(prefetch_count=1)
    
    # 메시지 소비 설정 (수동 ACK)
    channel.basic_consume(
        queue='task_queue',
        on_message_callback=process_task,
        auto_ack=False  # 수동 확인응답 사용
    )
    
    print("Worker started. Waiting for messages...")
    channel.start_consuming()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

# Publisher (이벤트 발행자)
def publish_event(event_type, event_data):
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    
    # Fanout Exchange 선언 (모든 큐에 브로드캐스트)
    channel.exchange_declare(exchange='events', exchange_type='fanout')
    
    event_message = {
        'type': event_type,
        'data': event_data,
        'timestamp': datetime.now().isoformat()
    }
    
    # 모든 바인딩된 큐로 메시지 발송
    channel.basic_publish(
        exchange='events',
        routing_key='',  # Fanout에서는 라우팅 키 무시됨
        body=json.dumps(event_message)
    )
    
    print(f"Event published: {event_type}")
    connection.close()

# Subscriber (이벤트 구독자)
def handle_user_service_events(ch, method, properties, body):
    """
    사용자 서비스에서 관심있는 이벤트만 처리
    - RabbitMQ의 Fanout Exchange를 통해 모든 서비스가 동일한 이벤트를 수신
    - 각 서비스는 자신이 처리해야 할 이벤트만 선별적으로 처리
    """
    event = json.loads(body)
    
    if event['type'] in ['user_registered', 'user_updated']:
        print(f"User service processing: {event['type']}")
        # 사용자 관련 비즈니스 로직 처리
        update_user_profile(event['data'])
    
    ch.basic_ack(delivery_tag=method.delivery_tag)

def start_subscriber(service_name):
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    
    channel.exchange_declare(exchange='events', exchange_type='fanout')
    
    # 서비스별 고유한 큐 생성
    result = channel.queue_declare(queue=f'{service_name}_events', exclusive=True)
    queue_name = result.method.queue
    
    # Exchange와 큐 바인딩
    channel.queue_bind(exchange='events', queue=queue_name)
    
    channel.basic_consume(
        queue=queue_name,
        on_message_callback=handle_user_service_events,
        auto_ack=False
    )
    
    print(f"{service_name} subscriber started...")
    channel.start_consuming()

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117

# RPC Server (원격 프로시저 제공자)
import uuid

def fibonacci(n):
    """실제 비즈니스 로직 (예: 피보나치 수열 계산)"""
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

def handle_rpc_request(ch, method, properties, body):
    """
    RPC 요청을 처리하고 응답을 반환하는 핵심 함수
    - properties.reply_to: 클라이언트의 응답 큐 이름
    - properties.correlation_id: 요청-응답 매칭을 위한 고유 ID
    """
    n = int(body)
    print(f"Computing fibonacci({n})")
    
    try:
        result = fibonacci(n)
        
        # 응답을 클라이언트의 응답 큐로 전송
        ch.basic_publish(
            exchange='',
            routing_key=properties.reply_to,
            properties=pika.BasicProperties(
                correlation_id=properties.correlation_id  # 요청 ID 매칭
            ),
            body=str(result)
        )
        
    except Exception as e:
        # 에러 응답 전송
        ch.basic_publish(
            exchange='',
            routing_key=properties.reply_to,
            properties=pika.BasicProperties(
                correlation_id=properties.correlation_id
            ),
            body=f"ERROR: {str(e)}"
        )
    
    ch.basic_ack(delivery_tag=method.delivery_tag)

def start_rpc_server():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    
    # RPC 요청 큐 선언
    channel.queue_declare(queue='rpc_queue')
    channel.basic_qos(prefetch_count=1)
    
    channel.basic_consume(
        queue='rpc_queue',
        on_message_callback=handle_rpc_request,
        auto_ack=False
    )
    
    print("RPC Server started. Waiting for requests...")
    channel.start_consuming()

# RPC Client (원격 프로시저 호출자)
class FibonacciRpcClient:
    def __init__(self):
        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.correlation_id = None
    
    def on_response(self, ch, method, properties, body):
        """
        RPC 응답을 처리하는 콜백 함수
        - correlation_id를 통해 요청-응답 매칭
        """
        if self.correlation_id == properties.correlation_id:
            self.response = body
    
    def call(self, n):
        """RPC 호출 메서드"""
        self.response = None
        self.correlation_id = str(uuid.uuid4())
        
        # RPC 요청 전송
        self.channel.basic_publish(
            exchange='',
            routing_key='rpc_queue',
            properties=pika.BasicProperties(
                reply_to=self.callback_queue,
                correlation_id=self.correlation_id,
            ),
            body=str(n)
        )
        
        # 응답 대기
        while self.response is None:
            self.connection.process_data_events()
        
        return int(self.response)

# 사용 예시
rpc_client = FibonacciRpcClient()
result = rpc_client.call(30)
print(f"fibonacci(30) = {result}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

// Spring AMQP를 사용한 RabbitMQ 통합
@Component
public class MessageProducer {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    // Direct Exchange를 통한 메시지 발송
    public void sendOrderEvent(Order order) {
        rabbitTemplate.convertAndSend(
            "order.exchange",     // Exchange 이름
            "order.created",      // Routing key
            order                 // 메시지 객체 (자동 JSON 변환)
        );
    }
}

@RabbitListener(queues = "order.processing.queue")
public void handleOrderProcessing(Order order) {
    /**
     * Spring AMQP의 어노테이션 기반 메시지 처리
     * - 자동 역직렬화: JSON → Order 객체 변환
     * - 자동 ACK: 메서드 성공 완료 시 자동 확인응답
     * - 예외 처리: 예외 발생 시 자동 재시도 또는 DLQ 이동
     */
    System.out.println("Processing order: " + order.getId());
    // 주문 처리 비즈니스 로직
    processOrder(order);
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

// amqplib를 사용한 Node.js 통합
const amqp = require('amqplib');

class EventBus {
    constructor() {
        this.connection = null;
        this.channel = null;
    }
    
    async connect() {
        this.connection = await amqp.connect('amqp://localhost');
        this.channel = await this.connection.createChannel();
        
        // Topic Exchange 선언 (이벤트 기반 통신)
        await this.channel.assertExchange('events', 'topic', {durable: true});
    }
    
    // 이벤트 발행
    async publish(routingKey, event) {
        /**
         * Node.js에서 RabbitMQ 이벤트 발행
         * - routingKey: 'user.created', 'order.updated' 등 계층적 키
         * - event: 발행할 이벤트 데이터 객체
         * - 자동 JSON 직렬화 및 UTF-8 인코딩
         */
        const message = Buffer.from(JSON.stringify(event));
        return this.channel.publish('events', routingKey, message, {
            persistent: true,  // 메시지 지속성
            timestamp: Date.now()
        });
    }
    
    // 이벤트 구독
    async subscribe(pattern, handler) {
        // 서비스별 고유 큐 생성
        const queue = await this.channel.assertQueue('', {exclusive: true});
        
        // 패턴 기반 바인딩 (예: 'user.*', 'order.#')
        await this.channel.bindQueue(queue.queue, 'events', pattern);
        
        // 메시지 소비 설정
        await this.channel.consume(queue.queue, async (msg) => {
            if (msg) {
                const event = JSON.parse(msg.content.toString());
                await handler(event);
                this.channel.ack(msg);  // 처리 완료 확인
            }
        });
    }
}

// 사용 예시
const eventBus = new EventBus();
await eventBus.connect();

// 사용자 이벤트 구독
await eventBus.subscribe('user.*', (event) => {
    console.log('User event received:', event);
});

// 이벤트 발행
await eventBus.publish('user.created', {
    userId: 12345,
    email: 'user@example.com'
});

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

# docker-compose.yml - 모니터링 스택 구성
version: '3.8'
services:
  rabbitmq:
    image: rabbitmq:4-management
    ports:
      - "5672:5672"
      - "15672:15672"
      - "15692:15692"  # Prometheus 메트릭 포트
    environment:
      - RABBITMQ_DEFAULT_USER=admin
      - RABBITMQ_DEFAULT_PASS=secret
    plugins:
      - rabbitmq_management
      - rabbitmq_prometheus  # Prometheus 플러그인 활성화
    volumes:
      - rabbitmq_data:/var/lib/rabbitmq
  
  prometheus:
    image: prom/prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
  
  grafana:
    image: grafana/grafana
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin
    volumes:
      - grafana_data:/var/lib/grafana

volumes:
  rabbitmq_data:
  grafana_data:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

# 관리 API를 통한 자동화 스크립트
import requests
import json

class RabbitMQManager:
    def __init__(self, host='localhost', port=15672, username='admin', password='admin'):
        self.base_url = f'http://{host}:{port}/api'
        self.auth = (username, password)
    
    def get_cluster_status(self):
        """클러스터 상태 조회"""
        response = requests.get(f'{self.base_url}/nodes', auth=self.auth)
        return response.json()
    
    def create_queue(self, vhost, queue_name, queue_type='classic'):
        """
        큐 생성 자동화
        - vhost: 가상 호스트 이름
        - queue_name: 생성할 큐 이름  
        - queue_type: 큐 타입 (classic, quorum, stream)
        """
        queue_config = {
            'durable': True,
            'arguments': {
                'x-queue-type': queue_type
            }
        }
        
        response = requests.put(
            f'{self.base_url}/queues/{vhost}/{queue_name}',
            auth=self.auth,
            headers={'Content-Type': 'application/json'},
            data=json.dumps(queue_config)
        )
        return response.status_code == 201
    
    def get_queue_metrics(self, vhost, queue_name):
        """큐 메트릭 조회"""
        response = requests.get(
            f'{self.base_url}/queues/{vhost}/{queue_name}',
            auth=self.auth
        )
        
        if response.status_code == 200:
            data = response.json()
            return {
                'messages': data.get('messages', 0),
                'consumers': data.get('consumers', 0),
                'message_rate': data.get('message_stats', {}).get('publish_details', {}).get('rate', 0)
            }
        return None

# 사용 예시
manager = RabbitMQManager()

# 큐 생성
manager.create_queue('/', 'orders', 'quorum')

# 메트릭 모니터링
metrics = manager.get_queue_metrics('/', 'orders')
print(f"Queue metrics: {metrics}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# AMQP 1.0 클라이언트 예시 (Azure Service Bus 호환)
from azure.servicebus import ServiceBusClient, ServiceBusMessage

# RabbitMQ 4.0의 AMQP 1.0 지원으로 Azure Service Bus와 호환 가능
client = ServiceBusClient.from_connection_string(
    "amqp://localhost:5672"  # RabbitMQ 4.0 AMQP 1.0 엔드포인트
)

# 표준 AMQP 1.0 메시지 발송
with client.get_queue_sender("test_queue") as sender:
    message = ServiceBusMessage("Hello from AMQP 1.0")
    sender.send_messages(message)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

# rabbitmq.conf - 보안 설정
listeners.ssl.default = 5671
ssl_options.certfile = /path/to/server.crt
ssl_options.keyfile = /path/to/server.key
ssl_options.cacertfile = /path/to/ca.crt
ssl_options.verify = verify_peer
ssl_options.versions.1 = tlsv1.3
ssl_options.ciphers.1 = ECDHE-ECDSA-AES256-GCM-SHA384

# Management UI HTTPS
management.ssl.port = 15671
management.ssl.certfile = /path/to/server.crt
management.ssl.keyfile = /path/to/server.key

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# LDAP 인증 설정
auth_backends.1 = ldap
auth_backends.2 = internal

auth_ldap.servers.1 = ldap.company.com
auth_ldap.use_ssl = true
auth_ldap.port = 636
auth_ldap.user_dn_pattern = cn=${username},ou=Users,dc=company,dc=com

# 권한 매핑
auth_ldap.vhost_access_query = {constant, true}
auth_ldap.resource_access_query = {for, [{permission, configure, {in_group, "cn=admin,ou=Groups,dc=company,dc=com"}},
                                         {permission, write, {constant, true}},
                                         {permission, read, {constant, true}}]}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139

# shared/event_bus.py - 공통 이벤트 버스 클래스
import pika
import json
import logging
from typing import Callable, Dict, Any
from datetime import datetime

class EventBus:
    """
    RabbitMQ 기반 이벤트 버스 클래스
    - 서비스 간 이벤트 발행/구독을 위한 공통 인터페이스 제공
    - Topic Exchange를 통한 유연한 이벤트 라우팅
    """
    
    def __init__(self, connection_url: str = 'amqp://localhost'):
        self.connection_url = connection_url
        self.connection = None
        self.channel = None
        self.logger = logging.getLogger(__name__)
    
    async def connect(self):
        """RabbitMQ 서버 연결 및 채널 설정"""
        try:
            self.connection = pika.BlockingConnection(
                pika.URLParameters(self.connection_url)
            )
            self.channel = self.connection.channel()
            
            # Topic Exchange 선언 (이벤트 기반 라우팅)
            self.channel.exchange_declare(
                exchange='ecommerce.events',
                exchange_type='topic',
                durable=True
            )
            
            self.logger.info("Connected to RabbitMQ")
        except Exception as e:
            self.logger.error(f"Failed to connect to RabbitMQ: {e}")
            raise
    
    def publish_event(self, routing_key: str, event_data: Dict[Any, Any]):
        """
        이벤트 발행 메서드
        - routing_key: 'order.created', 'payment.completed' 등 계층적 이벤트 키
        - event_data: 이벤트와 함께 전달할 데이터
        """
        try:
            event_message = {
                'event_id': str(uuid.uuid4()),
                'event_type': routing_key,
                'timestamp': datetime.utcnow().isoformat(),
                'data': event_data
            }
            
            # Topic Exchange로 이벤트 발행
            self.channel.basic_publish(
                exchange='ecommerce.events',
                routing_key=routing_key,
                body=json.dumps(event_message),
                properties=pika.BasicProperties(
                    delivery_mode=2,  # 메시지 지속성 보장
                    content_type='application/json'
                )
            )
            
            self.logger.info(f"Event published: {routing_key}")
            
        except Exception as e:
            self.logger.error(f"Failed to publish event {routing_key}: {e}")
            raise
    
    def subscribe_to_events(self, routing_patterns: list, queue_name: str, 
                          callback: Callable):
        """
        이벤트 구독 메서드
        - routing_patterns: 구독할 이벤트 패턴 리스트 (['order.*', 'payment.*'])
        - queue_name: 서비스별 고유 큐 이름
        - callback: 이벤트 처리 콜백 함수
        """
        try:
            # 서비스별 전용 큐 선언
            self.channel.queue_declare(
                queue=queue_name,
                durable=True,
                arguments={
                    'x-dead-letter-exchange': 'dlx.ecommerce',  # DLQ 설정
                    'x-message-ttl': 300000  # 5분 TTL
                }
            )
            
            # 패턴별 바인딩 설정
            for pattern in routing_patterns:
                self.channel.queue_bind(
                    exchange='ecommerce.events',
                    queue=queue_name,
                    routing_key=pattern
                )
            
            # QoS 설정: 동시 처리 메시지 수 제한
            self.channel.basic_qos(prefetch_count=10)
            
            # 메시지 소비 설정
            self.channel.basic_consume(
                queue=queue_name,
                on_message_callback=self._wrap_callback(callback),
                auto_ack=False
            )
            
            self.logger.info(f"Subscribed to patterns: {routing_patterns}")
            
        except Exception as e:
            self.logger.error(f"Failed to subscribe to events: {e}")
            raise
    
    def _wrap_callback(self, user_callback: Callable):
        """콜백 함수 래퍼 - 에러 처리 및 ACK 관리"""
        def wrapped_callback(ch, method, properties, body):
            try:
                event_data = json.loads(body.decode('utf-8'))
                user_callback(event_data)
                
                # 성공적 처리 후 ACK
                ch.basic_ack(delivery_tag=method.delivery_tag)
                
            except json.JSONDecodeError as e:
                self.logger.error(f"Invalid JSON in message: {e}")
                ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
                
            except Exception as e:
                self.logger.error(f"Error processing message: {e}")
                # 처리 실패 시 재시도 (최대 3회)
                ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
        
        return wrapped_callback
    
    def start_consuming(self):
        """이벤트 소비 시작"""
        self.logger.info("Starting event consumption...")
        self.channel.start_consuming()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89

# services/order_service.py - 주문 서비스 구현
from shared.event_bus import EventBus
import asyncio

class OrderService:
    """
    주문 처리 서비스
    - 고객 주문 접수 및 저장
    - 주문 관련 이벤트 발행 (order.created, order.cancelled 등)
    """
    
    def __init__(self):
        self.event_bus = EventBus()
        self.orders_db = {}  # 실제로는 데이터베이스 연결
    
    async def create_order(self, customer_id: str, items: list, total_amount: float):
        """주문 생성 및 이벤트 발행"""
        try:
            # 주문 정보 저장
            order_id = f"order_{int(time.time())}"
            order_data = {
                'order_id': order_id,
                'customer_id': customer_id,
                'items': items,
                'total_amount': total_amount,
                'status': 'pending'
            }
            
            self.orders_db[order_id] = order_data
            
            # 주문 생성 이벤트 발행 - 다른 서비스들이 구독하여 처리
            self.event_bus.publish_event('order.created', order_data)
            
            print(f"Order created: {order_id}")
            return order_id
            
        except Exception as e:
            print(f"Failed to create order: {e}")
            raise
    
    def handle_inventory_response(self, event_data: dict):
        """
        재고 서비스 응답 처리
        - inventory.checked 이벤트 수신 시 호출
        - 재고 확인 결과에 따라 주문 상태 업데이트
        """
        order_id = event_data['data']['order_id']
        inventory_available = event_data['data']['available']
        
        if inventory_available:
            # 재고 충분 - 결제 프로세스 시작 이벤트 발행
            self.event_bus.publish_event('order.inventory_confirmed', {
                'order_id': order_id,
                'next_step': 'payment'
            })
        else:
            # 재고 부족 - 주문 취소
            self.orders_db[order_id]['status'] = 'cancelled'
            self.event_bus.publish_event('order.cancelled', {
                'order_id': order_id,
                'reason': 'insufficient_inventory'
            })
    
    async def start_service(self):
        """서비스 시작 및 이벤트 구독"""
        await self.event_bus.connect()
        
        # 재고 응답 이벤트 구독
        self.event_bus.subscribe_to_events(
            routing_patterns=['inventory.checked'],
            queue_name='order.service.queue',
            callback=self.handle_inventory_response
        )
        
        self.event_bus.start_consuming()

# 사용 예시
if __name__ == '__main__':
    order_service = OrderService()
    
    # 주문 생성
    asyncio.run(order_service.create_order(
        customer_id='customer_123',
        items=[{'product_id': 'prod_1', 'quantity': 2}],
        total_amount=99.99
    ))
    
    # 이벤트 수신 대기
    asyncio.run(order_service.start_service())