Message-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
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

# RabbitMQ를 이용한 Queue-Based 구현
import pika
import json
import time

class OrderProcessor:
    def __init__(self):
        # RabbitMQ 연결 설정
        self.connection = pika.BlockingConnection(
            pika.ConnectionParameters('localhost')
        )
        self.channel = self.connection.channel()
        
        # 주문 처리 큐 선언
        self.channel.queue_declare(
            queue='order_processing',
            durable=True  # 큐 지속성 보장
        )
    
    def publish_order(self, order_data):
        """주문 메시지를 큐에 발행"""
        message = json.dumps(order_data)
        
        self.channel.basic_publish(
            exchange='',
            routing_key='order_processing',
            body=message,
            properties=pika.BasicProperties(
                delivery_mode=2,  # 메시지 지속성
                correlation_id=order_data.get('order_id')
            )
        )
        print(f"주문 전송됨: {order_data['order_id']}")
    
    def process_orders(self):
        """주문 메시지 소비 및 처리"""
        def callback(ch, method, properties, body):
            try:
                # 메시지 역직렬화
                order_data = json.loads(body.decode())
                
                # 주문 처리 로직 (예: 재고 확인, 결제 처리)
                self.handle_order(order_data)
                
                # 처리 완료 확인 응답
                ch.basic_ack(delivery_tag=method.delivery_tag)
                print(f"주문 처리 완료: {order_data['order_id']}")
                
            except Exception as e:
                # 처리 실패 시 메시지 거부
                ch.basic_nack(
                    delivery_tag=method.delivery_tag,
                    requeue=False  # Dead Letter Queue로 전송
                )
                print(f"주문 처리 실패: {str(e)}")
        
        # QoS 설정 (한 번에 하나의 메시지만 처리)
        self.channel.basic_qos(prefetch_count=1)
        
        # 소비자 등록
        self.channel.basic_consume(
            queue='order_processing',
            on_message_callback=callback
        )
        
        print("주문 처리 대기 중…")
        self.channel.start_consuming()
    
    def handle_order(self, order_data):
        """실제 주문 처리 로직"""
        time.sleep(2)  # 처리 시간 시뮬레이션
        # 재고 확인, 결제 처리, 배송 준비 등
        pass

  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

// Apache Kafka를 이용한 Pub-Sub 구현
const kafka = require('kafkajs');

class EventBus {
    constructor() {
        // Kafka 클라이언트 초기화
        this.kafka = kafka({
            clientId: 'event-bus-app',
            brokers: ['localhost:9092']
        });
        
        this.producer = this.kafka.producer();
        this.consumer = this.kafka.consumer({ 
            groupId: 'notification-group' 
        });
    }
    
    async publishEvent(eventType, eventData) {
        /**
         * 이벤트 발행 메소드
         * @param {string} eventType - 이벤트 타입 (토픽명)
         * @param {object} eventData - 이벤트 데이터
         */
        try {
            await this.producer.connect();
            
            const message = {
                key: eventData.id || Date.now().toString(),
                value: JSON.stringify({
                    eventType,
                    timestamp: new Date().toISOString(),
                    data: eventData
                }),
                headers: {
                    'content-type': 'application/json',
                    'event-version': '1.0'
                }
            };
            
            await this.producer.send({
                topic: eventType,
                messages: [message]
            });
            
            console.log(`이벤트 발행됨: ${eventType}`, eventData.id);
            
        } catch (error) {
            console.error('이벤트 발행 실패:', error);
            throw error;
        }
    }
    
    async subscribeToEvents(eventTypes, handler) {
        /**
         * 이벤트 구독 메소드
         * @param {string[]} eventTypes - 구독할 이벤트 타입들
         * @param {function} handler - 이벤트 처리 함수
         */
        try {
            await this.consumer.connect();
            
            // 토픽 구독
            for (const eventType of eventTypes) {
                await this.consumer.subscribe({ 
                    topic: eventType,
                    fromBeginning: false 
                });
            }
            
            await this.consumer.run({
                eachMessage: async ({ topic, partition, message }) => {
                    try {
                        // 메시지 역직렬화
                        const eventData = JSON.parse(message.value.toString());
                        
                        // 이벤트 처리
                        await handler(topic, eventData);
                        
                        console.log(`이벤트 처리 완료: ${topic}`, eventData.data.id);
                        
                    } catch (error) {
                        console.error('이벤트 처리 실패:', error);
                        // 재시도 또는 DLQ 전송 로직
                    }
                }
            });
            
        } catch (error) {
            console.error('이벤트 구독 실패:', error);
            throw error;
        }
    }
}

// 사용 예시
class OrderService {
    constructor() {
        this.eventBus = new EventBus();
    }
    
    async createOrder(orderData) {
        // 주문 생성 로직
        const order = {
            id: this.generateOrderId(),
            customerId: orderData.customerId,
            items: orderData.items,
            totalAmount: this.calculateTotal(orderData.items),
            status: 'created'
        };
        
        // 주문 생성 이벤트 발행
        await this.eventBus.publishEvent('order-created', order);
        
        return order;
    }
}

class NotificationService {
    constructor() {
        this.eventBus = new EventBus();
        this.initEventListeners();
    }
    
    async initEventListeners() {
        // 주문 관련 이벤트 구독
        await this.eventBus.subscribeToEvents(
            ['order-created', 'order-completed', 'order-cancelled'],
            this.handleOrderEvent.bind(this)
        );
    }
    
    async handleOrderEvent(eventType, eventData) {
        /**
         * 주문 이벤트 처리
         * @param {string} eventType - 이벤트 타입
         * @param {object} eventData - 이벤트 데이터
         */
        switch (eventType) {
            case 'order-created':
                await this.sendOrderConfirmation(eventData.data);
                break;
            case 'order-completed':
                await this.sendCompletionNotification(eventData.data);
                break;
            case 'order-cancelled':
                await this.sendCancellationNotification(eventData.data);
                break;
        }
    }
    
    async sendOrderConfirmation(orderData) {
        // 주문 확인 알림 전송 로직
        console.log(`주문 확인 알림 전송: ${orderData.id}`);
    }
}

 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

# 메시지 브로커 클래스
class MessageBroker:
    def __init__(self):
        self.subscribers = {}

    def subscribe(self, topic, callback):
        if topic not in self.subscribers:
            self.subscribers[topic] = []
        self.subscribers[topic].append(callback)

    def publish(self, topic, data):
        if topic in self.subscribers:
            for callback in self.subscribers[topic]:
                callback(data)

# 메시지 발행자(주문 서비스)
def order_service(broker):
    broker.publish('order', {'order_id': 123, 'product': 'Laptop'})

# 메시지 소비자(결제, 배송, 알림 서비스)
def payment_service(data):
    print('결제 처리:', data)
def delivery_service(data):
    print('배송 처리:', data)
def notification_service(data):
    print('알림 처리:', data)

# 브로커 생성
broker = MessageBroker()
broker.subscribe('order', payment_service)
broker.subscribe('order', delivery_service)
broker.subscribe('order', notification_service)

# 메시지 발행
order_service(broker)

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

# producer.py - 주문 메시지 발행
import pika, json

def publish_order(order):
    conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    ch = conn.channel()
    ch.exchange_declare(exchange='order_exchange', exchange_type='fanout', durable=True)
    ch.basic_publish(
        exchange='order_exchange',
        routing_key='',
        body=json.dumps(order),
        properties=pika.BasicProperties(delivery_mode=2)  # 메시지 내구성
    )
    conn.close()

if __name__ == '__main__':
    publish_order({'id': '123', 'item': 'ABC', 'qty': 2})

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

# consumer.py - 재고 서비스
import pika, json

def callback(ch, method, properties, body):
    order = json.loads(body)
    try:
        print(f"재고 처리: {order}")
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception:
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
        ch.basic_publish(exchange='', routing_key='order.dlq', body=body)

conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
ch = conn.channel()
ch.exchange_declare(exchange='order_exchange', exchange_type='fanout', durable=True)
q = ch.queue_declare(queue='', exclusive=True).method.queue
ch.queue_bind(exchange='order_exchange', queue=q)
ch.basic_consume(queue=q, on_message_callback=callback)
ch.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

# 사용자 서비스 - 사용자 행동 이벤트 발행
import asyncio
import json
from datetime import datetime
from typing import Dict, Any
import aioredis
from kafka import KafkaProducer, KafkaConsumer
import logging

class UserService:
    """사용자 서비스 - 사용자 행동 추적 및 이벤트 발행"""
    
    def __init__(self, kafka_config: Dict[str, str]):
        """
        사용자 서비스 초기화
        Args:
            kafka_config: Kafka 브로커 설정
        """
        self.producer = KafkaProducer(
            bootstrap_servers=kafka_config['brokers'],
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            key_serializer=lambda k: str(k).encode('utf-8')
        )
        self.logger = logging.getLogger(__name__)
    
    async def track_user_activity(self, user_id: str, activity_type: str, 
                                content_id: str = None, metadata: Dict = None):
        """
        사용자 활동 추적 및 이벤트 발행
        Args:
            user_id: 사용자 식별자
            activity_type: 활동 타입 (view, search, rate, etc.)
            content_id: 컨텐츠 식별자
            metadata: 추가 메타데이터
        """
        try:
            # 사용자 활동 이벤트 생성
            event_data = {
                'event_id': f"{user_id}_{datetime.now().timestamp()}",
                'user_id': user_id,
                'activity_type': activity_type,
                'content_id': content_id,
                'timestamp': datetime.now().isoformat(),
                'metadata': metadata or {}
            }
            
            # 메시지 토픽 결정 (활동 타입에 따라)
            topic = f"user-{activity_type}-events"
            
            # 비동기 메시지 발행
            future = self.producer.send(
                topic=topic,
                key=user_id,  # 파티셔닝을 위한 키
                value=event_data
            )
            
            # 전송 결과 확인
            record_metadata = future.get(timeout=10)
            
            self.logger.info(
                f"사용자 활동 이벤트 발행: {activity_type}, "
                f"Topic: {record_metadata.topic}, "
                f"Partition: {record_metadata.partition}"
            )
            
            return True
            
        except Exception as e:
            self.logger.error(f"이벤트 발행 실패: {str(e)}")
            return False
    
    async def handle_user_viewing(self, user_id: str, content_id: str, 
                                duration: int, position: int):
        """
        시청 이벤트 처리 - 개인화 추천을 위한 중요 데이터
        Args:
            user_id: 사용자 ID
            content_id: 컨텐츠 ID  
            duration: 시청 시간 (초)
            position: 시청 위치 (초)
        """
        viewing_metadata = {
            'duration_seconds': duration,
            'current_position': position,
            'completion_rate': position / duration if duration > 0 else 0,
            'device_type': 'web',  # 실제로는 요청에서 추출
            'quality': '1080p'
        }
        
        await self.track_user_activity(
            user_id=user_id,
            activity_type='view',
            content_id=content_id,
            metadata=viewing_metadata
        )

class RecommendationService:
    """추천 서비스 - 사용자 이벤트 기반 실시간 추천"""
    
    def __init__(self, kafka_config: Dict[str, str], redis_config: Dict[str, str]):
        """
        추천 서비스 초기화
        Args:
            kafka_config: Kafka 설정
            redis_config: Redis 설정 (캐싱용)
        """
        self.consumer = KafkaConsumer(
            'user-view-events',
            'user-search-events', 
            'user-rate-events',
            bootstrap_servers=kafka_config['brokers'],
            group_id='recommendation-service',
            value_deserializer=lambda m: json.loads(m.decode('utf-8')),
            enable_auto_commit=True,
            auto_offset_reset='latest'
        )
        self.redis_config = redis_config
        self.logger = logging.getLogger(__name__)
    
    async def start_event_processing(self):
        """이벤트 소비 및 추천 업데이트 시작"""
        redis_client = await aioredis.from_url(
            f"redis://{self.redis_config['host']}:{self.redis_config['port']}"
        )
        
        try:
            for message in self.consumer:
                await self._process_user_event(message.value, redis_client)
                
        except Exception as e:
            self.logger.error(f"이벤트 처리 오류: {str(e)}")
        finally:
            await redis_client.close()
    
    async def _process_user_event(self, event_data: Dict[str, Any], 
                                redis_client):
        """
        사용자 이벤트 처리 및 추천 모델 업데이트
        Args:
            event_data: 사용자 이벤트 데이터
            redis_client: Redis 클라이언트
        """
        try:
            user_id = event_data['user_id']
            activity_type = event_data['activity_type']
            content_id = event_data.get('content_id')
            
            if activity_type == 'view' and content_id:
                # 시청 이벤트 기반 추천 업데이트
                await self._update_viewing_history(
                    user_id, content_id, event_data['metadata'], redis_client
                )
                
                # 실시간 추천 생성
                recommendations = await self._generate_recommendations(
                    user_id, redis_client
                )
                
                # 추천 결과 캐싱
                await redis_client.setex(
                    f"recommendations:{user_id}",
                    3600,  # 1시간 TTL
                    json.dumps(recommendations)
                )
                
                self.logger.info(
                    f"추천 업데이트 완료: User {user_id}, "
                    f"추천 수: {len(recommendations)}"
                )
                
        except Exception as e:
            self.logger.error(f"이벤트 처리 실패: {str(e)}")
    
    async def _update_viewing_history(self, user_id: str, content_id: str, 
                                    metadata: Dict, redis_client):
        """사용자 시청 이력 업데이트"""
        viewing_data = {
            'content_id': content_id,
            'completion_rate': metadata.get('completion_rate', 0),
            'timestamp': datetime.now().isoformat()
        }
        
        # Redis에 시청 이력 저장 (최근 100개 유지)
        await redis_client.lpush(
            f"viewing_history:{user_id}", 
            json.dumps(viewing_data)
        )
        await redis_client.ltrim(f"viewing_history:{user_id}", 0, 99)
    
    async def _generate_recommendations(self, user_id: str, 
                                      redis_client) -> list:
        """
        사용자별 추천 생성 (간소화된 로직)
        실제로는 복잡한 ML 모델을 사용
        """
        # 사용자 시청 이력 조회
        viewing_history = await redis_client.lrange(
            f"viewing_history:{user_id}", 0, 9
        )
        
        if not viewing_history:
            return self._get_popular_content()
        
        # 간단한 협업 필터링 시뮬레이션
        viewed_content_ids = [
            json.loads(item.decode())['content_id'] 
            for item in viewing_history
        ]
        
        # 유사한 사용자들이 본 컨텐츠 추천 (실제로는 ML 모델)
        recommendations = [
            {'content_id': f'rec_{i}', 'score': 0.9 - i * 0.1}
            for i in range(10)
        ]
        
        return recommendations
    
    def _get_popular_content(self) -> list:
        """인기 컨텐츠 반환 (신규 사용자용)"""
        return [
            {'content_id': f'popular_{i}', 'score': 1.0 - i * 0.1}
            for i in range(10)
        ]

class ContentDeliveryService:
    """컨텐츠 전송 서비스 - 시청 패턴 기반 최적화"""
    
    def __init__(self, kafka_config: Dict[str, str]):
        self.consumer = KafkaConsumer(
            'user-view-events',
            bootstrap_servers=kafka_config['brokers'],
            group_id='content-delivery-service',
            value_deserializer=lambda m: json.loads(m.decode('utf-8'))
        )
        self.logger = logging.getLogger(__name__)
    
    async def optimize_content_distribution(self):
        """시청 패턴 기반 컨텐츠 배포 최적화"""
        content_popularity = {}
        
        for message in self.consumer:
            event_data = message.value
            
            if event_data['activity_type'] == 'view':
                content_id = event_data.get('content_id')
                if content_id:
                    # 컨텐츠 인기도 집계
                    content_popularity[content_id] = \
                        content_popularity.get(content_id, 0) + 1
                    
                    # 임계치 도달 시 CDN 배포 최적화
                    if content_popularity[content_id] % 100 == 0:
                        await self._optimize_cdn_distribution(content_id)
    
    async def _optimize_cdn_distribution(self, content_id: str):
        """CDN 배포 최적화"""
        self.logger.info(
            f"CDN 최적화 실행: {content_id} - 인기도 증가로 인한 배포 확장"
        )
        # 실제로는 CDN API 호출하여 복제본 생성/배포

# 사용 예시 및 통합 실행
async def main():
    """메인 실행 함수"""
    
    # 설정
    kafka_config = {
        'brokers': ['localhost:9092']
    }
    redis_config = {
        'host': 'localhost',
        'port': 6379
    }
    
    # 서비스 인스턴스 생성
    user_service = UserService(kafka_config)
    recommendation_service = RecommendationService(kafka_config, redis_config)
    content_service = ContentDeliveryService(kafka_config)
    
    # 시뮬레이션: 사용자 시청 이벤트 발생
    await user_service.handle_user_viewing(
        user_id='user_123',
        content_id='movie_456', 
        duration=7200,  # 2시간
        position=3600   # 1시간 시청
    )
    
    # 추천 서비스와 컨텐츠 배포 서비스는 백그라운드에서 실행
    # 실제 환경에서는 별도 프로세스/컨테이너로 실행
    await asyncio.gather(
        recommendation_service.start_event_processing(),
        content_service.optimize_content_distribution()
    )

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

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

import pika

# 메시지 생산자
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='order_queue')
channel.basic_publish(exchange='', routing_key='order_queue', body='{"order_id": "ORD-12345"}')
connection.close()

# 메시지 소비자
def callback(ch, method, properties, body):
    print(f"주문 처리: {body}")

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='order_queue')
channel.basic_consume(queue='order_queue', 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

# Producer – OrderCancelled 이벤트 발행
from confluent_kafka import Producer
import json

p = Producer({'bootstrap.servers':'localhost:9092'})
event = {'order_id': 42, 'reason':'user_cancel'}
p.produce('OrderCancelled', json.dumps(event).encode('utf-8'))
p.flush()

# Consumer – CompensationSvc
from confluent_kafka import Consumer
import json

c = Consumer({
 'bootstrap.servers':'localhost:9092',
 'group.id':'compensation-svc',
 'auto.offset.reset':'earliest'
})
c.subscribe(['OrderCancelled'])
for msg in c:
    data=json.loads(msg.value())
    # 보상 로직 수행
    print(f"Order {data['order_id']} compensation executed.")
    # 보상 완료 이벤트 발행 생략 가능

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

@Testcontainers
class OrderServiceIntegrationTest {
  @Container
  static RabbitMQContainer rabbit = new RabbitMQContainer("rabbitmq:3.11")
    .withExchange("order.ex", "fanout")
    .withQueue("order.q")
    .withBinding("order.ex", "order.q", "");  // 바인딩 설정 :contentReference[oaicite:28]{index=28}

  @BeforeAll
  static void setup() { /* Spring 설정 조정 */ }

  @Test
  void testOrderFlow() {
    // message publish → consumer logic → DB persistence 검증
  }
}