告别数据库瓶颈！用这个技巧让你的程序跑得飞快！

import sqlite3
from queue import Queue

class SqliteConnectionPool:
    def __init__(self, db_path: str, max_connections: int = 10):
        # 初始化数据库文件路径和最大连接数
        self.db_path = db_path
        self.max_connections = max_connections
        # 使用队列来存储空闲连接
        self.free_connections = Queue(maxsize=max_connections)

    def get_connection(self):
        # 如果连接池为空，则创建一个新连接
        if self.free_connections.empty():
            self.create_connection()
        # 从连接池中获取一个连接
        return self.free_connections.get()

    def create_connection(self):
        # 创建一个新的数据库连接
        conn = sqlite3.connect(self.db_path)
        # 将连接放入连接池
        self.free_connections.put(conn)

    def close_connection(self, conn):
        # 将连接归还到连接池
        self.free_connections.put(conn)

# 创建连接池实例
sqlite_pool = SqliteConnectionPool(db_path='./alex.db', max_connections=10)

def concurrent_access():
    try:
        # 从连接池中获取一个连接
        conn = sqlite_pool.get_connection()
        cursor = conn.cursor()
        # 执行数据库查询操作
        cursor.execute('SELECT * FROM students')
        result = cursor.fetchall()
        return result
    except Exception as e:
        print(f"Error: {e}")
        return None
    finally:
        # 关闭连接（实际是归还到连接池）
        sqlite_pool.close_connection(conn)

if __name__ == "__main__":
    # 多次访问数据库，模拟高并发操作
    for _ in range(10):
        print(concurrent_access())

from sqlalchemy import create_engine, text

# 创建数据库引擎，并指定连接池大小
engine = create_engine('sqlite:///alex.db', pool_size=10)

def concurrent_access():
    try:
        # 使用 with 语句自动处理连接的回收
        with engine.connect() as conn:
            result = conn.execute(text('SELECT * FROM students'))
            return result.fetchall()
    except Exception as e:
        print(f"Error: {e}")
        return None

if __name__ == "__main__":
    # 多次访问数据库，模拟高并发操作
    for _ in range(4):
        print(concurrent_access())