Python 数据库连接池

什么是数据库连接池

在介绍数据库连接池之前，我们先来了解一下什么是数据库连接。

数据库连接是指应用程序和数据库之间的一种通信机制。当我们需要从应用程序中访问数据库时，需要建立一个与数据库的连接，然后通过这个连接来执行SQL语句，并得到数据库的返回结果。

数据库连接池是在应用程序启动时创建一些指定数量的数据库连接，并将这些连接保存在一个池子中。当应用程序需要连接数据库时，不再直接去创建一个新的连接，而是从连接池中获取一个已经创建好的连接来使用。在使用完后，连接并不会关闭，而是重新放回连接池中，供其他请求使用。

通过使用数据库连接池，可以避免频繁地创建和销毁数据库连接的开销，提高数据库访问的效率。

Python 中的数据库连接池

在 Python 中，有许多第三方库提供了数据库连接池的功能，如 MySQLdb、pymysql、psycopg2等。

我们将以 pymysql 为例，在 Python 中使用数据库连接池。

安装 pymysql

使用以下命令在 Python 中安装 pymysql：

pip install pymysql

创建数据库连接池

首先，我们需要导入 pymysql 和 DBUtils 模块：

import pymysql
from DBUtils.PooledDB import PooledDB

然后，我们创建一个连接池对象，并在其中指定数据库连接的参数：

pool = PooledDB(
    creator=pymysql,  # 指定数据库的库
    maxconnections=5,  # 连接池允许的最大连接数
    mincached=2,  # 初始化时，连接池至少创建的空闲连接数
    blocking=True,  # 当没有可用连接时，是否阻塞等待
    host='localhost',  # 数据库主机地址
    port=3306,  # 数据库端口号
    user='root',  # 数据库用户名
    password='password',  # 数据库密码
    database='test_db',  # 数据库名称
    charset='utf8'  # 数据库字符集
)

在创建连接池对象时，我们需要传入以下参数：

creator: 指定数据库的库，如 pymysql、psycopg2 等。
maxconnections: 连接池允许的最大连接数。
mincached: 初始化时，连接池至少创建的空闲连接数。
blocking: 当没有可用连接时，是否阻塞等待。
host: 数据库主机地址。
port: 数据库端口号。
user: 数据库用户名。
password: 数据库密码。
database: 数据库名称。
charset: 数据库字符集。

从连接池中获取连接

使用以下方法从连接池中获取一个数据库连接：

conn = pool.connection()

使用连接执行 SQL

获取到数据库连接后，我们就可以使用这个连接来执行 SQL 语句了。

cursor = conn.cursor()

sql = "SELECT * FROM users"
cursor.execute(sql)

result = cursor.fetchall()

for row in result:
    print(row)

cursor.close()

将连接放回连接池

使用完数据库连接后，记得将连接放回连接池供其他请求使用：

conn.close()

完整示例代码

下面是一个完整的示例代码，演示了如何使用 pymysql 连接池来执行数据库操作：

import pymysql
from DBUtils.PooledDB import PooledDB

pool = PooledDB(
    creator=pymysql,
    maxconnections=5,
    mincached=2,
    blocking=True,
    host='localhost',
    port=3306,
    user='root',
    password='password',
    database='test_db',
    charset='utf8'
)

def query_users():
    conn = pool.connection()
    cursor = conn.cursor()

    sql = "SELECT * FROM users"
    cursor.execute(sql)

    result = cursor.fetchall()

    for row in result:
        print(row)

    cursor.close()
    conn.close()

if __name__ == '__main__':
    query_users()