Python-SQLAlchemy主从数据库复制读写分离最佳实践咨询

核心思路 #

SQLAlchemy 通过绑定多数据库引擎 + 自定义会话路由逻辑，实现主库处理写操作（INSERT/UPDATE/DELETE）、从库处理读操作（SELECT）的分离，本质是让会话根据当前操作类型自动选择对应的数据库连接。

具体实现步骤 #

1. 配置主从库引擎

分别创建主库（写）和从库（读）的数据库引擎，从库可配置多个实现负载均衡：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
import random

# 主库连接配置
master_engine = create_engine(
    "mysql+pymysql://username:password@master-host:3306/your_db",
    pool_size=10,
    max_overflow=20
)

# 多从库连接配置
slave_engines = [
    create_engine(
        "mysql+pymysql://username:password@slave1-host:3306/your_db",
        pool_size=10,
        max_overflow=20
    ),
    create_engine(
        "mysql+pymysql://username:password@slave2-host:3306/your_db",
        pool_size=10,
        max_overflow=20
    )
]

2. 自定义路由会话类

继承 SQLAlchemy 的 Session 类，重写 get_bind 方法实现操作路由：

class RoutingSession(Session):
    def get_bind(self, mapper=None, clause=None):
        # 写操作（触发flush时）走主库
        if self._flushing:
            return master_engine
        
        # 读操作随机选择从库（可替换为轮询等负载均衡策略）
        return random.choice(slave_engines)

# 创建会话工厂
SessionFactory = sessionmaker(class_=RoutingSession)

3. 业务代码中使用

通过会话工厂创建会话，读写操作自动路由：

# 获取会话
session = SessionFactory()

# 写操作：自动走主库
new_user = User(name="Bob", email="bob@example.com")
session.add(new_user)
session.commit()

# 读操作：自动随机走从库
users = session.query(User).filter(User.name.like("%ob%")).all()

# 手动指定主库读（适用于读取刚写入数据的场景）
users_from_master = session.query(User).with_session(session.bind(master_engine)).all()

进阶优化点 #

• 事务内强制主库读：事务中需要保证数据一致性时，强制所有操作走主库：

  class RoutingSession(Session):
      def get_bind(self, mapper=None, clause=None):
          # 事务中强制走主库
          if self.is_active:
              return master_engine
          if self._flushing:
              return master_engine
          return random.choice(slave_engines)
  

• 从库轮询负载均衡：替换随机选择，用计数器实现轮询：

  slave_index = 0
  def get_slave_engine():
      global slave_index
      engine = slave_engines[slave_index]
      slave_index = (slave_index + 1) % len(slave_engines)
      return engine
  
  # 在RoutingSession的get_bind中调用get_slave_engine()
  

• 连接池优化：根据业务量调整每个引擎的pool_size、max_overflow参数，避免连接耗尽。

注意事项 #

• 主从同步延迟：对数据一致性要求极高的场景，刚写入后需手动指定主库读取。
• 只读会话：纯读业务可创建只读会话，强制所有操作走从库，降低主库压力。
• 故障转移：路由逻辑中可增加从库可用性检测，跳过不可用节点。

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