Message Broker

1
2
3

| Magic Byte | Schema ID | Serialized Avro Payload |
|------------|-----------|--------------------------|
|     0x0    |    001    | binary encoded message   |

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

import pika

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

# 메시지 소비 콜백
def callback(ch, method, properties, body):
	print("받은 메시지:", body.decode())
	# 메시지 처리 완료 후 ACK
	ch.basic_ack(delivery_tag=method.delivery_tag)

# 큐 소비 시작
channel.basic_consume(queue='task_queue', on_message_callback=callback)

print('메시지 대기 중...')
channel.start_consuming()

1
2
3
4
5

1. 트랜잭션 시작
2. 메시지 publish (Producer.send)
3. 비즈니스 로직 실행
4. 결과 저장 (예: DB insert)
5. commitTransaction()

1

만약 처리 중 에러 발생 시 → abortTransaction() 수행

1
2

Properties props = new Properties();
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);

 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

# 메시지 브로커(RabbitMQ) 기반 주문 처리 예시

import pika

# 주문 메시지 발신(Producer)
def send_order(order_msg):
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='order_queue', durable=True)
    channel.basic_publish(exchange='', routing_key='order_queue', body=order_msg, properties=pika.BasicProperties(delivery_mode=2))
    print(f"Order sent: {order_msg}")
    connection.close()

# 주문 메시지 수신/처리(Consumer/Worker)
def process_orders():
    def callback(ch, method, properties, body):
        print(f"Order received: {body.decode()}")
        # 주문 처리 로직(DB저장 등)
        ch.basic_ack(delivery_tag=method.delivery_tag)
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='order_queue', durable=True)
    channel.basic_qos(prefetch_count=1) # 병렬 처리 시 각각 하나 씩
    channel.basic_consume(queue='order_queue', on_message_callback=callback)
    print('Waiting for orders…')
    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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234

import json
from kafka import KafkaProducer, KafkaConsumer
from dataclasses import dataclass
from typing import Dict, Any
import logging

@dataclass
class OrderEvent:
    """주문 이벤트 데이터 클래스"""
    order_id: str
    user_id: str
    items: list
    total_amount: float
    timestamp: str
    event_type: str  # order_created, payment_processed, etc.

class OrderService:
    """주문 서비스 - 메시지 프로듀서 역할"""
    
    def __init__(self, kafka_servers: str):
        # Kafka 프로듀서 초기화
        self.producer = KafkaProducer(
            bootstrap_servers=kafka_servers,
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            key_serializer=lambda k: k.encode('utf-8') if k else None
        )
        self.logger = logging.getLogger(__name__)
    
    def create_order(self, order_data: Dict[str, Any]) -> str:
        """주문 생성 및 이벤트 발행"""
        try:
            # 주문 데이터 검증
            order_id = self._generate_order_id()
            
            # 주문 이벤트 생성
            event = OrderEvent(
                order_id=order_id,
                user_id=order_data['user_id'],
                items=order_data['items'],
                total_amount=order_data['total_amount'],
                timestamp=self._get_current_timestamp(),
                event_type="order_created"
            )
            
            # 메시지 브로커에 이벤트 발행
            self._publish_event('order-events', event)
            
            self.logger.info(f"Order {order_id} created successfully")
            return order_id
            
        except Exception as e:
            self.logger.error(f"Failed to create order: {e}")
            raise
    
    def _publish_event(self, topic: str, event: OrderEvent):
        """이벤트를 Kafka 토픽에 발행"""
        message = {
            'order_id': event.order_id,
            'user_id': event.user_id,
            'items': event.items,
            'total_amount': event.total_amount,
            'timestamp': event.timestamp,
            'event_type': event.event_type
        }
        
        # 파티션 키로 order_id 사용 (순서 보장)
        self.producer.send(
            topic=topic,
            key=event.order_id,
            value=message
        )
        
        # 메시지 전송 완료 대기
        self.producer.flush()
    
    def _generate_order_id(self) -> str:
        """주문 ID 생성"""
        import uuid
        return str(uuid.uuid4())
    
    def _get_current_timestamp(self) -> str:
        """현재 타임스탬프 반환"""
        from datetime import datetime
        return datetime.now().isoformat()

class PaymentService:
    """결제 서비스 - 메시지 컨슈머 역할"""
    
    def __init__(self, kafka_servers: str):
        # Kafka 컨슈머 초기화
        self.consumer = KafkaConsumer(
            'order-events',
            bootstrap_servers=kafka_servers,
            value_deserializer=lambda m: json.loads(m.decode('utf-8')),
            key_deserializer=lambda k: k.decode('utf-8') if k else None,
            group_id='payment-service-group',  # 컨슈머 그룹
            auto_offset_reset='latest'  # 최신 메시지부터 처리
        )
        
        # 결제 완료 이벤트 발행용 프로듀서
        self.producer = KafkaProducer(
            bootstrap_servers=kafka_servers,
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            key_serializer=lambda k: k.encode('utf-8') if k else None
        )
        
        self.logger = logging.getLogger(__name__)
    
    def start_processing(self):
        """메시지 처리 시작"""
        self.logger.info("Payment service started processing messages")
        
        try:
            for message in self.consumer:
                event_data = message.value
                
                # 주문 생성 이벤트만 처리
                if event_data.get('event_type') == 'order_created':
                    self._process_payment(event_data)
                    
        except KeyboardInterrupt:
            self.logger.info("Payment service stopped")
        finally:
            self.consumer.close()
            self.producer.close()
    
    def _process_payment(self, order_data: Dict[str, Any]):
        """결제 처리 로직"""
        order_id = order_data['order_id']
        amount = order_data['total_amount']
        
        try:
            # 실제 결제 처리 (외부 결제 게이트웨이 호출)
            payment_result = self._call_payment_gateway(order_id, amount)
            
            if payment_result['success']:
                # 결제 완료 이벤트 발행
                payment_event = {
                    'order_id': order_id,
                    'payment_id': payment_result['payment_id'],
                    'amount': amount,
                    'status': 'completed',
                    'timestamp': self._get_current_timestamp(),
                    'event_type': 'payment_completed'
                }
                
                self.producer.send(
                    topic='payment-events',
                    key=order_id,
                    value=payment_event
                )
                
                self.logger.info(f"Payment completed for order {order_id}")
            else:
                # 결제 실패 처리
                self._handle_payment_failure(order_id, payment_result['error'])
                
        except Exception as e:
            self.logger.error(f"Payment processing failed for order {order_id}: {e}")
            self._handle_payment_failure(order_id, str(e))
    
    def _call_payment_gateway(self, order_id: str, amount: float) -> Dict[str, Any]:
        """결제 게이트웨이 호출 시뮬레이션"""
        import random
        import uuid
        
        # 90% 성공률로 시뮬레이션
        if random.random() < 0.9:
            return {
                'success': True,
                'payment_id': str(uuid.uuid4()),
                'transaction_id': f"TXN_{order_id}"
            }
        else:
            return {
                'success': False,
                'error': 'Insufficient funds'
            }
    
    def _handle_payment_failure(self, order_id: str, error: str):
        """결제 실패 처리"""
        failure_event = {
            'order_id': order_id,
            'error': error,
            'timestamp': self._get_current_timestamp(),
            'event_type': 'payment_failed'
        }
        
        self.producer.send(
            topic='payment-events',
            key=order_id,
            value=failure_event
        )
        
        self.logger.error(f"Payment failed for order {order_id}: {error}")
    
    def _get_current_timestamp(self) -> str:
        """현재 타임스탬프 반환"""
        from datetime import datetime
        return datetime.now().isoformat()

# 사용 예시
if __name__ == "__main__":
    import threading
    import time
    
    # Kafka 서버 설정
    KAFKA_SERVERS = 'localhost:9092'
    
    # 주문 서비스 초기화
    order_service = OrderService(KAFKA_SERVERS)
    
    # 결제 서비스 초기화 및 백그라운드 실행
    payment_service = PaymentService(KAFKA_SERVERS)
    payment_thread = threading.Thread(target=payment_service.start_processing)
    payment_thread.daemon = True
    payment_thread.start()
    
    # 테스트 주문 생성
    test_order = {
        'user_id': 'user123',
        'items': [
            {'product_id': 'prod1', 'quantity': 2, 'price': 50.0},
            {'product_id': 'prod2', 'quantity': 1, 'price': 30.0}
        ],
        'total_amount': 130.0
    }
    
    # 주문 생성 및 메시지 브로커를 통한 처리
    order_id = order_service.create_order(test_order)
    print(f"Order {order_id} submitted for processing")
    
    # 결제 처리 완료 대기
    time.sleep(5)