분산 시스템(Distributed Systems)

1
2

If there is a Partition (P), then choose Availability (A) or Consistency (C),  
Else (E), choose Latency (L) or Consistency (C)

  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

# Netflix Eureka 기반 서비스 디스커버리 구현 예시
import requests
import json
from typing import List, Dict

class ServiceRegistry:
    """
    서비스 레지스트리: 마이크로서비스 등록 및 발견 관리
    분산 시스템에서 서비스 간 통신을 위한 핵심 컴포넌트
    """
    
    def __init__(self, eureka_url: str):
        self.eureka_url = eureka_url
        self.services = {}  # 로컬 캐시
        
    def register_service(self, service_name: str, instance_info: Dict):
        """
        서비스 인스턴스를 레지스트리에 등록
        - 서비스 가용성 보장을 위한 기본 기능
        """
        registration_data = {
            "instance": {
                "instanceId": f"{service_name}-{instance_info['port']}",
                "hostName": instance_info['host'],
                "app": service_name.upper(),
                "ipAddr": instance_info['ip'],
                "port": {"$": instance_info['port'], "@enabled": "true"},
                "securePort": {"$": 443, "@enabled": "false"},
                "status": "UP",  # 서비스 상태
                "healthCheckUrl": f"http://{instance_info['host']}:{instance_info['port']}/health",
                "statusPageUrl": f"http://{instance_info['host']}:{instance_info['port']}/status"
            }
        }
        
        response = requests.post(
            f"{self.eureka_url}/apps/{service_name}",
            json=registration_data,
            headers={"Content-Type": "application/json"}
        )
        return response.status_code == 204
    
    def discover_service(self, service_name: str) -> List[Dict]:
        """
        서비스 인스턴스 목록 조회
        - 로드 밸런싱을 위한 인스턴스 정보 제공
        """
        try:
            response = requests.get(f"{self.eureka_url}/apps/{service_name}")
            if response.status_code == 200:
                app_data = response.json()
                instances = app_data['application']['instance']
                
                # 활성 인스턴스만 필터링
                active_instances = [
                    {
                        'host': instance['hostName'],
                        'port': instance['port']['$'],
                        'status': instance['status'],
                        'health_check': instance['healthCheckUrl']
                    }
                    for instance in instances
                    if instance['status'] == 'UP'
                ]
                
                # 로컬 캐시 업데이트 (네트워크 장애 대비)
                self.services[service_name] = active_instances
                return active_instances
                
        except requests.RequestException:
            # 네트워크 장애 시 캐시된 데이터 반환
            return self.services.get(service_name, [])
    
    def health_check(self, service_name: str) -> bool:
        """
        서비스 헬스 체크
        - 장애 감지 및 자동 복구를 위한 모니터링
        """
        instances = self.discover_service(service_name)
        for instance in instances:
            try:
                response = requests.get(instance['health_check'], timeout=5)
                if response.status_code == 200:
                    return True
            except requests.RequestException:
                continue
        return False

class LoadBalancer:
    """
    로드 밸런서: 요청을 여러 서비스 인스턴스에 분산
    분산 시스템의 성능과 가용성 향상을 위한 핵심 기능
    """
    
    def __init__(self, service_registry: ServiceRegistry):
        self.service_registry = service_registry
        self.round_robin_index = {}
        
    def get_instance(self, service_name: str, strategy: str = "round_robin") -> Dict:
        """
        로드 밸런싱 전략에 따른 서비스 인스턴스 선택
        """
        instances = self.service_registry.discover_service(service_name)
        
        if not instances:
            raise Exception(f"No available instances for service: {service_name}")
            
        if strategy == "round_robin":
            # 라운드 로빈: 순차적으로 인스턴스 선택
            current_index = self.round_robin_index.get(service_name, 0)
            selected_instance = instances[current_index]
            self.round_robin_index[service_name] = (current_index + 1) % len(instances)
            
        elif strategy == "random":
            # 랜덤: 무작위 인스턴스 선택
            import random
            selected_instance = random.choice(instances)
            
        return selected_instance

class CircuitBreaker:
    """
    회로 차단기: 장애 전파 방지를 위한 패턴
    분산 시스템의 내결함성 보장을 위한 핵심 메커니즘
    """
    
    def __init__(self, failure_threshold: int = 5, timeout: int = 60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN
        
    def call(self, func, *args, **kwargs):
        """
        서비스 호출 시 회로 차단기 적용
        """
        if self.state == "OPEN":
            if self._should_attempt_reset():
                self.state = "HALF_OPEN"
            else:
                raise Exception("Circuit breaker is OPEN")
                
        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
            
        except Exception as e:
            self._on_failure()
            raise e
            
    def _on_success(self):
        """성공 시 상태 초기화"""
        self.failure_count = 0
        self.state = "CLOSED"
        
    def _on_failure(self):
        """실패 시 카운터 증가 및 상태 변경"""
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.failure_count >= self.failure_threshold:
            self.state = "OPEN"
            
    def _should_attempt_reset(self) -> bool:
        """재시도 가능 여부 확인"""
        import time
        return (time.time() - self.last_failure_time) >= self.timeout

# 사용 예시
if __name__ == "__main__":
    # 서비스 레지스트리 초기화
    registry = ServiceRegistry("http://eureka-server:8761/eureka")
    
    # 사용자 서비스 등록
    user_service_info = {
        'host': 'user-service-host',
        'ip': '192.168.1.10',
        'port': 8080
    }
    registry.register_service("user-service", user_service_info)
    
    # 로드 밸런서로 서비스 인스턴스 선택
    load_balancer = LoadBalancer(registry)
    instance = load_balancer.get_instance("user-service")
    
    # 회로 차단기로 안전한 서비스 호출
    circuit_breaker = CircuitBreaker()
    
    def call_user_service():
        """사용자 서비스 호출 함수"""
        url = f"http://{instance['host']}:{instance['port']}/users"
        response = requests.get(url, timeout=5)
        return response.json()
    
    try:
        result = circuit_breaker.call(call_user_service)
        print(f"Service response: {result}")
    except Exception as e:
        print(f"Service call failed: {e}")

 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

# order_service.py
from fastapi import FastAPI
from aiokafka import AIOKafkaProducer
import uvicorn

app = FastAPI()
producer = None

@app.on_event("startup")
async def start_kafka():
    global producer
    producer = AIOKafkaProducer(bootstrap_servers="localhost:9092")
    await producer.start()

@app.post("/order")
async def create_order(order: dict):
    await producer.send_and_wait("order-created", json.dumps(order).encode())
    return {"status": "order received"}

@app.on_event("shutdown")
async def stop_kafka():
    await producer.stop()

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

 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

# inventory_service.py
from aiokafka import AIOKafkaConsumer, AIOKafkaProducer
import asyncio

async def consume():
    consumer = AIOKafkaConsumer(
        "order-created",
        bootstrap_servers="localhost:9092",
        group_id="inventory_group"
    )
    producer = AIOKafkaProducer(bootstrap_servers="localhost:9092")
    await consumer.start()
    await producer.start()
    try:
        async for msg in consumer:
            order = json.loads(msg.value.decode())
            # 재고 처리 로직
            result = check_stock(order)
            topic = "stock-checked" if result else "stock-failed"
            await producer.send_and_wait(topic, msg.value)
    finally:
        await consumer.stop()
        await producer.stop()

def check_stock(order):
    # 단순 재고 로직 예시
    return True

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

  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

"""
분산 스트리밍 서비스 구현 예시 - Netflix 스타일 마이크로서비스
이 코드는 분산 시스템의 핵심 패턴들을 보여줍니다.
"""

import asyncio
import json
import random
import time
from typing import Dict, List, Optional
from dataclasses import dataclass
from enum import Enum
import hashlib

# ==============================================================================
# 데이터 모델 정의
# ==============================================================================

@dataclass
class User:
    user_id: str
    email: str
    subscription_type: str
    viewing_history: List[str]

@dataclass
class Content:
    content_id: str
    title: str
    duration: int
    genre: str
    rating: float

@dataclass
class ViewingSession:
    session_id: str
    user_id: str
    content_id: str
    start_time: float
    current_position: int
    status: str

class ServiceHealth(Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    DOWN = "down"

# ==============================================================================
# 서킷 브레이커 패턴 구현
# ==============================================================================

class CircuitBreaker:
    """서킷 브레이커 패턴 - 장애 전파 방지"""
    
    def __init__(self, failure_threshold: int = 5, timeout: int = 60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failure_count = 0
        self.last_failure_time = 0
        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN
    
    async def call(self, func, *args, **kwargs):
        """서킷 브레이커를 통한 함수 호출"""
        if self.state == "OPEN":
            if time.time() - self.last_failure_time > self.timeout:
                self.state = "HALF_OPEN"
            else:
                raise Exception("서킷 브레이커가 열린 상태입니다")
        
        try:
            result = await func(*args, **kwargs)
            if self.state == "HALF_OPEN":
                self.state = "CLOSED"
                self.failure_count = 0
            return result
        
        except Exception as e:
            self.failure_count += 1
            if self.failure_count >= self.failure_threshold:
                self.state = "OPEN"
                self.last_failure_time = time.time()
            raise e

# ==============================================================================
# 분산 캐시 시스템
# ==============================================================================

class DistributedCache:
    """분산 캐시 구현 - 일관된 해싱 사용"""
    
    def __init__(self, nodes: List[str]):
        self.nodes = nodes
        self.cache_data = {node: {} for node in nodes}
        self.virtual_nodes = 100  # 가상 노드 수
        self.ring = self._build_ring()
    
    def _build_ring(self) -> Dict[int, str]:
        """일관된 해싱 링 구축"""
        ring = {}
        for node in self.nodes:
            for i in range(self.virtual_nodes):
                key = f"{node}:{i}"
                hash_value = int(hashlib.md5(key.encode()).hexdigest(), 16)
                ring[hash_value] = node
        return dict(sorted(ring.items()))
    
    def _get_node(self, key: str) -> str:
        """키에 해당하는 노드 찾기"""
        hash_value = int(hashlib.md5(key.encode()).hexdigest(), 16)
        for ring_key in sorted(self.ring.keys()):
            if hash_value <= ring_key:
                return self.ring[ring_key]
        # 링의 끝에 도달하면 첫 번째 노드 반환
        return self.ring[min(self.ring.keys())]
    
    async def get(self, key: str) -> Optional[any]:
        """캐시에서 값 조회"""
        node = self._get_node(key)
        await asyncio.sleep(0.01)  # 네트워크 지연 시뮬레이션
        return self.cache_data[node].get(key)
    
    async def set(self, key: str, value: any, ttl: int = 300):
        """캐시에 값 저장"""
        node = self._get_node(key)
        await asyncio.sleep(0.01)  # 네트워크 지연 시뮬레이션
        self.cache_data[node][key] = {
            'value': value,
            'expires': time.time() + ttl
        }
        print(f"캐시 저장: {key} -> 노드 {node}")

# ==============================================================================
# 마이크로서비스 구현
# ==============================================================================

class BaseService:
    """기본 서비스 클래스"""
    
    def __init__(self, service_name: str):
        self.service_name = service_name
        self.health = ServiceHealth.HEALTHY
        self.circuit_breaker = CircuitBreaker()
    
    async def health_check(self) -> Dict:
        """헬스체크 엔드포인트"""
        return {
            "service": self.service_name,
            "status": self.health.value,
            "timestamp": time.time()
        }

class UserService(BaseService):
    """사용자 관리 서비스"""
    
    def __init__(self, cache: DistributedCache):
        super().__init__("user-service")
        self.cache = cache
        self.users = {
            "user1": User("user1", "user1@example.com", "premium", ["movie1", "movie2"]),
            "user2": User("user2", "user2@example.com", "basic", ["movie1"])
        }
    
    async def get_user(self, user_id: str) -> Optional[User]:
        """사용자 정보 조회 (캐시 우선)"""
        # 캐시에서 먼저 조회
        cached_user = await self.cache.get(f"user:{user_id}")
        if cached_user:
            print(f"캐시에서 사용자 조회: {user_id}")
            return User(**cached_user['value'])
        
        # 데이터베이스에서 조회 (시뮬레이션)
        await asyncio.sleep(0.1)  # DB 조회 지연 시뮬레이션
        user = self.users.get(user_id)
        
        if user:
            # 캐시에 저장
            await self.cache.set(f"user:{user_id}", user.__dict__)
            print(f"데이터베이스에서 사용자 조회: {user_id}")
        
        return user

class RecommendationService(BaseService):
    """추천 서비스"""
    
    def __init__(self, user_service: UserService, cache: DistributedCache):
        super().__init__("recommendation-service")
        self.user_service = user_service
        self.cache = cache
        self.content_catalog = [
            Content("movie1", "액션 영화", 120, "액션", 4.5),
            Content("movie2", "로맨스 영화", 105, "로맨스", 4.2),
            Content("movie3", "SF 영화", 140, "SF", 4.7),
            Content("movie4", "코미디 영화", 95, "코미디", 4.0)
        ]
    
    async def get_recommendations(self, user_id: str) -> List[Content]:
        """개인화된 추천 목록 생성"""
        try:
            # 서킷 브레이커를 통한 사용자 서비스 호출
            user = await self.circuit_breaker.call(
                self.user_service.get_user, user_id
            )
            
            if not user:
                return []
            
            # 캐시된 추천 확인
            cache_key = f"recommendations:{user_id}"
            cached_recs = await self.cache.get(cache_key)
            if cached_recs:
                print(f"캐시된 추천 목록 반환: {user_id}")
                return [Content(**content) for content in cached_recs['value']]
            
            # 추천 알고리즘 실행 (간단한 예시)
            await asyncio.sleep(0.2)  # ML 모델 추론 시간 시뮬레이션
            
            viewed_content = set(user.viewing_history)
            recommendations = [
                content for content in self.content_catalog
                if content.content_id not in viewed_content
            ]
            
            # 평점 기준 정렬
            recommendations.sort(key=lambda x: x.rating, reverse=True)
            
            # 상위 3개만 반환
            top_recommendations = recommendations[:3]
            
            # 캐시에 저장 (5분 TTL)
            await self.cache.set(
                cache_key, 
                [rec.__dict__ for rec in top_recommendations], 
                300
            )
            
            print(f"새로운 추천 목록 생성: {user_id}")
            return top_recommendations
            
        except Exception as e:
            print(f"추천 서비스 오류: {e}")
            # 폴백: 인기 콘텐츠 반환
            return self.content_catalog[:2]

class ViewingService(BaseService):
    """시청 관리 서비스"""
    
    def __init__(self):
        super().__init__("viewing-service")
        self.active_sessions = {}
    
    async def start_viewing(self, user_id: str, content_id: str) -> str:
        """시청 세션 시작"""
        session_id = f"session_{user_id}_{content_id}_{int(time.time())}"
        session = ViewingSession(
            session_id=session_id,
            user_id=user_id,
            content_id=content_id,
            start_time=time.time(),
            current_position=0,
            status="playing"
        )
        
        self.active_sessions[session_id] = session
        print(f"시청 세션 시작: {session_id}")
        return session_id
    
    async def update_position(self, session_id: str, position: int):
        """재생 위치 업데이트"""
        if session_id in self.active_sessions:
            self.active_sessions[session_id].current_position = position
            print(f"재생 위치 업데이트: {session_id} -> {position}초")
    
    async def stop_viewing(self, session_id: str):
        """시청 세션 종료"""
        if session_id in self.active_sessions:
            self.active_sessions[session_id].status = "stopped"
            print(f"시청 세션 종료: {session_id}")

# ==============================================================================
# API 게이트웨이
# ==============================================================================

class APIGateway:
    """API 게이트웨이 - 요청 라우팅 및 로드 밸런싱"""
    
    def __init__(self):
        self.services = {}
        self.request_count = 0
    
    def register_service(self, service_name: str, service_instance):
        """서비스 등록"""
        if service_name not in self.services:
            self.services[service_name] = []
        self.services[service_name].append(service_instance)
        print(f"서비스 등록: {service_name}")
    
    def get_service_instance(self, service_name: str):
        """라운드 로빈 로드 밸런싱"""
        if service_name not in self.services or not self.services[service_name]:
            return None
        
        instances = self.services[service_name]
        instance = instances[self.request_count % len(instances)]
        self.request_count += 1
        return instance
    
    async def route_request(self, service_name: str, method: str, *args, **kwargs):
        """요청 라우팅"""
        instance = self.get_service_instance(service_name)
        if not instance:
            raise Exception(f"서비스를 찾을 수 없습니다: {service_name}")
        
        if not hasattr(instance, method):
            raise Exception(f"메서드를 찾을 수 없습니다: {method}")
        
        return await getattr(instance, method)(*args, **kwargs)

# ==============================================================================
# 분산 시스템 시뮬레이션
# ==============================================================================

async def simulate_distributed_streaming_system():
    """분산 스트리밍 시스템 시뮬레이션"""
    print("=== 분산 스트리밍 시스템 시작 ===\n")
    
    # 분산 캐시 초기화 (3개 노드)
    cache = DistributedCache(["cache-node-1", "cache-node-2", "cache-node-3"])
    
    # 마이크로서비스 인스턴스 생성
    user_service = UserService(cache)
    recommendation_service = RecommendationService(user_service, cache)
    viewing_service = ViewingService()
    
    # API 게이트웨이 설정
    gateway = APIGateway()
    gateway.register_service("user", user_service)
    gateway.register_service("recommendation", recommendation_service)
    gateway.register_service("viewing", viewing_service)
    
    # 사용자 요청 시뮬레이션
    user_id = "user1"
    
    print("1. 사용자 정보 조회")
    user = await gateway.route_request("user", "get_user", user_id)
    print(f"사용자: {user.email}, 구독: {user.subscription_type}\n")
    
    print("2. 개인화된 추천 목록 요청")
    recommendations = await gateway.route_request(
        "recommendation", "get_recommendations", user_id
    )
    print("추천 목록:")
    for rec in recommendations:
        print(f"  - {rec.title} (평점: {rec.rating})")
    print()
    
    print("3. 동일한 추천 요청 (캐시 히트 예상)")
    recommendations = await gateway.route_request(
        "recommendation", "get_recommendations", user_id
    )
    print()
    
    print("4. 시청 세션 시작")
    session_id = await gateway.route_request(
        "viewing", "start_viewing", user_id, "movie3"
    )
    print()
    
    print("5. 재생 위치 업데이트 시뮬레이션")
    for position in [30, 60, 90]:
        await gateway.route_request(
            "viewing", "update_position", session_id, position
        )
        await asyncio.sleep(0.5)
    print()
    
    print("6. 시청 세션 종료")
    await gateway.route_request("viewing", "stop_viewing", session_id)
    print()
    
    print("7. 시스템 헬스체크")
    for service_name, instances in gateway.services.items():
        for instance in instances:
            health = await instance.health_check()
            print(f"  - {health['service']}: {health['status']}")
    
    print("\n=== 분산 스트리밍 시스템 종료 ===")

# 실행
if __name__ == "__main__":
    asyncio.run(simulate_distributed_streaming_system())