杰瑞科技汇

Python MySQL连接池如何高效配置与使用?

Of course! Using a connection pool is a critical best practice for any Python application that interacts with a MySQL database. Creating a new database connection for every query is very slow and resource-intensive. A connection pool maintains a set of open connections that can be reused, dramatically improving performance.

Python MySQL连接池如何高效配置与使用?-图1
(图片来源网络,侵删)

Here’s a comprehensive guide on how to use MySQL connection pools in Python, covering the most popular libraries.


Why Use a Connection Pool?

  1. Performance: Opening a new connection involves a TCP handshake, authentication, and server-side setup. This can take milliseconds. A pool gives you a ready-to-use connection in microseconds.
  2. Resource Management: It limits the number of concurrent connections to your database, preventing you from overwhelming the database server and running out of connections.
  3. Stability: Reusing connections can be more stable than constantly creating and destroying them, especially in long-running applications.
  4. Simplicity: It simplifies your code. You don't have to manually create and close connections everywhere. You just "check out" a connection and "check it in" when you're done.

Option 1: The Modern Standard - mysql-connector-python with Pool

This is the official MySQL driver from Oracle. It has a built-in, easy-to-use connection pool.

Step 1: Install the library

pip install mysql-connector-python

Step 2: Create and Use the Pool

The key is to create the pool once when your application starts (e.g., in a global variable or a dedicated module) and then reuse that same pool instance everywhere.

import mysql.connector
from mysql.connector import pooling
import time
# --- 1. Configuration for the Pool ---
dbconfig = {
    "host": "localhost",
    "user": "your_username",
    "password": "your_password",
    "database": "your_database"
}
# --- 2. Create the Connection Pool ---
# pool_name: A name for the pool (optional)
# pool_size: The number of connections to create initially
try:
    connection_pool = pooling.MySQLConnectionPool(
        pool_name="mypool",
        pool_size=5,
        **dbconfig
    )
    print("Connection pool created successfully!")
except mysql.connector.Error as err:
    print(f"Error creating connection pool: {err}")
    exit(1)
# --- 3. Function to get a connection from the pool ---
def get_connection_from_pool():
    """Gets a connection from the pool."""
    try:
        connection_object = connection_pool.get_connection()
        print("Successfully got a connection from the pool.")
        return connection_object
    except mysql.connector.Error as err:
        print(f"Error getting connection from pool: {err}")
        return None
# --- 4. Function to release a connection back to the pool ---
def release_connection_to_pool(connection_object):
    """Releases a connection back to the pool."""
    if connection_object and connection_object.is_connected():
        connection_object.close()
        print("Connection released back to the pool.")
# --- 5. Example Usage in a Function ---
def perform_database_task():
    """A sample function that uses a pooled connection."""
    conn = None
    cursor = None
    try:
        # Get a connection from the pool
        conn = get_connection_from_pool()
        if not conn:
            return
        cursor = conn.cursor(dictionary=True) # dictionary=True for easy row access
        # Execute a query
        cursor.execute("SELECT id, name, email FROM users LIMIT 10;")
        results = cursor.fetchall()
        print("\n--- Query Results ---")
        for row in results:
            print(f"ID: {row['id']}, Name: {row['name']}, Email: {row['email']}")
    except mysql.connector.Error as err:
        print(f"Error during database operation: {err}")
    finally:
        # Ensure the cursor is closed
        if cursor:
            cursor.close()
        # Release the connection back to the pool
        release_connection_to_pool(conn)
# --- 6. Simulate Multiple Tasks ---
print("\n--- Simulating multiple tasks ---")
for i in range(3):
    print(f"\n--- Running Task {i+1} ---")
    perform_database_task()
    time.sleep(1) # Simulate some work
print("\nAll tasks completed.")
# --- 7. Close the Pool (when your application shuts down) ---
print("\nClosing the connection pool...")
connection_pool.close()
print("Connection pool closed.")

Key Points:

Python MySQL连接池如何高效配置与使用?-图2
(图片来源网络,侵删)
  • One Pool, Everywhere: The connection_pool object is created once and reused.
  • get_connection(): Checks out a connection. If all are in use, it waits until one becomes available (by default).
  • connection.close(): This is the magic. It doesn't close the underlying network connection. It simply returns the connection object to the pool for reuse.
  • Error Handling: Always wrap pool operations in try...except blocks to handle cases where the pool is exhausted or a connection is broken.

Option 2: The Popular Alternative - PyMySQL with DBUtils

PyMySQL is another very popular, pure-Python MySQL driver. It doesn't have a built-in pool, so you use a third-party library called DBUtils.

Step 1: Install the libraries

pip install pymysql
pip install DBUtils

Step 2: Create and Use the Pool

The setup is very similar. You create the pool once and reuse it.

import pymysql
from dbutils.pooled_db import PooledDB
import time
# --- 1. Configuration for the Pool ---
dbconfig = {
    "host": "localhost",
    "user": "your_username",
    "password": "your_password",
    "database": "your_database",
    "charset": "utf8mb4"
}
# --- 2. Create the Connection Pool ---
# mincached: Number of connections to keep open
# maxcached: Maximum number of connections to keep open
# maxconnections: Maximum number of connections the pool can have
try:
    connection_pool = PooledDB(
        creator=pymysql,
        mincached=2,
        maxcached=5,
        maxconnections=10,
        **dbconfig
    )
    print("DBUtils connection pool created successfully!")
except Exception as err:
    print(f"Error creating connection pool: {err}")
    exit(1)
# --- 3. Function to get a connection from the pool ---
def get_connection_from_pool():
    """Gets a connection from the DBUtils pool."""
    try:
        connection_object = connection_pool.connection()
        print("Successfully got a connection from the pool.")
        return connection_object
    except Exception as err:
        print(f"Error getting connection from pool: {err}")
        return None
# --- 4. Function to release a connection back to the pool ---
# With DBUtils, the connection is automatically released when it's garbage collected.
# However, it's good practice to explicitly close it.
def release_connection_to_pool(connection_object):
    """Releases a connection back to the pool."""
    if connection_object:
        connection_object.close()
        print("Connection released back to the pool.")
# --- 5. Example Usage in a Function ---
def perform_database_task_pymysql():
    """A sample function that uses a DBUtils pooled connection."""
    conn = None
    cursor = None
    try:
        # Get a connection from the pool
        conn = get_connection_from_pool()
        if not conn:
            return
        cursor = conn.cursor(pymysql.cursors.DictCursor) # DictCursor for easy row access
        # Execute a query
        cursor.execute("SELECT id, name, email FROM users LIMIT 10;")
        results = cursor.fetchall()
        print("\n--- Query Results (PyMySQL) ---")
        for row in results:
            print(f"ID: {row['id']}, Name: {row['name']}, Email: {row['email']}")
    except Exception as err:
        print(f"Error during database operation: {err}")
    finally:
        if cursor:
            cursor.close()
        if conn:
            release_connection_to_pool(conn)
# --- 6. Simulate Multiple Tasks ---
print("\n--- Simulating multiple tasks with PyMySQL ---")
for i in range(3):
    print(f"\n--- Running Task {i+1} ---")
    perform_database_task_pymysql()
    time.sleep(1)
print("\nAll tasks completed.")
# --- 7. Close the Pool (when your application shuts down) ---
# DBUtils pools don't have a simple 'close' method.
# The connections are closed when the pool object is garbage collected.
print("\nDBUtils pool will be garbage collected.")

Best Practices & Common Pitfalls

  1. Create One Pool per Database: If your application connects to multiple databases, create a separate pool for each one.
  2. Create the Pool at Startup: Initialize the connection pool when your application starts (e.g., in a __init__.py file or a main application class). Do not create a new pool for every request or function call.
  3. Set Appropriate Pool Size: The ideal pool size depends on your application's load and your database server's capacity.
    • Too small: Your application will wait for connections, becoming a bottleneck.
    • Too large: You might overwhelm your database server, causing it to run out of memory or hit its connection limit.
    • A good starting point is between 5 and 20. Monitor your application and database to tune it.
  4. ALWAYS Release Connections: This is the most common mistake. Forgetting to release a connection (by not calling close() or not using a finally block) will leak connections. Eventually, your pool will be empty, and your application will hang.
  5. Handle Connection Errors: Connections in a pool can become stale (e.g., if the database server restarts). Your code should handle mysql.connector.Error (or pymysql.Error) and try to get a new connection from the pool if an operation fails.
  6. Use Context Managers (Advanced): For even cleaner code, you can wrap the pool usage in a context manager. Here's a simple example for mysql-connector-python:
# A simple context manager for the mysql-connector-python pool
class PooledConnection:
    def __init__(self, pool):
        self.pool = pool
        self.conn = None
    def __enter__(self):
        self.conn = self.pool.get_connection()
        return self.conn
    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.conn:
            self.conn.close()
# Usage:
# with PooledConnection(connection_pool) as conn:
#     with conn.cursor() as cursor:
#         cursor.execute("...")
#         results = cursor.fetchall()
# The connection is automatically closed when the 'with' block exits.

Summary: Which One to Choose?

Feature mysql-connector-python (with built-in pool) PyMySQL + DBUtils
Ease of Use Very Easy. The pool is built-in and simple. Very Easy. DBUtils is a well-established, simple wrapper.
Dependencies One library (mysql-connector-python). Two libraries (PyMySQL, DBUtils).
Performance Good. It's a C-optimized driver. Good. PyMySQL is pure Python, but DBUtils is very efficient.
Maintenance Officially maintained by Oracle. Community-maintained, but very stable and popular.
Recommendation Excellent choice for most new projects. It's modern, official, and straightforward. A fantastic and equally valid alternative. A great choice if you're already using PyMySQL or prefer a pure-Python solution.

For most new applications, starting with mysql-connector-python and its built-in pool is the recommended path. It's clean, simple, and does exactly what you need.

Python MySQL连接池如何高效配置与使用?-图3
(图片来源网络,侵删)
分享:
扫描分享到社交APP
上一篇
下一篇