Messaging Systems

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

+------------+    +------------+    +------------+
|            |    |            |    |            |
| 생산자(들)  | -> |   큐(들)   | -> | 소비자(들)  |
|            |    |            |    |            |
+------------+    +------------+    +------------+
                       |
                       v
                +-------------+
                |             |
                | 임시 저장소  |
                |             |
                +-------------+

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

+------------+    +----------------+    +------------+
|            |    |                |    |            |
| 생산자(들)  | -> | 로그/파티션(들) | -> | 소비자(들)  |
|            |    |                |    |            |
+------------+    +----------------+    +------------+
                         |
                         v
                  +-------------+
                  |             |
                  | 스토리지 계층 |
                  |             |
                  +-------------+

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

+------------+    +------------+    +------------+
|            |    |            |    |            |
| 태스크 생성자| -> | 태스크 큐(들)| -> | 워커(들)   |
|            |    |            |    |            |
+------------+    +------------+    +------------+
                       |                  |
                       v                  v
                +-------------+    +-------------+
                |             |    |             |
                | 태스크 저장소 |    | 결과 저장소  |
                |             |    |             |
                +-------------+    +-------------+

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

[Web App]
   |
   | (JSON 로그 전송)
   v
[Kafka Producer]
   |
   | (user-events 토픽으로 게시)
   v
[Kafka Broker Cluster] --> [Kafka Topic]
                                  |
               ---------------------------
               |                         |
         [Flink Processor]        [Kafka Consumer - Archiver]
               |                         |
         [ClickHouse]               [S3 Storage]

  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

import json
import asyncio
from typing import Dict, Any
from kafka import KafkaProducer, KafkaConsumer
from kafka.errors import KafkaError
import logging

# 설정
KAFKA_BOOTSTRAP_SERVERS = ['localhost:9092']
ORDER_TOPIC = 'order-events'
PAYMENT_TOPIC = 'payment-events'

class MessageProducer:
    """메시지 발행자 클래스"""
    
    def __init__(self):
        self.producer = KafkaProducer(
            bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
            key_serializer=lambda k: str(k).encode('utf-8'),
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            acks='all',  # 모든 복제본에서 확인
            retries=3,   # 재시도 횟수
            batch_size=16384,  # 배치 크기
            linger_ms=10       # 배치 대기 시간
        )
    
    async def publish_event(self, topic: str, key: str, event_data: Dict[str, Any]):
        """이벤트 발행"""
        try:
            future = self.producer.send(
                topic=topic,
                key=key,
                value=event_data
            )
            
            # 동기적으로 결과 확인
            record_metadata = future.get(timeout=10)
            logging.info(f"Event published to {record_metadata.topic}:{record_metadata.partition}:{record_metadata.offset}")
            return True
            
        except KafkaError as e:
            logging.error(f"Failed to publish event: {e}")
            return False
    
    def close(self):
        """프로듀서 종료"""
        self.producer.close()

class OrderService:
    """주문 서비스 클래스"""
    
    def __init__(self):
        self.producer = MessageProducer()
    
    async def create_order(self, order_data: Dict[str, Any]) -> str:
        """주문 생성 및 이벤트 발행"""
        order_id = f"order_{order_data['customer_id']}_{int(asyncio.get_event_loop().time())}"
        
        # 주문 데이터베이스 저장 (시뮬레이션)
        await self._save_order_to_db(order_id, order_data)
        
        # 주문 생성 이벤트 발행
        event = {
            'event_type': 'order_created',
            'order_id': order_id,
            'customer_id': order_data['customer_id'],
            'items': order_data['items'],
            'total_amount': order_data['total_amount'],
            'timestamp': asyncio.get_event_loop().time()
        }
        
        success = await self.producer.publish_event(
            topic=ORDER_TOPIC,
            key=order_id,  # 파티션 키로 주문 ID 사용
            event_data=event
        )
        
        if success:
            logging.info(f"Order {order_id} created and event published")
            return order_id
        else:
            logging.error(f"Failed to publish order creation event for {order_id}")
            raise Exception("Order event publication failed")
    
    async def _save_order_to_db(self, order_id: str, order_data: Dict[str, Any]):
        """주문 데이터베이스 저장 시뮬레이션"""
        await asyncio.sleep(0.1)  # DB 작업 시뮬레이션
        logging.info(f"Order {order_id} saved to database")

class PaymentService:
    """결제 서비스 클래스"""
    
    def __init__(self):
        self.producer = MessageProducer()
        self.consumer = KafkaConsumer(
            ORDER_TOPIC,
            bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
            group_id='payment-service-group',
            key_deserializer=lambda k: k.decode('utf-8'),
            value_deserializer=lambda v: json.loads(v.decode('utf-8')),
            auto_offset_reset='latest',
            enable_auto_commit=True
        )
    
    async def start_consuming(self):
        """주문 이벤트 수신 및 처리"""
        logging.info("Payment service started consuming order events")
        
        for message in self.consumer:
            try:
                event_data = message.value
                
                if event_data['event_type'] == 'order_created':
                    await self._process_payment(event_data)
                    
            except Exception as e:
                logging.error(f"Error processing payment: {e}")
    
    async def _process_payment(self, order_event: Dict[str, Any]):
        """결제 처리"""
        order_id = order_event['order_id']
        amount = order_event['total_amount']
        
        # 결제 처리 시뮬레이션
        await asyncio.sleep(0.2)
        
        # 결제 완료 이벤트 발행
        payment_event = {
            'event_type': 'payment_completed',
            'order_id': order_id,
            'payment_id': f"pay_{order_id}",
            'amount': amount,
            'status': 'completed',
            'timestamp': asyncio.get_event_loop().time()
        }
        
        success = await self.producer.publish_event(
            topic=PAYMENT_TOPIC,
            key=order_id,
            event_data=payment_event
        )
        
        if success:
            logging.info(f"Payment completed for order {order_id}")
        else:
            logging.error(f"Failed to publish payment completion event for {order_id}")

class NotificationService:
    """알림 서비스 클래스"""
    
    def __init__(self):
        self.consumer = KafkaConsumer(
            ORDER_TOPIC, PAYMENT_TOPIC,
            bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
            group_id='notification-service-group',
            key_deserializer=lambda k: k.decode('utf-8'),
            value_deserializer=lambda v: json.loads(v.decode('utf-8')),
            auto_offset_reset='latest'
        )
    
    async def start_consuming(self):
        """이벤트 수신 및 알림 전송"""
        logging.info("Notification service started consuming events")
        
        for message in self.consumer:
            try:
                event_data = message.value
                await self._send_notification(event_data)
                
            except Exception as e:
                logging.error(f"Error sending notification: {e}")
    
    async def _send_notification(self, event_data: Dict[str, Any]):
        """알림 전송"""
        event_type = event_data['event_type']
        order_id = event_data['order_id']
        
        notification_messages = {
            'order_created': f"주문 {order_id}이 접수되었습니다.",
            'payment_completed': f"주문 {order_id}의 결제가 완료되었습니다."
        }
        
        if event_type in notification_messages:
            # 실제 알림 전송 (이메일, SMS, 푸시 등)
            await asyncio.sleep(0.1)  # 알림 전송 시뮬레이션
            logging.info(f"Notification sent: {notification_messages[event_type]}")

# 사용 예시
async def main():
    """메인 실행 함수"""
    logging.basicConfig(level=logging.INFO)
    
    # 서비스 인스턴스 생성
    order_service = OrderService()
    payment_service = PaymentService()
    notification_service = NotificationService()
    
    # 백그라운드에서 컨슈머 서비스 실행
    payment_task = asyncio.create_task(payment_service.start_consuming())
    notification_task = asyncio.create_task(notification_service.start_consuming())
    
    # 주문 생성 시뮬레이션
    order_data = {
        'customer_id': 'customer_123',
        'items': [
            {'product_id': 'prod_1', 'quantity': 2, 'price': 29.99},
            {'product_id': 'prod_2', 'quantity': 1, 'price': 49.99}
        ],
        'total_amount': 109.97
    }
    
    try:
        order_id = await order_service.create_order(order_data)
        logging.info(f"Order creation completed: {order_id}")
        
        # 메시지 처리를 위한 대기
        await asyncio.sleep(5)
        
    except Exception as e:
        logging.error(f"Order creation failed: {e}")
    
    finally:
        # 정리 작업
        order_service.producer.close()
        payment_service.producer.close()
        payment_task.cancel()
        notification_task.cancel()

if __name__ == "__main__":
    asyncio.run(main())