使用 mssql-django 重试逻辑和连接复原能力

由于与代码无关的原因，与SQL Server和Azure SQL的连接可能会暂时失败：

• AlwaysOn 可用性组故障转移。
• 在连接设置过程中，网络会丢弃数据包。
• Resource Governor限制数据库。
• Azure SQL 副本会在缩放或升级期间被回收。

其中大多数故障在几秒钟内就清除。 本文介绍如何在使用 mssql-django 后端的 Django 应用程序中重试瞬时错误，以及如何配置 Django 和 ODBC 驱动程序，以便在空闲连接断开后自动恢复。

暂时性错误

暂时性错误是自行解决的临时故障。 在短暂延迟后重试操作通常成功。

在建立连接期间发生或向服务器发送请求时，以下错误是暂时性的。 在短的、受限的退避时重试。 在几次重试之后保留的错误通常表示配置问题（服务器错误、权限缺失、配额耗尽），重试无法解决。

语句级错误不在此列表中，因为它们发生在连接建立之后，而且即使失败，会话仍可继续使用。 最常见的可重试语句错误是 1205（死锁受害者）和 1222（锁请求超时）。 重试整个事务，而不是单个失败语句。

错误消息文本来自Azure SQL暂时性连接错误。 各个驱动程序维护各自的内置重试列表；此目录说明了在 SQL Server、Azure SQL 数据库、Azure SQL 托管实例、Microsoft Fabric 中的 SQL 数据库以及 Azure Synapse Analytics 中的专用 SQL 池中，哪些错误符合重试条件。

ODBC 驱动程序空闲连接复原能力

Microsoft ODBC Driver for SQL Server 通过 ConnectRetryCount 和 ConnectRetryInterval 连接字符串关键字提供内置的空闲连接恢复能力。 在涉及应用程序代码之前，这些设置在驱动程序级别处理已删除的空闲连接。

在 extra_params 中启用空闲连接复原：

DATABASES = {
    "default": {
        "ENGINE": "mssql",
        "NAME": "<your-database>",
        "HOST": "<your-server>.database.windows.net",
        "PORT": "1433",
        "OPTIONS": {
            "driver": "ODBC Driver 18 for SQL Server",
            "extra_params": "ConnectRetryCount=3;ConnectRetryInterval=10",
        },
    },
}

用于重试的 Django 数据库中间件

创建一个 Django 中间件，用于捕获暂时性错误并重试数据库操作。 此方法适用于视图级请求处理：

# myproject/middleware.py
import random
import re
import time
import logging
from django.db import OperationalError, connection

logger = logging.getLogger(__name__)

TRANSIENT_ERROR_CODES = {
    "64", "233", "4221",
    "10053", "10054", "10928", "10929",
    "40197", "40501", "40613",
    "49918", "49919", "49920",
    # Include "4060" only if targeting Azure SQL with geo-replication failover.
    # It is usually a permanent error (wrong database name or missing permissions).
}

# Microsoft ODBC driver formats native error codes as "(<number>)" in the
# message. Extracting parenthesized codes avoids false positives that a plain
# substring match would produce for short codes like "64".
_CODE_RE = re.compile(r"\((\d+)\)")


def is_transient(error):
    codes_in_message = set(_CODE_RE.findall(str(error)))
    return bool(codes_in_message & TRANSIENT_ERROR_CODES)


class DatabaseRetryMiddleware:
    """Retry database operations on transient errors."""

    def __init__(self, get_response):
        self.get_response = get_response
        self.max_retries = 3
        self.base_delay = 1   # seconds; doubled each attempt
        self.max_delay = 30   # cap on a single sleep, regardless of attempt

    def __call__(self, request):
        for attempt in range(self.max_retries + 1):
            try:
                return self.get_response(request)
            except OperationalError as e:
                if attempt < self.max_retries and is_transient(e):
                    # Exponential backoff with full jitter, capped at max_delay.
                    # Jitter spreads simultaneous retries so many clients
                    # don't hammer the server in lock-step during an outage.
                    capped = min(self.max_delay, self.base_delay * (2 ** attempt))
                    delay = random.uniform(0, capped)
                    logger.warning(
                        "Transient DB error (attempt %d/%d), retrying in %.2fs: %s",
                        attempt + 1, self.max_retries, delay, e
                    )
                    connection.close()
                    time.sleep(delay)
                    continue
                raise

在settings.py中注册中间件：

MIDDLEWARE = [
    "myproject.middleware.DatabaseRetryMiddleware",
    "django.middleware.security.SecurityMiddleware",
    # ... other middleware
]

用于特定操作的重试修饰器

若要进行更细粒度的控制，请在各个单独的函数上使用修饰器：

import random
import re
import time
import functools
import logging
from django.db import OperationalError, connection

logger = logging.getLogger(__name__)

TRANSIENT_ERROR_CODES = {
    "64", "233", "4221",
    "10053", "10054", "10928", "10929",
    "40197", "40501", "40613",
    "49918", "49919", "49920",
    # Include "4060" only if targeting Azure SQL with geo-replication failover.
}

_CODE_RE = re.compile(r"\((\d+)\)")


def is_transient(error):
    codes_in_message = set(_CODE_RE.findall(str(error)))
    return bool(codes_in_message & TRANSIENT_ERROR_CODES)


def retry_on_transient(max_retries=3, base_delay=1, max_delay=30):
    """Retry on transient database errors with exponential backoff and full jitter."""

    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except OperationalError as e:
                    if attempt < max_retries and is_transient(e):
                        # Exponential cap doubled per attempt, then jittered
                        # within [0, cap] and limited by max_delay.
                        capped = min(max_delay, base_delay * (2 ** attempt))
                        delay = random.uniform(0, capped)
                        logger.warning(
                            "Transient error in %s (attempt %d/%d), retrying in %.2fs: %s",
                            func.__name__, attempt + 1, max_retries, delay, e
                        )
                        connection.close()
                        time.sleep(delay)
                        continue
                    raise
        return wrapper
    return decorator

将修饰器应用于数据库密集型函数：

from myproject.retry import retry_on_transient

@retry_on_transient(max_retries=3, base_delay=2)
def process_order(order_id):
    """Process an order with automatic retry on transient failures."""
    order = Order.objects.select_for_update().get(id=order_id)
    order.status = "processing"
    order.save()
    return order

使用事务重试

当事务中发生瞬态错误时，服务器会回滚整个事务。 重试完整的事务，而不仅仅是失败的语句：

from django.db import transaction

@retry_on_transient(max_retries=3)
def transfer_funds(from_account_id, to_account_id, amount):
    """Transfer funds between accounts with retry."""
    with transaction.atomic():
        from_account = Account.objects.select_for_update().get(id=from_account_id)
        to_account = Account.objects.select_for_update().get(id=to_account_id)

        from_account.balance -= amount
        to_account.balance += amount

        from_account.save()
        to_account.save()

语句级错误

上一部分中的错误列表涵盖连接级故障。 另外两种错误通常也会在语句级别进行重试：

• 1205：该会话被选作死锁牺牲品。 重新运行事务。
• 1222：已超过锁请求超时时间。 重新运行事务;如果默认值过于咄咄逼人，则重新运行事务或增加 LOCK_TIMEOUT 会话。

ConnectRetryCount 重试断开的连接，因此它不适用于这些语句级错误。 对于可安全重新运行的事务，通过将 "1205" 和 "1222" 添加到 TRANSIENT_ERROR_CODES，使用相同的装饰器模式来处理它们。

CONN_MAX_AGE 和陈旧连接

设置时 CONN_MAX_AGE ，Django 会跨请求重复使用数据库连接。 如果服务器关闭该连接（例如，在Azure SQL缩放操作或防火墙超时期间），长期连接可能会过时。

将 CONN_MAX_AGE 设置为在复用和陈旧性之间取得平衡：

DATABASES = {
    "default": {
        "ENGINE": "mssql",
        "NAME": "<your-database>",
        "HOST": "<your-server>.database.windows.net",
        "OPTIONS": {
            "driver": "ODBC Driver 18 for SQL Server",
        },
        "CONN_MAX_AGE": 600,  # Close and reopen connections after 10 minutes
    },
}

• CONN_MAX_AGE=0 （默认值）：关闭每个请求末尾的连接。 最安全但最慢。
• CONN_MAX_AGE=600：重用连接 10 分钟。 大多数 Web 应用程序的平衡良好。
• CONN_MAX_AGE=None：无限期保持连接打开状态。 仅可与针对失效连接的重试机制一起使用。

CONN_HEALTH_CHECKS （Django 4.1 及更高版本）

Django 4.1 引入了 CONN_HEALTH_CHECKS，用于在每次请求之前验证复用的连接。 将它与 CONN_MAX_AGE 一起启用，以自动检测失效连接：

DATABASES = {
    "default": {
        "ENGINE": "mssql",
        "NAME": "<your-database>",
        "HOST": "<your-server>.database.windows.net",
        "OPTIONS": {
            "driver": "ODBC Driver 18 for SQL Server",
        },
        "CONN_MAX_AGE": 600,
        "CONN_HEALTH_CHECKS": True,
    },
}

启用运行状况检查后，Django 会在重用连接之前发出轻型验证查询。 如果连接断开，Django 会透明地打开一个新连接，而不是引发错误。

最佳做法

• 将指数退避与完全抖动一起使用。 每次尝试都将上限值翻倍，然后在 [0, cap] 范围内随机等待一段时间。 抖动可防止大量客户端在区域故障期间同步重试，否则原本短暂的故障可能会演变为持续过载。 限制每次尝试睡眠（例如 30 秒），因此总恢复时间保持有限。
• 设置重试上限。 使用指数退避的三次重试是合理的默认值。 重试超过 5 次通常表明存在非暂时性问题。
• 在重试之前关闭连接。 调用 connection.close() 以便 Django 在下一次尝试时打开新的连接。
• 记录每次重试。 静默成功的重试可能会掩盖性能问题。 以 WARNING 级别记录，以便跟踪频率。
• 不要重试非暂时性错误。 身份验证失败、权限错误和语法错误不受益于重试。
• 重试整个事务。 将 transaction.atomic() 包裹在重试逻辑中，而不是反过来。
• 启用 CONN_HEALTH_CHECKS（Django 4.1 及更高版本），用于使用 CONN_MAX_AGE 的 Web 应用程序。

相关内容

• mssql-django 的连接选项
• mssql-django 中的连接池
• mssql-django 中的事务管理
• mssql-django 的性能优化
• 排查 mssql-django 故障