Message-Driven vs. Event-Driven Architecture

 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

# 주문 생성 → 메시지 브로커 → 결제 서비스
import pika
import json

# 메시지 브로커 연결
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 큐 생성
channel.queue_declare(queue='payment_queue', durable=True)

# 메시지 전송 함수
def send_order_to_payment(order_id, amount):
    message = json.dumps({
        "order_id": order_id,
        "amount": amount
    })
    channel.basic_publish(
        exchange='',
        routing_key='payment_queue',
        body=message,
        properties=pika.BasicProperties(delivery_mode=2)
    )
    print(f"Sent order {order_id} for payment.")

send_order_to_payment("ORD-1001", 19900)

connection.close()

 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

# 주문 발생 시 메시지 발행과 큐 소비자 예시

import pika

# 프로듀서: 주문 발생시 메시지 발행
def send_order_message(order_id, data):
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='order_queue')
    channel.basic_publish(exchange='',
                          routing_key='order_queue',
                          body=f"{order_id}:{data}")
    connection.close()

# 컨슈머: 주문 메시지 처리(재고 서비스)
def inventory_consumer():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='order_queue')
    def callback(ch, method, properties, body):
        order_id, data = body.decode().split(":")
        print(f"재고 처리: 주문번호 {order_id}, 데이터 {data}")
        # 재고 차감 처리 코드 위치
    channel.basic_consume(queue='order_queue', on_message_callback=callback, auto_ack=True)
    print('재고 서비스 대기 중…')
    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
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461

"""
전자상거래 주문 처리 시스템
Message-Driven과 Event-Driven Architecture 하이브리드 구현
"""

import asyncio
import json
import uuid
from datetime import datetime
from typing import Dict, List, Any, Callable
from enum import Enum
from dataclasses import dataclass, asdict
import logging

# ============================================================================
# 1. 기본 데이터 모델
# ============================================================================

class OrderStatus(Enum):
    """주문 상태 열거형"""
    PENDING = "pending"
    PAYMENT_PROCESSING = "payment_processing"
    PAID = "paid"
    SHIPPED = "shipped"
    DELIVERED = "delivered"
    CANCELLED = "cancelled"

@dataclass
class Order:
    """주문 데이터 클래스"""
    id: str
    customer_id: str
    items: List[Dict[str, Any]]
    total_amount: float
    status: OrderStatus
    created_at: datetime

@dataclass
class Message:
    """메시지 데이터 클래스 (Message-Driven Architecture용)"""
    id: str
    type: str
    payload: Dict[str, Any]
    sender: str
    receiver: str
    created_at: datetime
    correlation_id: str = None

@dataclass
class Event:
    """이벤트 데이터 클래스 (Event-Driven Architecture용)"""
    id: str
    type: str
    payload: Dict[str, Any]
    source: str
    created_at: datetime
    version: str = "1.0"

# ============================================================================
# 2. Message-Driven Architecture 구현
# ============================================================================

class MessageQueue:
    """
    메시지 큐 구현
    - 순차적 메시지 처리를 위한 FIFO 큐
    - 메시지 지속성 및 재시도 메커니즘 제공
    """
    def __init__(self, name: str):
        self.name = name
        self.queue: List[Message] = []
        self.handlers: Dict[str, Callable] = {}
        self.dead_letter_queue: List[Message] = []
        
    async def send_message(self, message: Message):
        """메시지를 큐에 전송"""
        self.queue.append(message)
        logging.info(f"Message sent to {self.name}: {message.type}")
        
    async def subscribe(self, message_type: str, handler: Callable):
        """특정 메시지 타입에 대한 핸들러 등록"""
        self.handlers[message_type] = handler
        
    async def process_messages(self):
        """큐의 메시지들을 순차적으로 처리"""
        while self.queue:
            message = self.queue.pop(0)
            try:
                if message.type in self.handlers:
                    await self.handlers[message.type](message)
                    logging.info(f"Message processed: {message.id}")
                else:
                    logging.warning(f"No handler for message type: {message.type}")
            except Exception as e:
                # 실패한 메시지를 Dead Letter Queue로 이동
                self.dead_letter_queue.append(message)
                logging.error(f"Message processing failed: {e}")

class MessageBroker:
    """
    메시지 브로커 - 여러 큐를 관리하고 메시지 라우팅 담당
    """
    def __init__(self):
        self.queues: Dict[str, MessageQueue] = {}
        
    def create_queue(self, name: str) -> MessageQueue:
        """새로운 메시지 큐 생성"""
        self.queues[name] = MessageQueue(name)
        return self.queues[name]
        
    async def route_message(self, queue_name: str, message: Message):
        """메시지를 지정된 큐로 라우팅"""
        if queue_name in self.queues:
            await self.queues[queue_name].send_message(message)
        else:
            logging.error(f"Queue not found: {queue_name}")

# ============================================================================
# 3. Event-Driven Architecture 구현
# ============================================================================

class EventBus:
    """
    이벤트 버스 구현
    - Publish-Subscribe 패턴으로 이벤트 배포
    - 다수의 구독자에게 동시 이벤트 전달
    """
    def __init__(self):
        self.subscribers: Dict[str, List[Callable]] = {}
        self.event_store: List[Event] = []
        
    async def publish(self, event: Event):
        """이벤트 발행 - 모든 구독자에게 브로드캐스트"""
        # 이벤트 저장 (Event Sourcing)
        self.event_store.append(event)
        
        # 구독자들에게 이벤트 전달
        if event.type in self.subscribers:
            tasks = []
            for subscriber in self.subscribers[event.type]:
                tasks.append(subscriber(event))
            await asyncio.gather(*tasks, return_exceptions=True)
            
        logging.info(f"Event published: {event.type} to {len(self.subscribers.get(event.type, []))} subscribers")
        
    async def subscribe(self, event_type: str, handler: Callable):
        """이벤트 타입에 대한 핸들러 구독"""
        if event_type not in self.subscribers:
            self.subscribers[event_type] = []
        self.subscribers[event_type].append(handler)

# ============================================================================
# 4. 핵심 비즈니스 서비스 (Message-Driven)
# ============================================================================

class OrderService:
    """
    주문 서비스 - 핵심 주문 처리 로직
    Message-Driven 방식으로 순차적 처리 보장
    """
    def __init__(self, message_broker: MessageBroker, event_bus: EventBus):
        self.message_broker = message_broker
        self.event_bus = event_bus
        self.orders: Dict[str, Order] = {}
        
    async def create_order(self, customer_id: str, items: List[Dict], total_amount: float) -> str:
        """주문 생성"""
        order_id = str(uuid.uuid4())
        order = Order(
            id=order_id,
            customer_id=customer_id,
            items=items,
            total_amount=total_amount,
            status=OrderStatus.PENDING,
            created_at=datetime.now()
        )
        
        self.orders[order_id] = order
        
        # Message-Driven: 재고 확인 메시지 전송
        inventory_message = Message(
            id=str(uuid.uuid4()),
            type="check_inventory",
            payload={"order_id": order_id, "items": items},
            sender="order_service",
            receiver="inventory_service",
            created_at=datetime.now(),
            correlation_id=order_id
        )
        
        await self.message_broker.route_message("inventory_queue", inventory_message)
        
        # Event-Driven: 주문 생성 이벤트 발행
        await self.event_bus.publish(Event(
            id=str(uuid.uuid4()),
            type="order_created",
            payload=asdict(order),
            source="order_service",
            created_at=datetime.now()
        ))
        
        return order_id
        
    async def handle_inventory_confirmed(self, message: Message):
        """재고 확인 완료 메시지 처리"""
        order_id = message.payload["order_id"]
        if order_id in self.orders:
            # Payment 서비스로 메시지 전송
            payment_message = Message(
                id=str(uuid.uuid4()),
                type="process_payment",
                payload={
                    "order_id": order_id,
                    "amount": self.orders[order_id].total_amount
                },
                sender="order_service",
                receiver="payment_service",
                created_at=datetime.now(),
                correlation_id=order_id
            )
            
            await self.message_broker.route_message("payment_queue", payment_message)

class PaymentService:
    """
    결제 서비스 - 결제 처리 담당
    """
    def __init__(self, message_broker: MessageBroker, event_bus: EventBus):
        self.message_broker = message_broker
        self.event_bus = event_bus
        
    async def handle_payment_request(self, message: Message):
        """결제 처리 메시지 핸들러"""
        order_id = message.payload["order_id"]
        amount = message.payload["amount"]
        
        # 결제 처리 로직 (시뮬레이션)
        await asyncio.sleep(1)  # 결제 처리 시간 시뮬레이션
        
        # 배송 서비스로 메시지 전송
        shipping_message = Message(
            id=str(uuid.uuid4()),
            type="prepare_shipping",
            payload={"order_id": order_id},
            sender="payment_service",
            receiver="shipping_service",
            created_at=datetime.now(),
            correlation_id=order_id
        )
        
        await self.message_broker.route_message("shipping_queue", shipping_message)
        
        # 결제 완료 이벤트 발행
        await self.event_bus.publish(Event(
            id=str(uuid.uuid4()),
            type="payment_completed",
            payload={"order_id": order_id, "amount": amount},
            source="payment_service",
            created_at=datetime.now()
        ))

# ============================================================================
# 5. 부가 서비스들 (Event-Driven)
# ============================================================================

class NotificationService:
    """
    알림 서비스 - 고객에게 실시간 알림 전송
    Event-Driven 방식으로 다양한 이벤트에 반응
    """
    def __init__(self):
        self.sent_notifications: List[Dict] = []
        
    async def handle_order_created(self, event: Event):
        """주문 생성 이벤트 처리"""
        order_data = event.payload
        notification = {
            "type": "order_confirmation",
            "customer_id": order_data["customer_id"],
            "message": f"주문 {order_data['id']}가 접수되었습니다.",
            "timestamp": datetime.now()
        }
        self.sent_notifications.append(notification)
        logging.info(f"Notification sent: {notification['message']}")
        
    async def handle_payment_completed(self, event: Event):
        """결제 완료 이벤트 처리"""
        order_id = event.payload["order_id"]
        notification = {
            "type": "payment_confirmation",
            "order_id": order_id,
            "message": f"주문 {order_id}의 결제가 완료되었습니다.",
            "timestamp": datetime.now()
        }
        self.sent_notifications.append(notification)
        logging.info(f"Notification sent: {notification['message']}")

class AnalyticsService:
    """
    분석 서비스 - 주문 및 결제 데이터 분석
    실시간으로 비즈니스 메트릭 수집
    """
    def __init__(self):
        self.metrics: Dict[str, int] = {
            "total_orders": 0,
            "total_revenue": 0,
            "failed_payments": 0
        }
        
    async def handle_order_created(self, event: Event):
        """주문 생성 분석"""
        self.metrics["total_orders"] += 1
        logging.info(f"Analytics: Total orders = {self.metrics['total_orders']}")
        
    async def handle_payment_completed(self, event: Event):
        """결제 완료 분석"""
        amount = event.payload["amount"]
        self.metrics["total_revenue"] += amount
        logging.info(f"Analytics: Total revenue = {self.metrics['total_revenue']}")

# ============================================================================
# 6. 시스템 통합 및 실행
# ============================================================================

class ECommerceSystem:
    """
    전자상거래 시스템 메인 클래스
    Message-Driven과 Event-Driven 아키텍처를 통합 관리
    """
    def __init__(self):
        # 인프라 컴포넌트 초기화
        self.message_broker = MessageBroker()
        self.event_bus = EventBus()
        
        # 큐 생성
        self.inventory_queue = self.message_broker.create_queue("inventory_queue")
        self.payment_queue = self.message_broker.create_queue("payment_queue")
        self.shipping_queue = self.message_broker.create_queue("shipping_queue")
        
        # 서비스 초기화
        self.order_service = OrderService(self.message_broker, self.event_bus)
        self.payment_service = PaymentService(self.message_broker, self.event_bus)
        self.notification_service = NotificationService()
        self.analytics_service = AnalyticsService()
        
    async def setup(self):
        """시스템 설정 및 핸들러 등록"""
        # Message-Driven 핸들러 등록
        await self.inventory_queue.subscribe("check_inventory", self._handle_inventory_check)
        await self.payment_queue.subscribe("process_payment", self.payment_service.handle_payment_request)
        await self.shipping_queue.subscribe("prepare_shipping", self._handle_shipping_preparation)
        
        # Event-Driven 구독자 등록
        await self.event_bus.subscribe("order_created", self.notification_service.handle_order_created)
        await self.event_bus.subscribe("order_created", self.analytics_service.handle_order_created)
        await self.event_bus.subscribe("payment_completed", self.notification_service.handle_payment_completed)
        await self.event_bus.subscribe("payment_completed", self.analytics_service.handle_payment_completed)
        
    async def _handle_inventory_check(self, message: Message):
        """재고 확인 처리 (시뮬레이션)"""
        order_id = message.payload["order_id"]
        items = message.payload["items"]
        
        # 재고 확인 로직 (시뮬레이션)
        await asyncio.sleep(0.5)
        
        # 재고 확인 완료 메시지를 주문 서비스로 전송
        confirmation_message = Message(
            id=str(uuid.uuid4()),
            type="inventory_confirmed",
            payload={"order_id": order_id, "status": "confirmed"},
            sender="inventory_service",
            receiver="order_service",
            created_at=datetime.now(),
            correlation_id=order_id
        )
        
        # 주문 서비스의 핸들러 직접 호출 (실제로는 큐를 통해 처리)
        await self.order_service.handle_inventory_confirmed(confirmation_message)
        
    async def _handle_shipping_preparation(self, message: Message):
        """배송 준비 처리 (시뮬레이션)"""
        order_id = message.payload["order_id"]
        
        # 배송 준비 로직
        await asyncio.sleep(0.3)
        
        # 배송 준비 완료 이벤트 발행
        await self.event_bus.publish(Event(
            id=str(uuid.uuid4()),
            type="shipping_prepared",
            payload={"order_id": order_id},
            source="shipping_service",
            created_at=datetime.now()
        ))
        
    async def process_all_queues(self):
        """모든 큐의 메시지 처리"""
        tasks = []
        for queue in self.message_broker.queues.values():
            tasks.append(queue.process_messages())
        await asyncio.gather(*tasks)
        
    async def create_order(self, customer_id: str, items: List[Dict], total_amount: float):
        """주문 생성 및 처리"""
        order_id = await self.order_service.create_order(customer_id, items, total_amount)
        await self.process_all_queues()
        return order_id
        
    def get_metrics(self):
        """시스템 메트릭 조회"""
        return {
            "analytics": self.analytics_service.metrics,
            "notifications_sent": len(self.notification_service.sent_notifications),
            "events_stored": len(self.event_bus.event_store)
        }

# ============================================================================
# 7. 사용 예시 및 테스트
# ============================================================================

async def main():
    """시스템 실행 및 테스트"""
    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
    
    # 시스템 초기화
    ecommerce = ECommerceSystem()
    await ecommerce.setup()
    
    print("=== 전자상거래 시스템 시작 ===")
    
    # 테스트 주문 생성
    test_items = [
        {"product_id": "P001", "name": "노트북", "quantity": 1, "price": 1500000},
        {"product_id": "P002", "name": "마우스", "quantity": 2, "price": 50000}
    ]
    
    # 첫 번째 주문
    print("\n--- 첫 번째 주문 처리 ---")
    order1 = await ecommerce.create_order("CUST001", test_items, 1600000)
    print(f"주문 생성됨: {order1}")
    
    # 두 번째 주문
    print("\n--- 두 번째 주문 처리 ---")
    order2 = await ecommerce.create_order("CUST002", test_items[:1], 1500000)
    print(f"주문 생성됨: {order2}")
    
    # 시스템 메트릭 출력
    print("\n--- 시스템 메트릭 ---")
    metrics = ecommerce.get_metrics()
    print(f"분석 데이터: {metrics['analytics']}")
    print(f"전송된 알림 수: {metrics['notifications_sent']}")
    print(f"저장된 이벤트 수: {metrics['events_stored']}")
    
    # 이벤트 스토어 내용 출력
    print("\n--- 이벤트 히스토리 ---")
    for event in ecommerce.event_bus.event_store:
        print(f"{event.created_at}: {event.type} from {event.source}")

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

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

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "CreateInvoiceCommand",
  "type": "object",
  "properties": {
    "command_id": { "type": "string", "format": "uuid" },
    "order_id": { "type": "string" },
    "customer_id": { "type": "string" },
    "amount": { "type": "number" },
    "issued_at": { "type": "string", "format": "date-time" }
  },
  "required": ["command_id", "order_id", "customer_id", "amount", "issued_at"]
}

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

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "OrderCreatedEvent",
  "type": "object",
  "properties": {
    "event_id": { "type": "string", "format": "uuid" },
    "event_type": { "type": "string", "enum": ["OrderCreated"] },
    "occurred_at": { "type": "string", "format": "date-time" },
    "payload": {
      "type": "object",
      "properties": {
        "order_id": { "type": "string" },
        "customer_id": { "type": "string" },
        "total_price": { "type": "number" }
      },
      "required": ["order_id", "customer_id", "total_price"]
    }
  },
  "required": ["event_id", "event_type", "occurred_at", "payload"]
}

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

# publisher.py
import pika
import json
import uuid
from datetime import datetime

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='invoice_commands')

message = {
    "command_id": str(uuid.uuid4()),
    "order_id": "ORD-1001",
    "customer_id": "CUST-789",
    "amount": 120.5,
    "issued_at": datetime.utcnow().isoformat()
}
channel.basic_publish(
    exchange='',
    routing_key='invoice_commands',
    body=json.dumps(message)
)
print("Message sent.")
connection.close()

1
2
3
4
5
6
7

# consumer.py
def callback(ch, method, properties, body):
    data = json.loads(body)
    print(f"Processing command: {data['command_id']} → Create invoice for order {data['order_id']}")

channel.basic_consume(queue='invoice_commands', on_message_callback=callback, auto_ack=True)
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

// producer.js
const { Kafka } = require('kafkajs');
const kafka = new Kafka({ clientId: 'order-app', brokers: ['localhost:9092'] });

const producer = kafka.producer();

async function publishEvent() {
  await producer.connect();
  const event = {
    event_id: 'evt-' + Date.now(),
    event_type: 'OrderCreated',
    occurred_at: new Date().toISOString(),
    payload: {
      order_id: 'ORD-1001',
      customer_id: 'CUST-789',
      total_price: 99.99
    }
  };

  await producer.send({
    topic: 'order.events',
    messages: [{ key: event.event_id, value: JSON.stringify(event) }]
  });

  console.log('Event published.');
  await producer.disconnect();
}

publishEvent();

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

// consumer.js
const kafka = new Kafka({ clientId: 'invoice-service', brokers: ['localhost:9092'] });
const consumer = kafka.consumer({ groupId: 'invoice-consumer-group' });

async function run() {
  await consumer.connect();
  await consumer.subscribe({ topic: 'order.events', fromBeginning: true });

  await consumer.run({
    eachMessage: async ({ message }) => {
      const event = JSON.parse(message.value.toString());
      console.log(`Received event ${event.event_type} for order ${event.payload.order_id}`);
    }
  });
}
run();

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

# 1. Command 처리 시 Outbox 삽입
def create_order(order_data):
    order = Order(**order_data)
    db.session.add(order)

    event = OutboxEvent(
        type="OrderCreated",
        payload=json.dumps({"order_id": order.id, "total": order.total}),
        status="PENDING"
    )
    db.session.add(event)
    db.session.commit()

1
2
3
4
5
6
7

# 2. Poller: 주기적으로 Kafka에 발행
def publish_events():
    events = db.session.query(OutboxEvent).filter_by(status="PENDING").all()
    for event in events:
        kafka_producer.send("order.events", event.payload)
        event.status = "SENT"
    db.session.commit()

 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

# send_command.py
from kafka import KafkaProducer
import json
import uuid
from datetime import datetime

producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

command = {
    "command_id": str(uuid.uuid4()),
    "command_type": "CreateOrder",
    "timestamp": datetime.utcnow().isoformat(),
    "payload": {
        "order_id": "ORD-1001",
        "customer_id": "C-123",
        "amount": 99.99
    }
}

producer.send("order.commands", value=command)
producer.flush()
print(f"✅ Command sent: {command}")

 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

# command_consumer.py
from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    "order.commands",
    bootstrap_servers='localhost:9092',
    group_id="order-command-handler",
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    auto_offset_reset='earliest',
    enable_auto_commit=True
)

for message in consumer:
    command = message.value
    print(f"📥 Received Command: {command['command_type']} for Order {command['payload']['order_id']}")
    
    # Command 처리 후 이벤트 발행 (EDA 트리거)
    from kafka import KafkaProducer
    producer = KafkaProducer(
        bootstrap_servers='localhost:9092',
        value_serializer=lambda v: json.dumps(v).encode('utf-8')
    )
    event = {
        "event_id": str(uuid.uuid4()),
        "event_type": "OrderCreated",
        "occurred_at": command["timestamp"],
        "payload": command["payload"]
    }
    producer.send("order.events", value=event)
    producer.flush()
    print(f"📤 Published Event: {event['event_type']}")

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

# invoice_consumer.py
from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    "order.events",
    bootstrap_servers='localhost:9092',
    group_id="invoice-service",
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    auto_offset_reset='earliest',
    enable_auto_commit=True
)

for message in consumer:
    event = message.value
    print(f"🧾 [Invoice Service] 처리 중: {event['event_type']} - Order {event['payload']['order_id']}")

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

# notification_consumer.py
from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    "order.events",
    bootstrap_servers='localhost:9092',
    group_id="notification-service",
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    auto_offset_reset='earliest',
    enable_auto_commit=True
)

for message in consumer:
    event = message.value
    print(f"📣 [Notification Service] 사용자에게 알림 전송: {event['payload']['customer_id']}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

{
  "event_id": "abc-123",
  "event_type": "OrderCreated",
  "occurred_at": "2025-07-17T14:00:00Z",
  "payload": {
    "order_id": "ORD-101",
    "customer_id": "C-999",
    "is_vip": true
  }
}