在 Python 中使用连接池提升性能

2024-05-15码农

来源：Python七号

在数据库操作中，频繁地打开和关闭连接会消耗大量的资源和时间，在一些需要高并发的场景，我们就需要连接池技术优化这一问题。本文以 Python 中的 SQLite 数据库为例，介绍如何使用连接池来提升数据操作的效率。

对于使用 SQLite 这样的轻量级数据库，虽然连接的开销相对较小，但在高并发或要求快速响应的应用场景中，优化连接过程仍然非常重要。

import sqlite3 # Connect to database ( 这里会很耗时 ) conn = sqlite3.connect('mydatabase.db') # Create a cursor object cursor = conn.cursor() # Create a table (if it doesn't exist) cursor.execute('''sql statement''')

首先，我们来理解一下什么是连接池。在数据库操作中，每次访问数据库都需要和数据库服务器建立一个连接（ conn = sqlite3.connect('mydatabase.db') ），这个过程包括创建连接对象、与数据库建立通信、验证等多个步骤。连接池，顾名思义，就是预创建并存储一系列连接对象，当程序需要与数据库交互时，直接从池中取用，用完了再放回池中，避免了频繁创建和销毁连接的开销。

使用数据库连接池的步骤：

安装必要的 Python 库

首先确保已经安装了 sqlite3 模块，虽然 SQLite 在 Python 标准库中自带了 sqlite3 接口，但是没有实现连接池的功能，为此我们需要自行编写连接池，或使用第三方库，例如 sqlalchemy . 通过 pip 可以轻松安装 :

pip install sqlalchemy

接下来分别介绍这两种方法。

自行编写连接池

自行编写连接池，可以更好的理解连接池到底做了什么。

import sqlite3 from queue import Queue classSQLiteConnectionPool: """ SQLite 连接池类，用于管理数据库连接，避免频繁创建和关闭连接的开销。 """ def__init__(self, db_path, max_connections=10): """ 初始化连接池。 Args: db_path (str): 数据库文件路径。 max_connections (int): 连接池允许的最大连接数，默认为 10。 """ self.db_path = db_path # 数据库文件路径 self.max_connections = max_connections # 最大连接数 self.free_connections = Queue(maxsize=max_connections) # 存储空闲连接的队列 defget_connection(self): """ 从连接池获取一个数据库连接。 Returns: sqlite3.Connection: 一个 SQLite 数据库连接对象。 """ if self.free_connections.empty(): # 如果队列为空，则创建新的连接 conn = sqlite3.connect(self.db_path) else: # 否则从队列中获取一个空闲连接 conn = self.free_connections.get() return conn defput_connection(self, conn): """ 将一个数据库连接放回连接池。 Args: conn (sqlite3.Connection): 要放回的数据库连接对象。 """ self.free_connections.put(conn) # 将连接放回队列

代码解释 :

SQLiteConnectionPool 类 : 定义了连接池类，用于管理数据库连接。

构造函数 : 接收 db_path ( 数据库文件路径 ) 和 max_connections ( 最大连接数 ) 参数。创建了一个 Queue 对象来存储空闲连接，最大容量为 max_connections。

get_connection() 方法 : 尝试从 free_connections 队列获取一个空闲连接。如果队列为空，则创建一个新的连接。返回获取到的连接。

put_connection() 方法 : 将用完的连接放回 free_connections 队列，以便后续复用。

该连接池的使用方法：

pool = SQLiteConnectionPool('yoursqlite.db') defconcurrent_access(pool): conn = pool.get_connection() cur = conn.cursor() # 参数化查询 cur.execute("sql") rows = cur.fetchall() pool.put_connection(conn) return rows

函数 concurrent_access 可以在高频场景下调用，本质上，连接池相当于一个全局变量。

使用 sqlalchemy 创建连接池

使用 sqlalchemy 的 create_engine 函数，我们可以创建一个 SQLite 连接池。

from sqlalchemy import create_engine engine = create_engine('sqlite:///my_database.db', pool_size=10)

使用方法也比较简单 :

from sqlalchemy import create_engine,text engine = create_engine('sqlite:///my_database.db', pool_size=10) defconcurrent_access(): with engine.connect() as conn: result = conn.execute(text("select * from reply")) for row in result: print(row)