Python MySQL 线程池封装类：高效数据库连接管理

以下是一个 Python 连接 MySQL 的线程池封装类的示例代码，使用 Python 内置的线程池模块 concurrent.futures 来管理多个数据库连接。在类的初始化方法中，我们可以设置线程池的大小和数据库连接信息。在使用数据库连接时，我们可以使用 with 语句来获得一个连接对象，然后在 with 语句块中执行需要的操作。当 with 语句块结束时，连接对象会自动回收到线程池中，以供其他线程使用。

import mysql.connector
from mysql.connector import Error
from concurrent.futures import ThreadPoolExecutor


class MySQLPool:
    def __init__(self, pool_name, pool_size, **db_config):
        self.pool_name = pool_name
        self.pool_size = pool_size
        self.db_config = db_config
        self.pool = ThreadPoolExecutor(pool_size)
        self.connections = []

    def __del__(self):
        self.pool.shutdown(wait=False)

    def get_connection(self):
        try:
            connection = self.connections.pop()
        except IndexError:
            connection = mysql.connector.connect(**self.db_config)
        return connection

    def return_connection(self, connection):
        self.connections.append(connection)

    def execute(self, query, args=None):
        with self.pool.submit(self._execute, query, args) as future:
            return future.result()

    def _execute(self, query, args):
        connection = self.get_connection()
        try:
            cursor = connection.cursor()
            cursor.execute(query, args)
            result = cursor.fetchall()
            connection.commit()
        except Error as error:
            connection.rollback()
            raise error
        finally:
            cursor.close()
            self.return_connection(connection)
        return result

这个类的使用方法如下：

db_config = {
    'host': 'localhost',
    'user': 'root',
    'password': 'password',
    'database': 'test_db'
}

pool = MySQLPool('test_pool', 5, **db_config)

result = pool.execute('SELECT * FROM customers')
print(result)

在这个例子中，我们创建了名为 test_pool 的线程池，大小为 5。然后我们执行了一个查询操作，从名为 customers 的表中获取所有数据，并打印结果。注意，我们并没有显式地打开或关闭数据库连接，这些操作都是由 MySQLPool 类自动处理的。