MySQL数据库读写分离与负载均衡的实现逻辑

-- 存储过程无法通过语法区分读写，只能路由至主库
delimiter //
create procedure getorderdetails(in orderid int)
begin 
  select * from orders where id = orderid; -- 读操作，但中间件无法识别
  update orders set status = 'processed' where id = orderid; -- 写操作
end //
delimiter ;

# maxscale 配置文件 (maxscale.cnf)  
# 定义主库
[server1]
type=server
address=192.168.1.101 
port=3306
protocol=mysqlbackend 
# 定义从库 
[server2]
type=server 
address=192.168.1.102 
port=3306
protocol=mysqlbackend
# 定义从库
[server3]
type=server
address=192.168.1.103
port=3306 
protocol=mysqlbackend 
# 读写分离路由 
[readwritesplit]
type=service 
router=readwritesplit
servers=server1,server2,server3
user=maxscale
password=maxscale_pass 
# 负载均衡策略 
[readonlyservice]
type=service 
router=readconnroute 
router_options=slave 
servers=server2,server3
# 监控主从状态  
[replicationmonitor]  
type=monitor  
module=mariadbmon  
servers=server1, server2, server3  
user=monitor_user  
password=monitor_password

-- 主库配置 (my.cnf)  
[mysqld]  
server-id=1  
log-bin=mysql-bin  
binlog-format=row
binlog-do-db=your_database  
-- 从库配置 (my.cnf)  
[mysqld]  
server-id=2  
relay-log=mysql-relay-bin  
read-only=1  

change master to  
	master_host='master_ip',  
	master_user='replica_user',  
	master_password='password',  
	master_log_file='mysql-bin.000001',  
	master_log_pos=107;  
start slave;

// database.providers.ts  
import { datasource } from 'typeorm';  
export const databaseproviders = [  
  {  
    provide: 'master_database',  
    usefactory: () => new datasource({  
      type: 'mysql',  
      host: 'master_host',  
      port: 3306,  
      username: 'master_user',  
      password: 'master_password',  
      database: 'your_db',  
      entities: [__dirname + '/..//*.entity{.ts,.js}'],  
      synchronize: false,  
    }).initialize(),  
  },  
  {  
    provide: 'slave_database',  
    usefactory: () => new datasource({  
      type: 'mysql',  
      host: 'slave_host',  
      port: 3306,  
      username: 'slave_user',  
      password: 'slave_password',  
      database: 'your_db',  
      entities: [__dirname + '/..//*.entity{.ts,.js}'],  
      synchronize: false,  
    }).initialize(),  
  },  
];  
// user.service.ts  
import { injectable, inject } from '@nestjs/common';  
import { injectdatasource } from '@nestjs/typeorm';  
import { datasource } from 'typeorm';  
@injectable()  
export class userservice {  
  constructor(  
    @inject('master_database') private masterdatasource: datasource,  
    @inject('slave_database') private slavedatasource: datasource,  
  ) {}  
  // 写操作强制使用主库  
  async createuser(userdata: any) {  
    return this.masterdatasource.query(  
      `insert into users (name, email) values (?, ?)`,  
      [userdata.name, userdata.email],  
    );  
  }  
  // 读操作使用从库（非实时敏感查询）  
  async getusers() {  
    return this.slavedatasource.query(`select * from users`);  
  }  
  // 实时敏感查询使用主库  
  async getrealtimedata() {  
    return this.masterdatasource.query(  
      `/* force_master */ select * from transactions where id = ?`,  
      [transactionid],  
    );  
  }  
}  

// database.providers.ts - 定义主从库连接池  
import { createpool, pool } from 'mysql2/promise';  
const masterpool: pool = createpool({  
  host: 'master_db_host',  
  user: 'write_user',  
  password: 'write_password',  
  database: 'app_db',  
  connectionlimit: 10,  
});  
const slavepool: pool = createpool({  
  host: 'slave_db_host', // 可扩展为数组实现负载均衡  
  user: 'read_user',  
  password: 'read_password',  
  database: 'app_db',  
  connectionlimit: 20,  
});  
// query.service.ts - 根据操作类型路由查询  
import { injectable } from '@nestjs/common';  
@injectable()  
export class queryservice {  
  async executequery(sql: string, iswrite: boolean = false) {  
    const pool = iswrite ? masterpool : slavepool;  
    const [rows] = await pool.query(sql);  
    return rows;  
  }  
  // 实时性敏感查询显式指定主库  
  async criticalquery(sql: string) {  
    return this.executequery(`/*master*/ ${sql}`, true);  
  }  
}  
// user.controller.ts - 业务层调用示例  
@controller('users')  
export class usercontroller {  
  constructor(private queryservice: queryservice) {}  
  @get(':id')  
  async getuser(@param('id') id: string) {  
    // 读操作路由到从库  
    return this.queryservice.executequery(  
      `select * from users where id = ${id}`  
    );  
  }  
  @post()  
  async createuser(@body() userdata) {  
    // 写操作路由到主库  
    return this.queryservice.executequery(  
      `insert into users (name) values ('${userdata.name}')`,  
      true  
    );  
  }  
}

// 配置主库和从库连接 
const masterdbconfig = { host: 'master-db', user: 'admin', password: 'pass' };
const slavedbconfigs = [
  { host: 'slave-db1', user: 'readonly', password: 'pass' },
  { host: 'slave-db2', user: 'readonly', password: 'pass' }
];
// 手动路由读写操作 
import { datasource } from 'typeorm';
const masterdatasource = new datasource({ ...masterdbconfig, type: 'mysql' });
const slavedatasources = slavedbconfigs.map(config => new datasource({ ...config, type: 'mysql' }));
// 写操作路由至主库
async function executewrite(query: string) {
  await masterdatasource.query(query);
}
// 读操作轮询路由至从库（简单负载均衡）
let currentslaveindex = 0;
async function executeread(query: string) {
  const datasource = slavedatasources[currentslaveindex];
  currentslaveindex = (currentslaveindex + 1) % slavedatasources.length;
  return datasource.query(query);
}

-- proxysql 读分组配置示例  
insert into mysql_servers (hostgroup_id, hostname, port) values  
(10, 'slave1', 3306),  
(10, 'slave2', 3306);  
update mysql_query_rules set destination_hostgroup=10 where match_pattern='^select';  
load mysql servers to runtime;  

-- 启用gtid与半同步  
set global gtid_mode=on;  
set global enforce_gtid_consistency=on;  
install plugin rpl_semi_sync_master soname 'semisync_master.so';  

-- 主库配置
install plugin rpl_semi_sync_master soname 'semisync_master.so';
set global rpl_semi_sync_master_enabled = 1;
-- 从库配置
install plugin rpl_semi_sync_slave soname 'semisync_slave.so';
set global rpl_semi_sync_slave_enabled = 1;

// mha状态监控端点 (nestjs)  
@get('/mha-status')  
async checkmha() {  
 const ssh = await exec('masterha_check_status --conf=/etc/mha.cnf');  
 return { status: ssh.includes('alive') ? 'ok' : 'down' };  
}