Django读写分离架构下Celery任务关联查询强制使用主库方法咨询

有4种常用的实现方案，可根据你的业务场景选择：

方案1：关联查询手动指定数据库（最简单，适用关联调用少的场景） #

所有关联查询链式调用using()方法，显式指定走主库，示例代码：

# 直接在关联管理器后追加using指定主库
my_obj.certain_many_to_many_objects.using('default').all()

该方案不需要修改任何全局配置、实例属性，只针对单次查询生效，适合一个实例只需要少数几个关联查询走主库的场景。

方案2：修改实例的数据库绑定属性（最常用，适用单实例全生命周期走主库） #

Django ORM 实例的_state.db属性存储了该实例默认绑定的数据库别名，你从主库查询到实例后直接修改该属性即可，后续该实例的所有关联查询、保存操作都会默认走指定的主库，示例代码：

# 从主库查询实例
my_obj = MyModel.objects.using('default').get(pk=given_id)
# 强制绑定该实例后续所有操作都走主库
my_obj._state.db = 'default'
# 后续关联查询不需要额外指定，自动走主库
my_obj.certain_many_to_many_objects.all()
my_obj.other_foreign_key_field

该方案对业务代码侵入极低，只需加一行属性赋值就能覆盖单个实例所有后续数据库操作，适配绝大多数Celery异步任务的场景。

方案3：使用上下文管理器临时全局路由到主库（适用单任务所有操作都走主库） #

如果整个Celery任务的所有数据库查询都需要走主库，可以通过线程本地变量配合自定义数据库路由实现全局临时切换，步骤如下：

1. 先定义线程本地存储变量

import threading
thread_local = threading.local()

2. 修改你的数据库路由逻辑，增加判断：如果当前线程标记了强制走主库，所有读请求都路由到主库

class YourDBRouter:
    def db_for_read(self, model, **hints):
        if getattr(thread_local, "force_use_master", False):
            return "default"
        # 原有读路由逻辑不变
        ...

3. 给Celery任务加装饰器/上下文管理器，执行时标记强制走主库

from contextlib import contextmanager

@contextmanager
def force_master_db():
    thread_local.force_use_master = True
    try:
        yield
    finally:
        del thread_local.force_use_master

# 任务中使用
@app.task
def your_async_task(given_id):
    with force_master_db():
        # 该代码块内所有数据库操作自动走主库，不需要手动指定
        my_obj = MyModel.objects.get(pk=given_id)
        my_obj.certain_many_to_many_objects.all()

该方案适合大量Celery任务都需要规避主从同步延迟的场景，不需要修改每个任务的业务查询代码。

方案4：提前预加载所有需要的关联数据 #

如果异步任务需要用到的关联字段是固定的，可以在第一次查询主库的时候就用select_related、prefetch_related把所有关联数据提前加载出来，后续不需要再触发懒查询，自然不会走到只读副本：

my_obj = MyModel.objects.using('default').select_related(
    'foreign_key_field'
).prefetch_related(
    'certain_many_to_many_objects'
).get(pk=given_id)
# 后续调用关联属性不会触发新的查询，直接用预加载的数据
my_obj.certain_many_to_many_objects.all()

该方案还能减少数据库查询次数，适合关联数据量不大、结构固定的场景。

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