Connection Pooling

 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

# psycopg2의 SimpleConnectionPool을 활용한 PostgreSQL 커넥션 풀 예시

import psycopg2
from psycopg2 import pool

# 커넥션 풀 생성 (최소 1, 최대 10개 커넥션)
connection_pool = pool.SimpleConnectionPool(
    1, 10,
    user="dbuser",
    password="password",
    host="localhost",
    port="5432",
    database="mydb"
)

# 커넥션 할당 및 사용
conn = connection_pool.getconn()
try:
    with conn.cursor() as cur:
        cur.execute("SELECT * FROM users;")
        rows = cur.fetchall()
        print(rows)
finally:
    # 커넥션 반환
    connection_pool.putconn(conn)

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

@Configuration
public class DBConfig {
    @Bean
    public DataSource dataSource() {
        HikariConfig config = new HikariConfig();
        config.setJdbcUrl("jdbc:postgresql://localhost:5432/mydb");
        config.setUsername("user");
        config.setPassword("pass");
        config.setMaximumPoolSize(10);
        config.setIdleTimeout(30000);
        return new HikariDataSource(config);
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

const { Pool } = require('pg')
const pool = new Pool({
  max: 20,
  idleTimeoutMillis: 30000,
  connectionTimeoutMillis: 2000,
})

pool.query('SELECT NOW()', (err, res) => {
  console.log(err, res)
  pool.end()
})

1
2
3
4
5
6
7
8

[databases]
mydb = host=127.0.0.1 port=5432 dbname=mydb

[pgbouncer]
listen_port = 6432
pool_mode = transaction
max_client_conn = 100
default_pool_size = 20

 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

# db_pool.py

import psycopg2
from psycopg2 import pool

# PostgreSQL 연결 풀 생성
connection_pool = psycopg2.pool.SimpleConnectionPool(
    minconn=1,           # 최소 커넥션 수
    maxconn=10,          # 최대 커넥션 수
    user="dbuser",
    password="dbpass",
    host="localhost",
    port="5432",
    database="mydb"
)

# 커넥션 가져오기
def fetch_data():
    try:
        # 커넥션 풀에서 하나 가져오기
        conn = connection_pool.getconn()
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM users;")
        rows = cursor.fetchall()
        print(rows)
    except Exception as e:
        print("DB Error:", e)
    finally:
        # 사용한 커넥션 반환
        if conn:
            cursor.close()
            connection_pool.putconn(conn)

1

초기화 → 연결 생성 → 풀 등록 → 대여 → 반환 → 정리

1
2
3
4
5
6
7
8
9

애플리케이션 시작
    ↓
설정 로드 (최소/최대 크기, 타임아웃 등)
    ↓
최소 연결 수만큼 연결 생성
    ↓
풀에 연결 등록
    ↓
모니터링 스레드 시작

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

연결 요청 수신
    ↓
사용 가능한 연결 검색
    ↓
연결 존재 여부 확인
    ↓
[존재] 연결 대여 → 상태 변경 (IDLE → ACTIVE)
    ↓
[부재] 새 연결 생성 (최대 크기 내) 또는 대기
    ↓
연결 반환

1
2
3
4
5
6
7
8
9

연결 반환 요청
    ↓
연결 상태 검증
    ↓
[정상] 상태 변경 (ACTIVE → IDLE)
    ↓
[비정상] 연결 제거 후 새 연결 생성
    ↓
풀에 연결 등록

  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

import threading
import time
import psycopg2
from contextlib import contextmanager
from queue import Queue, Empty
from typing import Optional

class DatabaseConnectionPool:
    """
    데이터베이스 연결 풀 구현
    - 스레드 안전한 연결 관리
    - 동적 크기 조정 지원
    - 연결 검증 및 자동 복구
    """
    
    def __init__(self, min_connections: int = 5, max_connections: int = 20, 
                 connection_params: dict = None):
        """
        연결 풀 초기화
        
        Args:
            min_connections: 최소 연결 수
            max_connections: 최대 연결 수  
            connection_params: 데이터베이스 연결 파라미터
        """
        self.min_connections = min_connections
        self.max_connections = max_connections
        self.connection_params = connection_params or {}
        
        # 스레드 안전한 연결 큐
        self._available_connections = Queue(maxsize=max_connections)
        self._all_connections = set()
        self._lock = threading.RLock()
        
        # 풀 상태 추적
        self._created_connections = 0
        self._active_connections = 0
        
        # 초기 연결 생성
        self._initialize_pool()
        
        # 백그라운드 관리 스레드 시작
        self._maintenance_thread = threading.Thread(
            target=self._maintain_pool, daemon=True
        )
        self._maintenance_thread.start()
    
    def _create_connection(self):
        """
        새로운 데이터베이스 연결 생성
        
        Returns:
            psycopg2.connection: 데이터베이스 연결 객체
        """
        try:
            connection = psycopg2.connect(**self.connection_params)
            connection.autocommit = False
            return connection
        except Exception as e:
            print(f"연결 생성 실패: {e}")
            return None
    
    def _initialize_pool(self):
        """
        풀 초기화 - 최소 연결 수만큼 미리 생성
        """
        with self._lock:
            for _ in range(self.min_connections):
                conn = self._create_connection()
                if conn:
                    self._available_connections.put(conn)
                    self._all_connections.add(conn)
                    self._created_connections += 1
    
    def _validate_connection(self, connection) -> bool:
        """
        연결 상태 검증
        
        Args:
            connection: 검증할 연결 객체
            
        Returns:
            bool: 연결이 유효한지 여부
        """
        try:
            with connection.cursor() as cursor:
                cursor.execute("SELECT 1")
                return True
        except Exception:
            return False
    
    def _maintain_pool(self):
        """
        백그라운드에서 풀 상태 관리
        - 유휴 연결 정리
        - 최소 연결 수 유지
        - 장애 연결 제거
        """
        while True:
            try:
                time.sleep(30)  # 30초마다 유지보수 실행
                self._cleanup_invalid_connections()
                self._ensure_minimum_connections()
            except Exception as e:
                print(f"풀 유지보수 오류: {e}")
    
    def _cleanup_invalid_connections(self):
        """
        유효하지 않은 연결들을 풀에서 제거
        """
        with self._lock:
            connections_to_remove = []
            temp_connections = []
            
            # 큐에서 모든 연결을 꺼내서 검증
            while not self._available_connections.empty():
                try:
                    conn = self._available_connections.get_nowait()
                    if self._validate_connection(conn):
                        temp_connections.append(conn)
                    else:
                        connections_to_remove.append(conn)
                except Empty:
                    break
            
            # 유효한 연결들을 다시 큐에 넣기
            for conn in temp_connections:
                self._available_connections.put(conn)
            
            # 유효하지 않은 연결들 정리
            for conn in connections_to_remove:
                try:
                    conn.close()
                except:
                    pass
                self._all_connections.discard(conn)
                self._created_connections -= 1
    
    def _ensure_minimum_connections(self):
        """
        최소 연결 수 보장
        """
        with self._lock:
            current_available = self._available_connections.qsize()
            if current_available < self.min_connections:
                needed = self.min_connections - current_available
                for _ in range(needed):
                    if self._created_connections < self.max_connections:
                        conn = self._create_connection()
                        if conn:
                            self._available_connections.put(conn)
                            self._all_connections.add(conn)
                            self._created_connections += 1
    
    @contextmanager
    def get_connection(self, timeout: int = 5):
        """
        연결 대여 (컨텍스트 매니저)
        
        Args:
            timeout: 연결 대기 시간 (초)
            
        Yields:
            psycopg2.connection: 데이터베이스 연결
        """
        connection = None
        try:
            # 연결 획득 시도
            try:
                connection = self._available_connections.get(timeout=timeout)
            except Empty:
                # 사용 가능한 연결이 없고 최대 크기에 도달하지 않은 경우 새 연결 생성
                with self._lock:
                    if self._created_connections < self.max_connections:
                        connection = self._create_connection()
                        if connection:
                            self._all_connections.add(connection)
                            self._created_connections += 1
                    else:
                        raise Exception("연결 풀이 고갈되었습니다")
            
            if not connection:
                raise Exception("연결을 생성할 수 없습니다")
            
            # 연결 검증
            if not self._validate_connection(connection):
                # 유효하지 않은 연결인 경우 새로 생성
                try:
                    connection.close()
                except:
                    pass
                connection = self._create_connection()
                if not connection:
                    raise Exception("유효한 연결을 생성할 수 없습니다")
            
            with self._lock:
                self._active_connections += 1
            
            yield connection
            
        finally:
            # 연결 반환
            if connection:
                with self._lock:
                    self._active_connections -= 1
                
                # 연결을 다시 검증한 후 풀에 반환
                if self._validate_connection(connection):
                    self._available_connections.put(connection)
                else:
                    # 유효하지 않은 연결은 제거
                    try:
                        connection.close()
                    except:
                        pass
                    with self._lock:
                        self._all_connections.discard(connection)
                        self._created_connections -= 1
    
    def get_pool_status(self) -> dict:
        """
        현재 풀 상태 정보 반환
        
        Returns:
            dict: 풀 상태 정보
        """
        with self._lock:
            return {
                'total_connections': self._created_connections,
                'available_connections': self._available_connections.qsize(),
                'active_connections': self._active_connections,
                'max_connections': self.max_connections,
                'min_connections': self.min_connections
            }
    
    def close_all(self):
        """
        모든 연결 종료 및 풀 정리
        """
        with self._lock:
            # 사용 가능한 연결들 종료
            while not self._available_connections.empty():
                try:
                    conn = self._available_connections.get_nowait()
                    conn.close()
                except:
                    pass
            
            # 모든 연결 종료
            for conn in self._all_connections:
                try:
                    conn.close()
                except:
                    pass
            
            self._all_connections.clear()
            self._created_connections = 0
            self._active_connections = 0

# 사용 예시
if __name__ == "__main__":
    # 연결 풀 설정
    connection_params = {
        'host': 'localhost',
        'database': 'mydb',
        'user': 'myuser',
        'password': 'mypassword',
        'port': 5432
    }
    
    # 연결 풀 생성
    pool = DatabaseConnectionPool(
        min_connections=5,
        max_connections=20,
        connection_params=connection_params
    )
    
    # 연결 사용 예시
    def process_order(order_id: int):
        """
        주문 처리 함수 - 연결 풀 사용
        """
        try:
            with pool.get_connection() as conn:
                with conn.cursor() as cursor:
                    # 주문 정보 조회
                    cursor.execute(
                        "SELECT * FROM orders WHERE id = %s", 
                        (order_id,)
                    )
                    order = cursor.fetchone()
                    
                    if order:
                        # 주문 상태 업데이트
                        cursor.execute(
                            "UPDATE orders SET status = %s WHERE id = %s",
                            ('processed', order_id)
                        )
                        conn.commit()
                        print(f"주문 {order_id} 처리 완료")
                    else:
                        print(f"주문 {order_id}를 찾을 수 없습니다")
                        
        except Exception as e:
            print(f"주문 처리 오류: {e}")
    
    # 동시 요청 시뮬레이션
    import concurrent.futures
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        # 100개의 주문을 동시에 처리
        futures = [
            executor.submit(process_order, order_id) 
            for order_id in range(1, 101)
        ]
        
        # 모든 작업 완료 대기
        concurrent.futures.wait(futures)
    
    # 풀 상태 확인
    print("풀 상태:", pool.get_pool_status())
    
    # 풀 정리
    pool.close_all()

1
2
3
4
5
6
7
8

// Hibernate 설정 파일 (hibernate.cfg.xml)
<property name="hibernate.connection.provider_class">
  org.hibernate.connection.C3P0ConnectionProvider
</property>
<property name="hibernate.c3p0.min_size">5</property>
<property name="hibernate.c3p0.max_size">50</property>
<property name="hibernate.c3p0.timeout">300</property>
<property name="hibernate.c3p0.idle_test_period">150</property>

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

# SQLAlchemy 내장 풀 구현 예시
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

class DatabaseManager:
    def __init__(self):
        # 엔진 생성 및 풀 설정
        self.engine = create_engine(
            'mysql://user:password@localhost/mydb',
            pool_size=5,               # 초기 풀 크기
            max_overflow=10,           # 추가 생성 가능한 최대 연결 수
            pool_timeout=30,           # 연결 대기 시간
            pool_recycle=3600,         # 연결 재사용 주기
            echo=True                  # SQL 로깅 활성화
        )
        
        # 세션 팩토리 생성
        self.Session = sessionmaker(bind=self.engine)
    
    def get_session(self):
        return self.Session()

 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

// Sequelize 내장 풀 구현 예시
const { Sequelize } = require('sequelize');

class SequelizeManager {
    constructor() {
        this.sequelize = new Sequelize('mydb', 'user', 'password', {
            host: 'localhost',
            dialect: 'mysql',
            
            // 풀 설정
            pool: {
                max: 5,        // 최대 연결 수
                min: 0,        // 최소 연결 수
                acquire: 30000,// 연결 획득 타임아웃
                idle: 10000    // 유휴 시간
            }
        });
    }
    
    async testConnection() {
        try {
            await this.sequelize.authenticate();
            console.log('데이터베이스 연결 성공');
        } catch (error) {
            console.error('데이터베이스 연결 실패:', error);
        }
    }
}

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

// HikariCP 설정
HikariConfig config = new HikariConfig();
config.setJdbcUrl("jdbc:mysql://localhost:3306/mydb");
config.setUsername("user");
config.setPassword("password");
config.setMaximumPoolSize(20);      // 최대 연결 수
config.setMinimumIdle(5);           // 최소 유휴 연결 수
config.setIdleTimeout(30000);      // 유휴 연결 유지 시간(30초)
config.setConnectionTestQuery("SELECT 1"); // 검증 쿼리

HikariDataSource dataSource = new HikariDataSource(config);

// 연결 사용
try (Connection conn = dataSource.getConnection()) {
    // 데이터베이스 작업 수행
}

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

<!-- context.xml -->
<Resource 
  name="jdbc/myDB"
  auth="Container"
  type="javax.sql.DataSource"
  driverClassName="com.mysql.cj.jdbc.Driver"
  url="jdbc:mysql://localhost:3306/mydb"
  username="user"
  password="password"
  maxTotal="100"          <!-- 최대 연결 수 -->
  maxIdle="20"            <!-- 유휴 연결 최대 수 -->
  minIdle="5"             <!-- 유휴 연결 최소 수 -->
  maxWaitMillis="30000"   <!-- 연결 대기 시간(30초) -->
  validationQuery="SELECT 1"
  testOnBorrow="true"     <!-- 사용 전 연결 검증 -->
/>

  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

import threading
import queue
import time
import mysql.connector
from typing import Optional, List
from dataclasses import dataclass
import logging

@dataclass
class PooledConnection:
    connection: mysql.connector.MySQLConnection
    last_used: float
    created_at: float
    is_busy: bool = False

class ConnectionPool:
    def __init__(
        self,
        host: str,
        user: str,
        password: str,
        database: str,
        min_connections: int = 5,
        max_connections: int = 10,
        connection_timeout: int = 30,
        idle_timeout: int = 300,
        validation_interval: int = 60
    ):
        # 기본 설정 초기화
        self.db_config = {
            "host": host,
            "user": user,
            "password": password,
            "database": database
        }
        self.min_connections = min_connections
        self.max_connections = max_connections
        self.connection_timeout = connection_timeout
        self.idle_timeout = idle_timeout
        self.validation_interval = validation_interval

        # 연결 풀 및 잠금 장치 초기화
        self.pool: queue.Queue = queue.Queue()
        self.active_connections: List[PooledConnection] = []
        self.lock = threading.Lock()
        self.last_validation = time.time()
        
        # 모니터링을 위한 메트릭 초기화
        self.metrics = {
            "total_connections": 0,
            "active_connections": 0,
            "waiting_requests": 0
        }

        # 로깅 설정
        self.logger = logging.getLogger(__name__)
        
        # 초기 연결 생성
        self._initialize_pool()

    def _initialize_pool(self):
        """초기 연결 풀 생성"""
        for _ in range(self.min_connections):
            self._add_connection()

    def _add_connection(self) -> Optional[PooledConnection]:
        """새로운 데이터베이스 연결 생성"""
        try:
            connection = mysql.connector.connect(**self.db_config)
            pooled_conn = PooledConnection(
                connection=connection,
                last_used=time.time(),
                created_at=time.time()
            )
            self.pool.put(pooled_conn)
            self.active_connections.append(pooled_conn)
            self.metrics["total_connections"] += 1
            return pooled_conn
        except Exception as e:
            self.logger.error(f"연결 생성 실패: {str(e)}")
            return None

    def _validate_connection(self, conn: PooledConnection) -> bool:
        """연결 유효성 검사"""
        try:
            conn.connection.ping(reconnect=True)
            return True
        except:
            return False

    def get_connection(self) -> Optional[PooledConnection]:
        """풀에서 연결 획득"""
        self.metrics["waiting_requests"] += 1
        start_time = time.time()

        # 연결 검증 주기 확인
        if time.time() - self.last_validation > self.validation_interval:
            self._validate_all_connections()

        while True:
            try:
                conn = self.pool.get(timeout=self.connection_timeout)
                
                # 연결 상태 검증
                if not self._validate_connection(conn):
                    self._remove_connection(conn)
                    if len(self.active_connections) < self.max_connections:
                        self._add_connection()
                    continue

                conn.is_busy = True
                conn.last_used = time.time()
                self.metrics["active_connections"] += 1
                self.metrics["waiting_requests"] -= 1
                return conn

            except queue.Empty:
                # 타임아웃 발생 시 새 연결 생성 시도
                if len(self.active_connections) < self.max_connections:
                    new_conn = self._add_connection()
                    if new_conn:
                        continue
                
                if time.time() - start_time > self.connection_timeout:
                    self.logger.error("연결 획득 타임아웃")
                    self.metrics["waiting_requests"] -= 1
                    return None

    def release_connection(self, conn: PooledConnection):
        """연결 반환"""
        conn.is_busy = False
        self.metrics["active_connections"] -= 1
        self.pool.put(conn)

    def _validate_all_connections(self):
        """모든 연결의 유효성 검사"""
        with self.lock:
            self.last_validation = time.time()
            for conn in self.active_connections[:]:
                if not self._validate_connection(conn):
                    self._remove_connection(conn)

    def _remove_connection(self, conn: PooledConnection):
        """손상된 연결 제거"""
        try:
            conn.connection.close()
        except:
            pass
        self.active_connections.remove(conn)
        self.metrics["total_connections"] -= 1

    def cleanup(self):
        """오래된 유휴 연결 정리"""
        current_time = time.time()
        with self.lock:
            for conn in self.active_connections[:]:
                if (not conn.is_busy and 
                    current_time - conn.last_used > self.idle_timeout and 
                    len(self.active_connections) > self.min_connections):
                    self._remove_connection(conn)

    def get_metrics(self):
        """풀 상태 메트릭 반환"""
        return {
            **self.metrics,
            "pool_size": len(self.active_connections),
            "available_connections": self.pool.qsize()
        }