Flask应用HAProxy负载AWS RDS MySQL时写API报错，需分读写URI

嘿，我来帮你搞定这个问题！你遇到的写入API频繁报错，核心原因肯定是HAProxy把写请求分发到AWS RDS的只读副本上了——毕竟RDS只读实例是拒绝写操作的。你想分读写两个URL的思路完全正确，下面给你一步步拆解实现方案：

首先得确保HAProxy能把读写请求分别路由到RDS主实例（写）和只读副本池（读），修改你的HAProxy.cfg：

# 写请求专属前端（绑定3307端口）
frontend mysql-write
    bind *:3307
    default_backend mysql-write-backend

# 读请求专属前端（绑定3308端口）
frontend mysql-read
    bind *:3308
    default_backend mysql-read-backend

# 写后端：只关联RDS主实例
backend mysql-write-backend
    mode tcp
    balance roundrobin
    server rds-main 你的主RDS端点:3306 check

# 读后端：关联所有只读副本
backend mysql-read-backend
    mode tcp
    balance roundrobin
    server rds-replica1 你的只读副本1端点:3306 check
    server rds-replica2 你的只读副本2端点:3306 check

这样写请求走3307端口只会到主实例，读请求走3308会被均衡到副本上，从根源避免写请求落到只读实例。

接下来在你的Flask应用里配置两个数据库连接URI，分别对应读写：

# 配置类
class Config:
    # 写连接：指向HAProxy的写端口
    SQLALCHEMY_DATABASE_URI_WRITE = "mysql+pymysql://用户名:密码@HAProxy地址:3307/数据库名"
    # 读连接：指向HAProxy的读端口
    SQLALCHEMY_DATABASE_URI_READ = "mysql+pymysql://用户名:密码@HAProxy地址:3308/数据库名"
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    # 可选：配置连接池参数，优化并发
    SQLALCHEMY_POOL_SIZE = 10
    SQLALCHEMY_MAX_OVERFLOW = 20

然后初始化两个SQLAlchemy实例（或者用绑定功能，看你习惯）：

方式1：双实例方案（简单直观） #

from flask import Flask, jsonify, request
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config.from_object(Config)

# 初始化写数据库实例
db_write = SQLAlchemy(app)
# 切换配置为读URI，初始化读数据库实例
app.config['SQLALCHEMY_DATABASE_URI'] = app.config['SQLALCHEMY_DATABASE_URI_READ']
db_read = SQLAlchemy(app)

# 你的模型类（用哪个实例定义都可以，因为结构一致）
class YourModel(db_write.Model):
    id = db_write.Column(db_write.Integer, primary_key=True)
    content = db_write.Column(db_write.String(255))

    def to_dict(self):
        return {"id": self.id, "content": self.content}

方式2：SQLAlchemy绑定方案（更优雅） #

如果不想维护两个实例，用SQLAlchemy的多绑定功能：

app.config['SQLALCHEMY_BINDS'] = {
    'write': Config.SQLALCHEMY_DATABASE_URI_WRITE,
    'read': Config.SQLALCHEMY_DATABASE_URI_READ
}
db = SQLAlchemy(app)

# 模型类可以指定默认绑定，或者操作时指定
class YourModel(db.Model):
    __bind_key__ = 'write'  # 默认用写绑定，写操作不用额外指定
    id = db.Column(db.Integer, primary_key=True)
    content = db.Column(db.String(255))

    def to_dict(self):
        return {"id": self.id, "content": self.content}

现在就可以在不同API里用对应的连接了：

只读API（用读连接） #

@app.route('/api/data', methods=['GET'])
def get_data():
    # 双实例方案：用db_read的会话
    data = db_read.session.query(YourModel).all()
    # 绑定方案：指定用read绑定
    # data = db.session.query(YourModel).bind('read').all()
    return jsonify([item.to_dict() for item in data])

写入API（用写连接） #

@app.route('/api/data', methods=['POST'])
def add_data():
    new_item = YourModel(content=request.json.get('content'))
    # 双实例方案：用db_write的会话
    db_write.session.add(new_item)
    db_write.session.commit()
    # 绑定方案：默认用write绑定，直接操作即可
    # db.session.add(new_item)
    # db.session.commit()
    return jsonify(new_item.to_dict()), 201

• 可以封装一个装饰器，自动给路由分配读写会话，减少重复代码
• 给写操作加上错误捕获，如果遇到只读实例的报错，可以自动重试到写端点
• 定期检查HAProxy的后端状态，确保主实例和副本都正常在线

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