项目同时使用MariaDB和MySQL连接器如何正确关闭连接避免连接数超限

问题根源 #

你遇到的Too many connections报错核心是数据库连接资源未正确释放，有两个直接原因：

1. 本地环境调用mariadb.connect每次都会创建新连接，仅返回游标对象时没有持有连接实例，查询结束后连接被遗弃无法自动关闭，持续累积耗尽连接数。
2. Django自带的connection.cursor()如果未主动关闭，高并发场景下也会长期占用Django维护的连接池资源。

最优解决方案：将get_cursor()改造为上下文管理器，自动处理连接/游标释放 #

该方案改动量极小，仅需调整get_cursor()实现，原有100个查询函数只需少量调整游标获取逻辑即可，无需修改核心查询代码。

第一步：改造get_cursor() #

引入Python内置的contextlib.contextmanager装饰器可以快速实现上下文管理器，查询结束后自动在finally块中释放资源：

from contextlib import contextmanager
import mariadb
from django.conf import settings
from django.db import connection

@contextmanager
def get_cursor():
    conn = None
    cursor = None
    try:
        if 'local' == settings.DJANGO_HOST:
            # 本地环境创建MariaDB连接
            conn = mariadb.connect(
                user="user",
                password="password",
                host="localhost",
                database="db-dummy"
            )
            cursor = conn.cursor(named_tuple=True)
        else:
            # 预发布环境使用Django托管的连接
            cursor = connection.cursor()
        # 返回游标给业务逻辑使用
        yield cursor
    finally:
        # 统一关闭游标
        if cursor:
            cursor.close()
        # 本地自建的MariaDB连接单独关闭，Django托管的连接由框架自行管理无需手动关闭
        if conn:
            conn.commit() # 兼容写入操作正常提交，只读查询无影响
            conn.close()

第二步：修改原有查询函数的游标获取逻辑 #

将原有cursor = get_cursor()的调用改为with get_cursor() as cursor:即可，示例如下：

# 修改后的MariaDB查询函数
def get_some_data_mariadb(record_id):
    results = []
    try:
        with get_cursor() as cursor:
            cursor.execute("SELECT blabla some query")
            columns = [column[0] for column in cursor.description]
            for row in cursor.fetchall():
                results.append(dict(zip(columns, row)))
        return results
    except mariadb.Error as e:
        print(f"Error: {e}")

# 修改后的MySQL查询函数
def get_some_data_mysql(record_id):
    results = []
    with get_cursor() as cursor:
        cursor.execute("SELECT blablabla")
        columns = [column[0] for column in cursor.description]
        for row in cursor.fetchall():
            results.append(dict(zip(columns, row)))
    return results

可选优化 #

如果本地环境查询量很大，可以进一步引入MariaDB连接池替代每次新建连接，降低连接创建销毁的性能开销，进一步减少连接资源占用。

内容的提问来源于stack exchange，提问作者Cheknov