Django读库路由查询触发操作错误时如何捕获异常切换写库重试

这类PostgreSQL备库恢复冲突错误属于django.db.utils.OperationalError的特定场景，实现自动切库的核心原则是精准匹配错误、单次重试、避免无差别切库给主库造成额外压力，不要直接全局捕获所有OperationalError重试。

方案1：轻量装饰器实现（适合局部业务逻辑兜底） #

侵入性最低，只需要给需要兜底的查询函数加上装饰器即可，不需要修改全局配置：

import functools
from django.db import OperationalError, connections

# 匹配备库恢复冲突的错误特征字符串
PG_CONFLICT_ERR_KEYWORD = "cancelling statement due to conflict with recovery"
WRITE_DB_ALIAS = "default"  # 写库在DATABASES配置中的别名，按实际配置修改

def fallback_to_write_db_on_recovery_conflict(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except OperationalError as e:
            # 非备库恢复冲突错误直接抛出，不触发重试
            if PG_CONFLICT_ERR_KEYWORD not in str(e).lower():
                raise
            # 清空路由缓存，强制本次重试走写库
            from django.db import router
            if hasattr(router, "_db_cache"):
                router._db_cache.clear()
            # 绑定写库连接执行原查询
            with connections[WRITE_DB_ALIAS]:
                return func(*args, **kwargs)
    return wrapper

使用方式直接装饰包含读查询的函数即可：

@fallback_to_write_db_on_recovery_conflict
def get_user_data(user_id):
    # 原有走读库的查询逻辑不需要修改
    return User.objects.using("replica").get(id=user_id)

方案2：自定义数据库后端（适合全局无侵入兜底） #

如果需要对所有走备库的查询自动生效，不需要给每个函数加装饰器，可以自定义PostgreSQL数据库后端，在游标层捕获异常自动切库：

1. 在项目目录下新建自定义后端文件，路径例如your_project/db_backends/postgresql/base.py：

   from django.db.backends.postgresql import base as pg_base
   from django.db import OperationalError, connections
   
   PG_CONFLICT_ERR_KEYWORD = "cancelling statement due to conflict with recovery"
   WRITE_DB_ALIAS = "default"
   
   class DatabaseWrapper(pg_base.DatabaseWrapper):
       def create_cursor(self, name=None):
           try:
               return super().create_cursor(name=name)
           except OperationalError as e:
               if PG_CONFLICT_ERR_KEYWORD not in str(e).lower():
                   raise
               # 捕获到备库冲突错误时，直接返回写库的游标
               return connections[WRITE_DB_ALIAS].create_cursor(name=name)
   

2. 修改settings.py中读库的数据库引擎配置，指向自定义的后端：

   DATABASES = {
       "default": {
           "ENGINE": "django.db.backends.postgresql",
           # 主库原有配置不变
           "NAME": "main_db",
           "USER": "db_user",
           # ...其他连接配置
       },
       "replica": {
           # 替换成自定义后端的导入路径
           "ENGINE": "your_project.db_backends.postgresql",
           # 备库原有配置不变
           "NAME": "replica_db",
           "USER": "db_user",
           # ...其他连接配置
       }
   }
   

这个方案对上层ORM、Raw SQL查询完全透明，不需要修改任何业务代码。

关键注意事项 #

• 必须精准匹配错误关键字再重试：禁止捕获所有OperationalError就切库，否则会把连接超时、权限错误、磁盘满等其他故障的流量全部打到主库，放大故障影响。
• 只做1次重试：主库不会触发流复制恢复冲突错误，如果切到写库后仍然抛出异常，直接抛出错误即可，不需要增加多次重试逻辑。
• 根源优化可以搭配PostgreSQL参数调整：在备库配置中设置hot_standby_feedback = on，适当调大max_standby_streaming_delay、max_standby_archive_delay参数值，可以大幅降低这类恢复冲突的出现概率，但无法100%避免，兜底逻辑必须保留。
• 如果代码中硬编码了.using('replica')指定读库，重试时需要强制将查询的数据库别名替换为写库别名，避免重试请求再次路由到故障备库。

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