第4章：Python实现连接池：基于queue模块实现简单连接池

连接池这东西，说白了就是「复用」。

我在做量化交易系统时，网关和交易所之间要维持大量TCP连接。每次新建连接都要三次握手，开销大得吓人。更麻烦的是，如果并发上来，系统直接崩给你看。

所以，我们需要一个池子。把连接提前建好，用的时候拿，用完放回去。

4.1 基于queue模块实现简单连接池

Python的queue.Queue是线程安全的。用它做连接池的底层容器，再合适不过。

先看一个最简版本：

import queue
import time
import random

class SimpleConnectionPool:
    def __init__(self, min_conn=2, max_conn=5):
        self._pool = queue.Queue(maxsize=max_conn)
        self._min_conn = min_conn
        self._max_conn = max_conn
        self._created = 0
        
        # 初始化时创建最小连接数
        for _ in range(min_conn):
            self._pool.put(self._create_connection())
            self._created += 1
    
    def _create_connection(self):
        """模拟创建连接"""
        conn_id = f"conn_{self._created}"
        print(f"创建新连接: {conn_id}")
        return conn_id
    
    def acquire(self, timeout=5):
        """获取连接"""
        try:
            conn = self._pool.get(timeout=timeout)
            print(f"获取连接: {conn}")
            return conn
        except queue.Empty:
            # 池子空了，看看能不能新建
            if self._created < self._max_conn:
                conn = self._create_connection()
                self._created += 1
                return conn
            raise Exception("连接池已满，无法获取连接")
    
    def release(self, conn):
        """归还连接"""
        print(f"归还连接: {conn}")
        self._pool.put(conn)

这个版本能跑，但问题不少。我早期项目就吃过亏——连接用完没归还，池子慢慢就空了。

4.2 连接池的线程安全设计

为什么用queue.Queue？因为它内部用了threading.Lock。多个线程同时get/put，不会乱。

但光靠Queue还不够。我们来看看真实场景下的坑：

import threading
import time

class ThreadSafePool:
    def __init__(self, min_conn=2, max_conn=5):
        self._pool = queue.Queue(maxsize=max_conn)
        self._lock = threading.Lock()  # 保护连接计数
        self._created = 0
        self._min_conn = min_conn
        self._max_conn = max_conn
        
        for _ in range(min_conn):
            self._pool.put(self._create_connection())
    
    def _create_connection(self):
        with self._lock:
            conn_id = f"conn_{self._created}"
            self._created += 1
            return conn_id
    
    def acquire(self, timeout=5):
        try:
            conn = self._pool.get(timeout=timeout)
            return conn
        except queue.Empty:
            with self._lock:
                if self._created < self._max_conn:
                    conn = self._create_connection()
                    return conn
            raise Exception("连接池已满")
    
    def release(self, conn):
        self._pool.put(conn)

看到self._lock了吗？这个锁只保护_created计数。为什么？因为Queue自己已经线程安全了，我们只需要保护那些「非线程安全」的状态。

我曾经遇到过一个线上事故：两个线程同时acquire，池子空了，都去创建新连接。没有锁保护计数，结果创建了3个连接，计数只加了1。最后池子爆了。

4.3 连接池的上下文管理器

每次都要手动acquire/release？太容易出错了。Python的上下文管理器就是为这种场景设计的。

from contextlib import contextmanager

class PoolWithContext:
    def __init__(self, min_conn=2, max_conn=5):
        self._pool = queue.Queue(maxsize=max_conn)
        self._lock = threading.Lock()
        self._created = 0
        self._min_conn = min_conn
        self._max_conn = max_conn
        
        for _ in range(min_conn):
            self._pool.put(self._create_connection())
    
    def _create_connection(self):
        with self._lock:
            conn_id = f"conn_{self._created}"
            self._created += 1
            return conn_id
    
    @contextmanager
    def get_connection(self, timeout=5):
        """上下文管理器，自动归还连接"""
        conn = None
        try:
            conn = self._acquire(timeout)
            yield conn
        finally:
            if conn is not None:
                self._release(conn)
    
    def _acquire(self, timeout):
        try:
            return self._pool.get(timeout=timeout)
        except queue.Empty:
            with self._lock:
                if self._created < self._max_conn:
                    return self._create_connection()
            raise Exception("连接池已满")
    
    def _release(self, conn):
        self._pool.put(conn)


# 使用示例
pool = PoolWithContext(min_conn=2, max_conn=5)

def worker(worker_id):
    with pool.get_connection() as conn:
        print(f"工作线程 {worker_id} 使用连接: {conn}")
        time.sleep(random.uniform(0.1, 0.5))
    print(f"工作线程 {worker_id} 已归还连接")

threads = []
for i in range(10):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

用with语句，连接自动归还。就算中间抛出异常，finally块也能保证归还。这就是我为什么说「上下文管理器是连接池的标配」。

嗯，这里要注意一点：连接池不是越大越好。我见过有人把max_conn设成1000，结果网关直接拒绝连接。要根据实际业务和网关限制来调参。

最后，画一张图帮大家理解整个流程：

这张图展示了完整的数据流：工作线程从连接池获取连接，使用完后归还。连接池管理着到交易网关的TCP连接复用。

好了，这一章就到这里。连接池的核心思想其实就三个字：复用、安全、自动。把这三点吃透，你就能写出生产级的连接池了。

---

公众号:蓝海资料掘金营，微信deep3321