Django读写分离配置失效：无法指定查询目标数据库

问题回顾 #

生产环境中配置Django 3.2主从数据库，期望读请求路由到只读副本，但实际所有读请求仍走主库，且.using("replica")指定数据库也无效。相关配置代码如下：

数据库配置（settings.py） #

DATABASES = {
    "default": {
        "NAME": "ironbank",
        "ENGINE": "django_postgrespool2",
        "USER": "postgres",
        "PASSWORD": os.getenv("DATABASE_PASSWORD", "postgres"),
        "HOST": os.getenv("DATABASE_HOST", "localhost"),
        "PORT": os.getenv("DATABASE_PORT", "5432"),
        "CONN_MAX_AGE": 0,
    },
    "replica": {
        "NAME": "ironbank_replica",
        "ENGINE": "django_postgrespool2",
        "USER": "postgres",
        "PASSWORD": os.getenv("DATABASE_PASSWORD", "postgres"),
        "HOST": os.getenv("REPLICA_DATABASE_HOST", "localhost"),
        "PORT": os.getenv("REPLICA_DATABASE_PORT", "5432"),
        "TEST": {
            "MIRROR": "default",
        },
        "CONN_MAX_AGE": 0,
    },
}

数据库路由配置 #

settings.py

DATABASE_ROUTERS = ["ironbank.router.PrimaryReplicaRouter"]

ironbank/router/PrimaryReplicaRouter.py

class PrimaryReplicaRouter:
    @staticmethod
    def db_for_read(model, **hints):
        """Reads go to a randomly-chosen replica."""
        # 移除了模型特定逻辑，简化后直接返回replica
        return "replica"

    @staticmethod
    def db_for_write(model, **hints):
        """Writes always go to primary."""
        return "default"

    @staticmethod
    def allow_relation(obj1, obj2, **hints):
        """Relations between objects are allowed if both objects are in the primary/replica pool."""
        return True

    @staticmethod
    def allow_migrate(db, app_label, model_name=None, **hints):
        """Allow migrations on default database only."""
        return db == "default"

排查与解决步骤 #

1. 验证副本数据库连接有效性 #

如果副本数据库配置错误（如HOST/PORT不对、库不存在、权限不足），Django会自动 fallback 到默认数据库，导致所有请求走主库：

• 手动测试副本库连接：使用psql命令直接连接，确认能正常访问：

  psql -h <REPLICA_HOST> -p <REPLICA_PORT> -U postgres -d ironbank_replica
  

• 检查环境变量：确认REPLICA_DATABASE_HOST和REPLICA_DATABASE_PORT已正确设置，本地测试时指向正确的副本实例。

2. 检查路由类的导入路径 #

确认DATABASE_ROUTERS中的路径完全正确：

• 检查ironbank/router/PrimaryReplicaRouter.py是否存在，文件名、类名拼写无错误。
• 确保ironbank是Django项目的根应用，包含__init__.py文件（避免命名空间导入问题）。

3. 开启数据库日志排查实际连接情况 #

在settings.py中添加数据库日志配置，查看每个查询实际使用的数据库：

LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
        },
    },
    'loggers': {
        'django.db.backends': {
            'handlers': ['console'],
            'level': 'DEBUG',
            'propagate': False,
        },
    },
}

启动项目后查看控制台日志，确认是否有尝试连接replica库的记录，若没有则说明路由未生效或连接失败。

4. 修正.using()方法的使用方式 #

.using()仅对查询集或未加载的对象操作有效：

• 对于查询集，需确保using()调用在查询执行前：

  # 正确用法
  MyModel.objects.using('replica').all()
  # 错误用法：先执行查询再指定数据库（不会生效）
  queryset = MyModel.objects.all()
  queryset.using('replica')
  

• 对于已加载的对象，需使用refresh_from_db()指定数据库：

  obj = MyModel.objects.first()
  obj.refresh_from_db(using='replica')
  

5. 调整连接池配置 #

CONN_MAX_AGE = 0会关闭数据库连接池，每次请求重新创建连接，可能导致路由逻辑异常。尝试设置合理的连接池时长：

"CONN_MAX_AGE": 60,  # 保持连接60秒

6. 排除第三方库或中间件干扰 #

检查是否有第三方中间件、ORM扩展库（如多租户类库）强制覆盖了数据库选择逻辑。可以临时移除第三方库，测试路由是否生效。

7. 验证数据差异 #

在主库插入一条测试数据，暂时不同步到副本库，然后执行查询：

# 主库插入数据
MyModel.objects.create(name="test_primary")
# 查询副本库（若未同步则不应返回该数据）
MyModel.objects.using('replica').filter(name="test_primary").exists()

通过这种方式可以直观验证是否真的连接到了副本库。

内容的提问来源于stack exchange，提问作者Yash Kumar Verma