本文目录导读：
在SQLAlchemy中实现读写分离主要有以下几种方式,我来详细介绍：
使用内置的HorizontallyShard（推荐）
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.ext.horizontal_shard import ShardedSession
# 配置主从数据库
engines = {
'master': create_engine('mysql://user:pass@master_host/dbname'),
'slave1': create_engine('mysql://user:pass@slave1_host/dbname'),
'slave2': create_engine('mysql://user:pass@slave2_host/dbname'),
}
class RoutingSession(Session):
# 获取查询使用的数据库（读操作）
def get_bind(self, mapper=None, clause=None, **kwargs):
if self._flushing or (mapper and self._is_write_operation(mapper)):
return engines['master']
else:
# 随机选择从库
import random
return random.choice([engines['slave1'], engines['slave2']])
def _is_write_operation(self, mapper):
# 判断当前操作是否为写操作
return self.is_modified() or self.deleted
# 创建Session工厂
Session = sessionmaker(class_=RoutingSession)
使用SQLAlchemy的Events方式
from sqlalchemy import event, create_engine
from sqlalchemy.orm import Session, sessionmaker
class RoutingSession(Session):
def __init__(self, *args, **kwargs):
self._write_engine = kwargs.pop('write_engine', None)
self._read_engines = kwargs.pop('read_engines', [])
super().__init__(*args, **kwargs)
def get_bind(self, mapper=None, clause=None, **kwargs):
# 获取当前是否处于写操作状态
if self.info.get('write'):
return self._write_engine
else:
import random
return self._read_engines[random.randint(0, len(self._read_engines)-1)]
# 创建数据库引擎
write_engine = create_engine('mysql://user:pass@master_host/dbname')
read_engines = [
create_engine('mysql://user:pass@slave1_host/dbname'),
create_engine('mysql://user:pass@slave2_host/dbname'),
]
# 绑定事件监听
@event.listens_for(RoutingSession, 'before_flush')
def before_flush(session, flush_context, instances):
session.info['write'] = True
@event.listens_for(RoutingSession, 'after_flush')
def after_flush(session, flush_context):
session.info['write'] = False
# 创建Session
session_factory = sessionmaker(
class_=RoutingSession,
write_engine=write_engine,
read_engines=read_engines
)
使用自定义Session实现
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.orm.session import SessionTransaction
import logging
logger = logging.getLogger(__name__)
class ReadWriteSession(Session):
def __init__(self, *args, **kwargs):
self.read_engine = kwargs.pop('read_engine', None)
self.write_engine = kwargs.pop('write_engine', None)
super().__init__(*args, **kwargs)
def get_bind(self, mapper=None, clause=None, **kw):
# 获取当前绑定的数据库
if hasattr(self, 'transaction') and self.transaction is not None:
# 如果在事务中，使用写库
return self.write_engine
# 检查是否写操作
if self.info.get('operating') == 'write':
return self.write_engine
return self.read_engine
# 配置数据库
read_engine = create_engine('mysql://user:pass@slave_host/dbname')
write_engine = create_engine('mysql://user:pass@master_host/dbname')
# 创建Session工厂
Session = sessionmaker(
class_=ReadWriteSession,
read_engine=read_engine,
write_engine=write_engine,
expire_on_commit=False
)
# 使用示例
def create_user(db_session, user_data):
# 手动标记为写操作
db_session.info['operating'] = 'write'
user = User(**user_data)
db_session.add(user)
db_session.commit()
db_session.info['operating'] = None
return user
def get_users(db_session):
# 读操作，默认使用从库
return db_session.query(User).all()
完整的生产环境实现
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.orm.session import SessionTransaction
from contextlib import contextmanager
from typing import List, Optional
import random
class DatabaseRouter:
"""数据库读写分离路由"""
def __init__(self, master_url: str, slave_urls: List[str], echo=False):
self.master_engine = create_engine(master_url, echo=echo)
self.slave_engines = [
create_engine(url, echo=echo) for url in slave_urls
]
self._session_factory = None
@property
def session_factory(self):
if self._session_factory is None:
self._session_factory = sessionmaker(
class_=_ReadWriteSession,
write_engine=self.master_engine,
read_engines=self.slave_engines
)
return self._session_factory
@contextmanager
def get_session(self, force_write=False):
"""获取数据库会话"""
session = self.session_factory()
try:
if force_write:
session.info['write'] = True
yield session
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
@contextmanager
def write_session(self):
"""获取写会话（自动标记为写操作）"""
with self.get_session(force_write=True) as session:
yield session
class _ReadWriteSession(Session):
def __init__(self, *args, **kwargs):
self.write_engine = kwargs.pop('write_engine', None)
self.read_engines = kwargs.pop('read_engines', [])
super().__init__(*args, **kwargs)
def get_bind(self, mapper=None, clause=None, **kw):
# 写操作或事务中使用主库
if self.info.get('write', False) or self._is_in_transaction():
return self.write_engine
# 读操作随机选择从库
return random.choice(self.read_engines)
def _is_in_transaction(self):
"""检查是否在事务中"""
try:
return self.transaction is not None
except Exception:
return False
# 使用示例
router = DatabaseRouter(
master_url='mysql://user:pass@master_host:3306/dbname',
slave_urls=[
'mysql://user:pass@slave1_host:3306/dbname',
'mysql://user:pass@slave2_host:3306/dbname',
],
echo=False
)
# 写操作
with router.write_session() as session:
user = User(name='Alice', email='alice@example.com')
session.add(user)
# 读操作（自动使用从库）
with router.get_session() as session:
users = session.query(User).all()
使用第三方库（简单方案）
# 安装：pip install sqlalchemy-readwrite
from sqlalchemy_readwrite.middleware import ReadWriteMiddleware
# 配置
middleware = ReadWriteMiddleware(
write_db_url='mysql://user:pass@master_host/dbname',
read_db_urls=[
'mysql://user:pass@slave1_host/dbname',
'mysql://user:pass@slave2_host/dbname',
]
)
# 获取session
session = middleware.get_session()
注意事项
- 事务一致性：在写操作事务中，后续读操作应使用主库，避免读取到不一致的数据
- 主从延迟：考虑主从复制延迟，需要一致性强的操作强制使用主库
- 连接池配置：根据不同业务量配置合适的连接池大小
- 监控告警：监控主从同步状态和数据库性能
选择哪种方式取决于你的具体需求,对于小型项目，使用内置的ShardedSession就足够了；对于生产环境，建议使用自定义的完整实现方案。
