读取从节点，读写主节点设置

28 投票

3 回答

16979 浏览

数据工程师

提问于 2025-04-17 10:44

我有一个使用Flask和SQLAlchemy的网页应用，它连接到一个MySQL数据库服务器。我想扩展这个数据库设置，增加一个只读的从服务器，这样我就可以把读取的请求分散到主服务器和从服务器上，同时继续向主数据库服务器写入数据。

我看过一些选项，觉得用普通的SQLAlchemy可能不行。我的计划是，在我的网页应用中创建两个数据库连接，一个连接主服务器，一个连接从服务器。然后我打算用一个简单的随机值来决定“SELECT”操作是用主服务器的连接还是从服务器的连接。

不过，我不太确定这样做是否正确，关于如何实现这个目标，有没有什么建议或者技巧？

3 个回答

也许这个回答来得有点晚！我使用一个 slave_session 来查询从数据库。

class RoutingSession(SignallingSession):
def __init__(self, db, bind_name=None, autocommit=False, autoflush=True, **options):
    self.app = db.get_app()
    if bind_name:
        bind = options.pop('bind', None)
    else:
        bind = options.pop('bind', None) or db.engine

    self._bind_name = bind_name
    SessionBase.__init__(
        self, autocommit=autocommit, autoflush=autoflush,
        bind=bind, binds=None, **options
    )

def get_bind(self, mapper=None, clause=None):
    if self._bind_name is not None:
        state = get_state(self.app)
        return state.db.get_engine(self.app, bind=self._bind_name)
    else:
        if mapper is not None:
            try:
                persist_selectable = mapper.persist_selectable
            except AttributeError:
                persist_selectable = mapper.mapped_table

            info = getattr(persist_selectable, 'info', {})
            bind_key = info.get('bind_key')
            if bind_key is not None:
                state = get_state(self.app)
                return state.db.get_engine(self.app, bind=bind_key)
        return SessionBase.get_bind(self, mapper, clause)


class RouteSQLAlchemy(SQLAlchemy):
    def __init__(self, *args, **kwargs):
        SQLAlchemy.__init__(self, *args, **kwargs)
        self.slave_session = self.create_scoped_session({'bind_name': 
'slave'})

    def create_session(self, options):
        return orm.sessionmaker(class_=RoutingSession,db=self,**options)

db = RouteSQLAlchemy(metadata=metadata, query_class=orm.Query)

回答于 2025-04-17 由 Python大师

分享举报

或者，我们可以尝试另一种方法。比如，我们可以声明两个不同的类，它们的实例属性都是一样的，但 __bind__ 这个类属性是不同的。这样，我们就可以用 rw 类来进行读写操作，用 r 类来进行只读操作。 :)

我觉得这种方法更简单也更可靠。 :)

我们声明两个数据库模型是因为我们可能在两个不同的数据库中有同名的表。这样做还可以避免当两个模型有相同的 __tablename__ 时出现的 'extend_existing' 错误。

下面是一个例子：

app = Flask(__name__)
app.config['SQLALCHEMY_BINDS'] = {'rw': 'rw', 'r': 'r'}
db = SQLAlchemy(app)
db.Model_RW = db.make_declarative_base()

class A(db.Model):
    __tablename__ = 'common'
    __bind_key__ = 'r'

class A(db.Model_RW):
    __tablename__ = 'common'
    __bind_key__ = 'rw'

回答于 2025-04-17 由 Python大师

分享举报

我在我的博客上有一个关于如何实现这个功能的例子，地址是 http://techspot.zzzeek.org/2012/01/11/django-style-database-routers-in-sqlalchemy/。简单来说，你可以增强会话（Session），让它在每次查询时选择主数据库或从数据库。这个方法可能会有一个小问题，就是如果你有一个事务调用了六个查询，你可能会在一次请求中使用到两个从数据库……不过我们只是想模仿Django的功能而已 :)

我用过一种稍微简单一点的方法，它更明确地定义了使用范围，就是在视图调用上使用装饰器（在Flask中不管它们叫什么），像这样：

@with_slave
def my_view(...):
   # ...

with_slave的作用大概是这样的，假设你已经设置好了会话和一些数据库引擎：

master = create_engine("some DB")
slave = create_engine("some other DB")
Session = scoped_session(sessionmaker(bind=master))

def with_slave(fn):
    def go(*arg, **kw):
        s = Session(bind=slave)
        return fn(*arg, **kw)
    return go

这个想法是，调用 Session(bind=slave) 会让注册表获取当前线程的实际会话对象，如果不存在就创建一个——但因为我们传递了一个参数，scoped_session会确保我们这里创建的会话是全新的。

你将它指向“从数据库”，这样后续的所有SQL操作都会使用这个“从数据库”。然后，当请求结束时，你需要确保你的Flask应用调用 Session.remove() 来清除该线程的注册表。当同一个线程下次使用注册表时，它会绑定回“主数据库”的一个新会话。

或者，如果你只想在那次调用中使用“从数据库”，这样做是“更安全”的，因为它会将任何现有的绑定恢复到会话：

def with_slave(fn):
    def go(*arg, **kw):
        s = Session()
        oldbind = s.bind
        s.bind = slave
        try:
            return fn(*arg, **kw)
        finally:
            s.bind = oldbind
    return go

对于这些装饰器，你可以反向操作，让会话绑定到“从数据库”，而装饰器则将其设置为“主数据库”进行写操作。如果你想要一个随机的从数据库，在这种情况下，如果Flask有某种“请求开始”事件，你可以在那个时候进行设置。

回答于 2025-04-17 由 Python大师

分享举报

读取从节点，读写主节点设置

3 个回答

撰写回答