Sample Code Interpretation

Connection Pool

In the sample code, the MySQL connection pool (DBUtils.PooledDB) is created, the built-in connection keepalive mechanism is provided, and the maximum number of connections (max_connections) and connection keepalive interval (keepalive_interval) are configured. The code snippet is as follows:

POOL_CONFIG = { # Connection pool configuration
        'max_connections': 5,           # Maximum number of connections
        'keepalive_interval': 60,       # Connection keepalive interval (s)
        'max_retries': 3,               # Maximum number of retries
        'retry_delay': 1                # Retry interval (s)
}

class MySQLConnectionPool:
    def __init__(self, context, pool_config, db_config):
        """
        Initialize the database connection pool
        :param db_config: database configuration
        :param pool_config: connection pool configuration
        """
        self.context = context
        self.logger = context.getLogger();

        self.db_config = db_config
        self.pool_config = pool_config
        self.pool = self._create_pool()
        self.last_keepalive_time = 0

    def _create_pool(self):
        """
        Create a connection pool
        :return: connection pool object
        """
        try:
            pool = PooledDB(
                creator=pymysql,
                maxconnections=self.pool_config['max_connections'],
                mincached=1,
                **self.db_config
            )
            return pool
        except Exception as e:
            self.logger.error(f"Failed to create connection pool: {e}")
            raise

    def _get_connection(self):
        """
        Obtain a connection from the connection pool and ensure that the connection is valid
        :return: database connection object
        """
        conn = self.pool.connection()
        if not self._is_connection_alive(conn):
            conn = self.pool.connection()
        return conn

    def _is_connection_alive(self, conn):
        """
        Check whether the connection is alive
        :param conn: database connection object
        :return: bool
        """
        try:
            with conn.cursor() as cursor:
                cursor.execute("SELECT 1")
                return True
        except Exception as e:
            self.logger.warning(f"Connection is not alive: {e}")
            return False

    def _close_connection(self, conn):
        """
        Closes the connection
        :param conn: database connection object
        """
        try:
            conn.close()
            self.logger.info("Connection closed")
        except Exception as e:
            self.logger.error(f"Failed to close connection: {e}")

    def _execute_query(self, conn, sql, params=None):
        """
        Query the database
        :param conn: database connection object
        :param sql: SQL statement
        :param params: SQL parameter
        :return: query result
        """
        try:
            with conn.cursor() as cursor:
                cursor.execute(sql, params)
                if sql.strip().lower().startswith('select'):
                    return cursor.fetchall()
                return None
        except Exception as e:
            self.logger.error(f"Query failed: {e}")
            raise

    def _execute_write(self, conn, sql, params=None):
        """
        Perform write operations (insert, update, and delete)
        :param conn: database connection object
        :param sql: SQL statement
        :param params: SQL parameter
        :return: number of affected rows
        """
        try:
            with conn.cursor() as cursor:
                cursor.execute(sql, params)
                conn.commit()
                return cursor.rowcount
        except Exception as e:
            self.logger.error(f"Write operation failed: {e}")
            conn.rollback()
            raise

Reusing created connections with the MySQL connection pool can improve program performance. The maximum number of connections ensures that connection resources are used within a controllable range and thread security is ensured.

Related concepts:

Retry

Build an automatic retry mechanism using decorators to ensure that MySQL operations are retried for a specified number of times after a failure. This significantly reduces the impact of temporary faults. For example, if the service is temporarily unavailable or the invoking times out due to instantaneous network jitter or disk jitter, the mechanism can improve the MySQL operation success rate.

The code snippet is as follows:

POOL_CONFIG = { # Connection pool configuration
        'max_connections': 5,           # Maximum number of connections
        'keepalive_interval': 60,       # Connection keepalive interval (s)
        'max_retries': 3,               # Maximum number of retries
        'retry_delay': 1                # Retry interval (s)
}

class Database:
    def __init__(self, context, pool_config, db_config):
        self.pool_config = pool_config
        self.db_config = db_config
        self.pool = MySQLConnectionPool(context, pool_config, db_config)

    @retry(max_retries=POOL_CONFIG['max_retries'], retry_delay=POOL_CONFIG['retry_delay'])
    def query(self, sql, params=None):
        """
        Perform the query operation
        :param sql: SQL statement
        :param params: SQL parameter
        :return: query result
        """
        conn = self.pool._get_connection()
        result = self.pool._execute_query(conn, sql, params)
        return result

    @retry(max_retries=POOL_CONFIG['max_retries'], retry_delay=POOL_CONFIG['retry_delay'])
    def execute(self, sql, params=None):
        """
        Perform write operations (insert, update, and delete)
        :param sql: SQL statement
        :param params: SQL parameter
        :return: number of affected rows
        """
        conn = self.pool._get_connection()
        result = self.pool._execute_write(conn, sql, params)
        return result

def retry(max_retries=3, retry_delay=1):
    """
    Retry decorator
    :param max_retries: Maximum retries
    :param retry_delay: Retry interval (s)
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            retries = 0
            while retries < max_retries:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    print(f"Attempt {retries + 1} failed: {e}")
                    if retries < max_retries - 1:
                        time.sleep(retry_delay)
                    retries += 1
            print(f"Failed after {max_retries} attempts")
            raise
        return wrapper
    return decorator