ShardingProxy读写分离之原理、配置与实践过程

-- 主库创建测试库
CREATE DATABASE IF NOT EXISTS sharding_db;
USE sharding_db;

-- 创建用户表
CREATE TABLE `t_user` (
  `id` BIGINT NOT NULL AUTO_INCREMENT,
  `username` VARCHAR(50) NOT NULL,
  `age` INT NOT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-- 插入测试数据（从库需同步该数据）
INSERT INTO t_user (username, age) VALUES ('zhangsan', 25), ('lisi', 30);

# 服务端口（应用连接ShardingProxy的端口）
serverPort: 3307
# 管理端口（用于监控和运维）
adminPort: 8088
# 数据库驱动类名
driverClassName: com.mysql.cj.jdbc.Driver
# 认证配置（应用连接ShardingProxy的账号密码）
authentication:
  users:
    root:
      password: 123456  # 应用连接密码
    sharding:
      password: sharding 
      authorizedSchemas: sharding_db  # 授权访问的数据库

# 日志配置（输出SQL路由日志，便于调试）
props:
  max.connections.size.per.query: 1
  acceptor.size: 16  # 用于接收连接的线程数
  executor.size: 16  # 用于处理SQL的线程数
  proxy.frontend.flush.threshold: 128  # 前端刷盘阈值
  proxy.transaction.type: LOCAL
  proxy.opentracing.enabled: false
  query.with.cipher.column: true
  sql.show: true  # 开启SQL显示（打印路由后的SQL，调试必备）

# 数据源配置（主库+从库）
dataSources:
  # 主库数据源
  master_ds:
    type: com.zaxxer.hikari.HikariDataSource
    driverClassName: com.mysql.cj.jdbc.Driver
    jdbcUrl: jdbc:mysql://192.168.1.100:3306/sharding_db?useSSL=false&serverTimezone=UTC&allowPublicKeyRetrieval=true
    username: root
    password: root123
    connectionTimeout: 30000
    idleTimeout: 60000
    maxLifetime: 1800000
    maximumPoolSize: 50
  
  # 从库1数据源
  slave_ds_0:
    type: com.zaxxer.hikari.HikariDataSource
    driverClassName: com.mysql.cj.jdbc.Driver
    jdbcUrl: jdbc:mysql://192.168.1.101:3306/sharding_db?useSSL=false&serverTimezone=UTC&allowPublicKeyRetrieval=true
    username: root
    password: root123
    connectionTimeout: 30000
    idleTimeout: 60000
    maxLifetime: 1800000
    maximumPoolSize: 50
  
  # 从库2数据源（可扩展多个从库）
  slave_ds_1:
    type: com.zaxxer.hikari.HikariDataSource
    driverClassName: com.mysql.cj.jdbc.Driver
    jdbcUrl: jdbc:mysql://192.168.1.102:3306/sharding_db?useSSL=false&serverTimezone=UTC&allowPublicKeyRetrieval=true
    username: root
    password: root123
    connectionTimeout: 30000
    idleTimeout: 60000
    maxLifetime: 1800000
    maximumPoolSize: 50

# 读写分离规则配置
rules:
- !READWRITE_SPLITTING
  dataSources:
    # 读写分离数据源名称（应用实际访问的数据源名）
    sharding_rw_ds:
      type: Static
      props:
        write-data-source-name: master_ds  # 写操作路由至主库
        read-data-source-names: slave_ds_0,slave_ds_1  # 读操作路由至两个从库
        load-balance-algorithm-type: ROUND_ROBIN  # 从库负载均衡算法（轮询）
        # 可选：主从延迟控制（超过500ms则读主库）
        # read-write-splitting-delay-threshold: 500

# 绑定表规则（无分表时可省略，此处仅作示例）
bindingTables:
- t_user

# 默认数据源（当SQL未匹配任何规则时，路由至该数据源）
defaultDataSourceName: sharding_rw_ds

# Linux启动命令（指定配置文件目录）
./start.sh ../conf

[INFO ] 2024-05-20 10:00:00.000 [main] o.a.s.p.frontend.ShardingSphereProxy - ShardingSphere-Proxy start success.

-- 1. 写操作（INSERT）：查看ShardingProxy日志，确认路由至master_ds
INSERT INTO t_user (username, age) VALUES ('wangwu', 35);

-- 2. 读操作（SELECT）：查看日志，确认轮询路由至slave_ds_0、slave_ds_1
SELECT * FROM t_user;

-- 3. 强制读主库（Hint语法）：核心业务读请求用此方式保障一致性
/*!SHARDINGSPHERE_HINT: DATA_SOURCE_NAME=master_ds */ 
SELECT * FROM t_user WHERE id = 3;

<dependencies>
    <!-- SpringBoot JDBC Starter -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-jdbc</artifactId>
    </dependency>
    
    <!-- MySQL驱动 -->
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>8.0.32</version>
        <scope>runtime</scope>
    </dependency>
    
    <!-- 数据库连接池（HikariCP） -->
    <dependency>
        <groupId>com.zaxxer</groupId>
        <artifactId>HikariCP</artifactId>
    </dependency>
    
    <!-- SpringBoot Test（用于测试） -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

spring:
  datasource:
    type: com.zaxxer.hikari.HikariDataSource
    driver-class-name: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql://192.168.1.103:3307/sharding_db?useSSL=false&serverTimezone=UTC&allowPublicKeyRetrieval=true
    username: root
    password: 123456
    hikari:
      maximum-pool-size: 20
      minimum-idle: 5
      connection-timeout: 30000
      idle-timeout: 600000

public class User {
    private Long id;
    private String username;
    private Integer age;

    // 无参构造、有参构造、getter、setter
    public User() {}
    public User(String username, Integer age) {
        this.username = username;
        this.age = age;
    }

    // getter和setter
    public Long getId() { return id; }
    public void setId(Long id) { this.id = id; }
    public String getUsername() { return username; }
    public void setUsername(String username) { this.username = username; }
    public Integer getAge() { return age; }
    public void setAge(Integer age) { this.age = age; }
}

import org.springframework.jdbc.core.BeanPropertyRowMapper;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Repository;
import javax.annotation.Resource;
import java.util.List;

@Repository
public class UserDao {
    @Resource
    private JdbcTemplate jdbcTemplate;

    // 写操作：新增用户（路由至主库）
    public int addUser(User user) {
        String sql = "INSERT INTO t_user (username, age) VALUES (?, ?)";
        return jdbcTemplate.update(sql, user.getUsername(), user.getAge());
    }

    // 读操作：查询所有用户（路由至从库，轮询）
    public List<User> listAllUsers() {
        String sql = "SELECT id, username, age FROM t_user";
        return jdbcTemplate.query(sql, new BeanPropertyRowMapper<>(User.class));
    }

    // 读操作：强制读主库（核心业务，保障数据一致性）
    public User getUserById(Long id) {
        // Hint语法：强制路由至主库
        String sql = "/*!SHARDINGSPHERE_HINT: DATA_SOURCE_NAME=master_ds */ " +
                     "SELECT id, username, age FROM t_user WHERE id = ?";
        return jdbcTemplate.queryForObject(sql, new BeanPropertyRowMapper<>(User.class), id);
    }
}

import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import javax.annotation.Resource;
import java.util.List;
import static org.junit.jupiter.api.Assertions.*;

@SpringBootTest
public class UserDaoTest {
    @Resource
    private UserDao userDao;

    // 测试写操作（路由至主库）
    @Test
    public void testAddUser() {
        User user = new User("zhaoliu", 40);
        int rows = userDao.addUser(user);
        assertEquals(1, rows); // 新增成功返回1
    }

    // 测试读操作（路由至从库，轮询）
    @Test
    public void testListAllUsers() {
        List<User> userList = userDao.listAllUsers();
        assertNotNull(userList);
        System.out.println("查询用户数量：" + userList.size());
        // 连续查询3次，观察ShardingProxy日志，确认从库轮询（slave_ds_0 → slave_ds_1 → slave_ds_0）
        for (int i = 0; i < 3; i++) {
            userDao.listAllUsers();
        }
    }

    // 测试强制读主库（保障数据一致性）
    @Test
    public void testGetUserById() {
        // 先新增用户（写主库）
        User user = new User("qianqi", 28);
        userDao.addUser(user);
        // 立即查询刚新增的用户（强制读主库，避免主从延迟导致查不到数据）
        User result = userDao.getUserById(user.getId());
        assertNotNull(result);
        assertEquals("qianqi", result.getUsername());
    }
}


## 五、常见问题与解决方案
### 5.1 主从数据不一致
#### 问题表现
应用执行“写主库后立即读从库”时，可能查询不到最新数据（因主从复制存在延迟）。

#### 解决方案
1. **强制路由主库**：对核心业务场景（如用户注册后立即查询用户信息），使用Hint语法强制读主库：
   ```sql
   /*!SHARDINGSPHERE_HINT: DATA_SOURCE_NAME=master_ds */ 
   SELECT * FROM t_user WHERE id = ?

props:
  read-write-splitting-delay-threshold: 500  # 从库延迟>500ms时读主库

dataSources:
  slave_ds_0:
    # 其他配置省略...
    health:
      check:
        enable: true  # 开启健康检查
        timeout: 3000  # 检查超时时间（毫秒）
        interval: 5000  # 检查间隔（毫秒）
  slave_ds_1:
    # 同上配置...

rules:
- !READWRITE_SPLITTING
  dataSources:
    sharding_rw_ds:
      # 其他配置省略...
      props:
        # 自定义SQL路由规则：将包含FOR UPDATE的SELECT视为写操作
        sql-parser-rule: "SELECT.*FOR UPDATE"
        sql-parser-rule-type: WRITE  # 匹配上述规则的SQL视为写操作