Event Broker vs. Message Broker

1
2
3
4
5
6
7

[이벤트 발생 (앱, IoT, 웹)]
        ↓
   [Message Broker (Kafka)]
        ↓
[실시간 처리 엔진 (Flink, Spark Streaming)]
        ↓
   [DB / 알림 시스템 / 추론 API]

1
2
3
4
5
6
7

[로그, 이벤트 수집]
        ↓
[Message Queue (SQS, RabbitMQ)]
        ↓
[Batch Consumer (Lambda, ETL Job)]
        ↓
[Data Lake, RDS, 보고서 시스템]

 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

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

import pika

# 메시지 발신자(Producer) - 주문 메시지 발행
def send_order(message):
    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=message,
                          properties=pika.BasicProperties(delivery_mode=2)) # 메시지 영속성
    print(f"Order sent: {message}")
    connection.close()

# 메시지 수신자(Consumer) - 워커가 메시지를 받아 처리
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

# Kafka Consumer - 주문 이벤트 수신 후 배송 처리
from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'order.created',
    bootstrap_servers='localhost:9092',
    group_id='shipping-service',
    value_deserializer=lambda x: json.loads(x.decode('utf-8'))
)

for message in consumer:
    order = message.value
    print(f"🚚 배송 시작: 주문번호 {order['order_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
 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

# Event Producer - 사용자 행동 이벤트 발행
from kafka import KafkaProducer
import json
import time
from datetime import datetime

class UserEventProducer:
    """사용자 행동 이벤트를 Kafka에 발행하는 프로듀서"""
    
    def __init__(self, bootstrap_servers='localhost:9092'):
        # Kafka 프로듀서 초기화
        self.producer = KafkaProducer(
            bootstrap_servers=bootstrap_servers,
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            # 높은 처리량을 위한 배치 설정
            batch_size=16384,
            linger_ms=10,
            # 데이터 안정성을 위한 ACK 설정
            acks='all'
        )
    
    def send_user_action(self, user_id, action_type, product_id, metadata=None):
        """사용자 행동 이벤트 발행"""
        event = {
            'user_id': user_id,
            'action_type': action_type,  # view, click, purchase
            'product_id': product_id,
            'timestamp': datetime.utcnow().isoformat(),
            'metadata': metadata or {}
        }
        
        # 사용자 ID를 키로 사용하여 파티션 분산
        partition_key = str(user_id)
        
        self.producer.send(
            topic='user-actions',
            key=partition_key.encode('utf-8'),
            value=event
        )
        
        print(f"이벤트 발행: {action_type} by user {user_id}")

# Event Consumer - 실시간 추천 처리
from kafka import KafkaConsumer
import redis
import json

class RecommendationProcessor:
    """실시간 추천 생성 및 업데이트 컨슈머"""
    
    def __init__(self, bootstrap_servers='localhost:9092'):
        # Kafka 컨슈머 초기화
        self.consumer = KafkaConsumer(
            'user-actions',
            bootstrap_servers=bootstrap_servers,
            value_deserializer=lambda m: json.loads(m.decode('utf-8')),
            # 컨슈머 그룹 설정으로 병렬 처리
            group_id='recommendation-processor',
            # 가장 빠른 처리를 위한 오프셋 설정
            auto_offset_reset='latest'
        )
        
        # Redis 연결 (추천 결과 캐싱)
        self.redis_client = redis.Redis(
            host='localhost',
            port=6379,
            decode_responses=True
        )
    
    def process_events(self):
        """이벤트 스트림을 지속적으로 처리"""
        print("실시간 추천 처리 시작…")
        
        for message in self.consumer:
            event = message.value
            user_id = event['user_id']
            action_type = event['action_type']
            product_id = event['product_id']
            
            # 사용자 행동 패턴 업데이트
            self._update_user_profile(user_id, action_type, product_id)
            
            # 실시간 추천 생성
            recommendations = self._generate_recommendations(user_id)
            
            # Redis에 추천 결과 캐싱
            self._cache_recommendations(user_id, recommendations)
            
            print(f"사용자 {user_id}에 대한 추천 업데이트 완료")
    
    def _update_user_profile(self, user_id, action_type, product_id):
        """사용자 프로필 업데이트 (행동 패턴 학습)"""
        profile_key = f"user_profile:{user_id}"
        
        # 행동 가중치 적용
        weights = {'view': 1, 'click': 3, 'purchase': 10}
        weight = weights.get(action_type, 1)
        
        # Redis에서 사용자 프로필 조회 및 업데이트
        current_score = float(
            self.redis_client.hget(profile_key, product_id) or 0
        )
        new_score = current_score + weight
        
        self.redis_client.hset(profile_key, product_id, new_score)
        self.redis_client.expire(profile_key, 86400)  # 24시간 TTL
    
    def _generate_recommendations(self, user_id):
        """협업 필터링 기반 실시간 추천 생성"""
        profile_key = f"user_profile:{user_id}"
        user_preferences = self.redis_client.hgetall(profile_key)
        
        # 간단한 협업 필터링 알고리즘
        recommendations = []
        
        if user_preferences:
            # 선호도가 높은 상품 카테고리 추출
            top_products = sorted(
                user_preferences.items(),
                key=lambda x: float(x[1]),
                reverse=True
            )[:5]
            
            # 유사한 상품 추천 (실제로는 ML 모델 사용)
            for product_id, score in top_products:
                similar_products = self._find_similar_products(product_id)
                recommendations.extend(similar_products)
        
        return list(set(recommendations))[:10]  # 중복 제거 후 상위 10개
    
    def _find_similar_products(self, product_id):
        """유사 상품 찾기 (Mock 구현)"""
        # 실제로는 Vector DB나 ML 모델 사용
        return [f"sim_{product_id}_{i}" for i in range(3)]
    
    def _cache_recommendations(self, user_id, recommendations):
        """추천 결과를 Redis에 캐싱"""
        cache_key = f"recommendations:{user_id}"
        
        self.redis_client.setex(
            cache_key,
            3600,  # 1시간 TTL
            json.dumps(recommendations)
        )

# Event Stream Processing - Kafka Streams 스타일의 처리
class EventStreamProcessor:
    """이벤트 스트림 실시간 집계 및 분석"""
    
    def __init__(self):
        self.redis_client = redis.Redis(decode_responses=True)
        # 윈도우 기반 집계를 위한 메모리 저장소
        self.window_data = {}
    
    def process_windowed_aggregation(self, events):
        """시간 윈도우 기반 이벤트 집계"""
        current_time = int(time.time())
        window_size = 300  # 5분 윈도우
        window_key = current_time // window_size
        
        for event in events:
            product_id = event['product_id']
            action_type = event['action_type']
            
            # 윈도우별 상품 인기도 집계
            metrics_key = f"metrics:{window_key}"
            field_key = f"{product_id}:{action_type}"
            
            self.redis_client.hincrby(metrics_key, field_key, 1)
            self.redis_client.expire(metrics_key, 3600)  # 1시간 보관
        
        # 실시간 트렌딩 상품 업데이트
        self._update_trending_products(window_key)
    
    def _update_trending_products(self, window_key):
        """트렌딩 상품 목록 업데이트"""
        metrics_key = f"metrics:{window_key}"
        metrics = self.redis_client.hgetall(metrics_key)
        
        # 상품별 종합 점수 계산
        product_scores = {}
        for field, count in metrics.items():
            product_id, action_type = field.split(':')
            weight = {'view': 1, 'click': 3, 'purchase': 10}[action_type]
            
            if product_id not in product_scores:
                product_scores[product_id] = 0
            product_scores[product_id] += int(count) * weight
        
        # 상위 트렌딩 상품 저장
        trending = sorted(
            product_scores.items(),
            key=lambda x: x[1],
            reverse=True
        )[:20]
        
        self.redis_client.setex(
            'trending_products',
            1800,  # 30분 TTL
            json.dumps([product_id for product_id, _ in trending])
        )

# 사용 예시
if __name__ == "__main__":
    # 프로듀서로 이벤트 발행
    producer = UserEventProducer()
    producer.send_user_action(
        user_id=12345,
        action_type='click',
        product_id='product_001',
        metadata={'category': 'electronics', 'price': 299.99}
    )
    
    # 컨슈머로 실시간 처리
    processor = RecommendationProcessor()
    processor.process_events()  # 무한 루프로 이벤트 처리