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

import sqlite3from 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())