Object Pooling

1
2

시스템 구성: 웹 애플리케이션 + 기본 데이터베이스 연결 풀
시나리오: 사용자 요청 → 연결 획득 → 쿼리 실행 → 연결 반환

1
2

시스템 구성: 게임 서버 + 플레이어 세션 풀 (최대 1000개)
시나리오: 접속 요청 → 풀 가용성 확인 → 세션 할당 또는 대기

1
2

시스템 구성: 마이크로서비스 + 스레드 풀 + 로드 밸런서
시나리오: 트래픽 증가 감지 → 풀 크기 자동 확장 → 부하 감소 시 축소

1
2
3
4
5
6
7

Client
   |
   v
Connection Pool
   |
   v
DB Connection1, DB Connection2, ..., DB ConnectionN

  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

import threading
import time
from queue import Queue, Empty
from abc import ABC, abstractmethod
from typing import Optional, Generic, TypeVar

T = TypeVar('T')

class PooledObject(ABC):
    """풀링 가능한 객체의 기본 인터페이스"""
    
    def __init__(self):
        self.is_in_use = False
        self.created_time = time.time()
        self.last_used = time.time()
    
    @abstractmethod
    def reset(self) -> None:
        """객체를 재사용 가능한 상태로 초기화"""
        pass
    
    @abstractmethod
    def validate(self) -> bool:
        """객체가 사용 가능한 상태인지 검증"""
        pass

class GameSession(PooledObject):
    """게임 세션 객체 예시"""
    
    def __init__(self):
        super().__init__()
        self.player_id = None
        self.game_data = {}
        self.connection = None
    
    def reset(self) -> None:
        """세션 데이터 초기화"""
        self.player_id = None
        self.game_data.clear()
        self.connection = None
        self.is_in_use = False
        print(f"세션 초기화 완료")
    
    def validate(self) -> bool:
        """세션 유효성 검증"""
        return not self.is_in_use and self.connection is None
    
    def start_session(self, player_id: str):
        """세션 시작"""
        self.player_id = player_id
        self.is_in_use = True
        self.last_used = time.time()
        print(f"플레이어 {player_id} 세션 시작")

class ObjectPool(Generic[T]):
    """제네릭 객체 풀 구현"""
    
    def __init__(self, factory_func, initial_size: int = 5, max_size: int = 20):
        self.factory_func = factory_func
        self.max_size = max_size
        self.available_objects = Queue(maxsize=max_size)
        self.used_objects = set()
        self.lock = threading.RLock()
        
        # 초기 객체 생성 (풀 예열)
        for _ in range(initial_size):
            obj = self._create_object()
            self.available_objects.put(obj)
        
        print(f"객체 풀 초기화 완료: {initial_size}개 객체 생성")
    
    def _create_object(self) -> T:
        """새 객체 생성"""
        obj = self.factory_func()
        print(f"새 객체 생성: {type(obj).__name__}")
        return obj
    
    def get_object(self, timeout: float = 5.0) -> Optional[T]:
        """풀에서 객체 획득"""
        with self.lock:
            try:
                # 사용 가능한 객체가 있으면 반환
                obj = self.available_objects.get(timeout=timeout)
                
                # 객체 유효성 검증
                if not obj.validate():
                    print("유효하지 않은 객체 감지, 새 객체 생성")
                    obj = self._create_object()
                
                self.used_objects.add(obj)
                obj.is_in_use = True
                print(f"객체 할당 완료. 사용 중인 객체: {len(self.used_objects)}")
                return obj
                
            except Empty:
                # 풀이 비어있고 최대 크기에 도달하지 않았으면 새 객체 생성
                if len(self.used_objects) < self.max_size:
                    obj = self._create_object()
                    self.used_objects.add(obj)
                    obj.is_in_use = True
                    return obj
                else:
                    print(f"풀 크기 한계 도달. 최대 크기: {self.max_size}")
                    return None
    
    def return_object(self, obj: T) -> None:
        """객체를 풀에 반환"""
        with self.lock:
            if obj in self.used_objects:
                self.used_objects.remove(obj)
                
                # 객체 초기화
                obj.reset()
                
                # 풀에 반환
                try:
                    self.available_objects.put_nowait(obj)
                    print(f"객체 반환 완료. 사용 가능한 객체: {self.available_objects.qsize()}")
                except:
                    print("풀이 가득 참. 객체 폐기")
            else:
                print("잘못된 객체 반환 시도")
    
    def get_pool_statistics(self) -> dict:
        """풀 통계 정보 반환"""
        with self.lock:
            return {
                "available_count": self.available_objects.qsize(),
                "used_count": len(self.used_objects),
                "total_count": self.available_objects.qsize() + len(self.used_objects),
                "max_size": self.max_size
            }

class GameServer:
    """게임 서버 시뮬레이션"""
    
    def __init__(self):
        # 게임 세션 풀 생성
        self.session_pool = ObjectPool(
            factory_func=lambda: GameSession(),
            initial_size=3,
            max_size=10
        )
    
    def handle_player_connection(self, player_id: str):
        """플레이어 접속 처리"""
        print(f"\n=== 플레이어 {player_id} 접속 요청 ===")
        
        # 세션 풀에서 세션 획득
        session = self.session_pool.get_object()
        
        if session:
            # 세션 시작
            session.start_session(player_id)
            
            # 게임 플레이 시뮬레이션 (실제로는 게임 로직 실행)
            time.sleep(1)  # 게임 플레이 시간 시뮬레이션
            
            # 세션 종료 및 반환
            print(f"플레이어 {player_id} 게임 종료")
            self.session_pool.return_object(session)
        else:
            print(f"세션 할당 실패: {player_id}")
        
        # 풀 상태 출력
        stats = self.session_pool.get_pool_statistics()
        print(f"풀 상태: {stats}")

# 사용 예시
def main():
    """메인 실행 함수"""
    print("=== Object Pooling 게임 서버 시뮬레이션 ===\n")
    
    # 게임 서버 생성
    game_server = GameServer()
    
    # 여러 플레이어 접속 시뮬레이션
    players = ["Player1", "Player2", "Player3", "Player4", "Player5"]
    
    # 순차적 접속 처리
    for player in players:
        game_server.handle_player_connection(player)
        time.sleep(0.5)
    
    print("\n=== 동시 접속 테스트 ===")
    
    # 멀티스레드로 동시 접속 시뮬레이션
    import concurrent.futures
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        concurrent_players = ["ConcurrentPlayer1", "ConcurrentPlayer2", 
                            "ConcurrentPlayer3", "ConcurrentPlayer4"]
        
        futures = [executor.submit(game_server.handle_player_connection, player) 
                  for player in concurrent_players]
        
        concurrent.futures.wait(futures)
    
    print("\n=== 시뮬레이션 완료 ===")

if __name__ == "__main__":
    main()