• SQLite 自定义连接池
• 使用 sqlalchemy 创建连接池

SQLite 自定义连接池

import sqlite3
from queue import Queue


class SQLiteConnectionPool:

    def __init__(self, db_path, max_connection=10):
        self.db_path = db_path  # 数据库文件路径
        self.max_connection = max_connection  # 最大连接数
        self.free_connections = Queue(maxsize=max_connection)  # 存储空闲连接的队列

    def get_connection(self):
        """ 从连接池获取一个数据库连接"""

        if self.free_connections.empty():
            conn = sqlite3.connect(self.db_path)
        else:
            conn = self.free_connections.get()
        return conn

    def put_connection(self, conn):
        """将一个数据库连接放回连接池"""
        self.free_connections.put(conn)

该连接池的使用方法：

pool = SQLiteConnectionPool('douban.db')


def access(pool: SQLiteConnectionPool):
    conn = pool.get_connection()
    cur = conn.cursor()
    cur.execute("select * from douban limit 10")

    for item in cur.fetchall():
        print(item)

    pool.put_connection(conn) # 将连接返回连接池

使用 sqlalchemy 创建连接池

from sqlalchemy import create_engine, text

def access():
    engine = create_engine('sqlite:///douban.db', pool_size=10)
    with engine.connect() as conn:
        result = conn.execute(text('select * from douban limit 10'))
        for item in result:
            print(item)

通过使用连接池，我们有效地减少了连接数据库的开销，并且使数据库连接变得更简单、更稳定。尤其是在Web 应用 和数据密集型服务中，使用连接池技术可以大大提高性能和用户体验