【feature】 jdbc支持statement级别根据sql进行读写分离路由功能 · Pull Request !162 · openGauss/openGauss-connector-jdbc

## **实现功能：**
支持 openGauss-jdbc 驱动语句级别的读写分离路由功能。
openGauss-jdbc  驱动支持根据 sql 解析结果判断，若是读操作路由到从库，写操作路由到主库。

## **应用场景：**
1.根据 sql 语义判断需要路由到主库还是从库。目前 openGauss-jdbc 不支持根据 sql 语义区分读写操作，以此控制路由到主库和从库。
2.支持语句级别的负载均衡算法。目前 openGauss-jdbc 在同一个连接里路由到哪一个实例是固定的，只支持连接之间的负载均衡路由。

## **设计原理：**
当jdbc url配置 `enableStatementLoadBalance` 则启用该功能:
1.返回逻辑 connection,用于管理多个物理的 connections,用于控制路由到主库或不同的从库；
2.ReadWriteSplittingPgStatement 在执行 executeQuery() 等操作时，进行 sql 解析路由，根据路由结果获取具体的物理 connection,通过 connection 创建物理 statement,执行 sql。
3.ReadWriteSplittingPgPreparedStatement 实现和上面类似。

## **自测试用例：**
配置 openGauss 一主两从集群，创建 account 表，测试开启 enableStatementLoadBalance=true 之后的读写分离路由行为。
1.测试 roundrobin 路由算法
测例名称: roundRobinLoadBalanceTest
获取连接，多次创建 statement 和 preparedStatement 执行多次查询操作，会路由到从库，路由结果符合轮询算法。
多次创建 statement 执行 update 操作，会一直路由到主库。
测试结果符合预期。

2.测试 shuffle 路由算法
测例名称: shuffleLoadBalanceTest
获取连接，多次创建 statement 和 preparedStatement 执行多次查询操作，会路由随机到从库，路由结果符合 shuffle 算法。
多次创建 statement 执行 update 操作，会一直路由到主库。
测试结果符合预期。

3.测试 priority 路由算法
测例名称: priorityLoadBalanceTest
获取连接，多次创建 statement 和 preparedStatement 执行多次查询操作，会路由到从库，路由结果比例符合 priority 算法。
多次创建 statement 执行 update 操作，会一直路由到主库。
测试结果符合预期。

4.测试 targetServerType=master 路由
测例名称: targetServerTypeOfMasterTest
获取连接，多次创建 statement 和 preparedStatement 执行多次查询和修改操作，均只会路由到主库。

5.测试 targetServerType=slave 路由
测例名称: targetServerTypeOfSlaveTest
获取连接，多次创建 statement 和 preparedStatement 执行多次查询和修改操作，均只会路由到从库。
测试结果符合预期。

## 手动测试
6.手动测试：移除 `shardingsphere-parser-sql-engine` 依赖
期望：无法使用 `enableStatementLoadBalance=true` 功能，其余功能正常。
![](https://foruda.gitee.com/images/1697189839859026122/a36f07bd_1285944.png "屏幕截图")

7.手动测试：移除 `shardingsphere-parser-sql-opengauss` 依赖
期望：无法使用 `enableStatementLoadBalance=true` 功能，其余功能正常。
![输入图片说明](https://foruda.gitee.com/images/1697190235121685115/2a59f22d_1285944.png "屏幕截图")

8.添加完整所需的 shardingsphere 依赖，`enableStatementLoadBalance=true` 功能正常使用。
![输入图片说明](https://foruda.gitee.com/images/1697190411008172385/e7cc9376_1285944.png "屏幕截图")

## **自测试用例代码：**
```java
// 省略部分代码
@Test
    public void roundRobinLoadBalanceTest() throws SQLException {
        String urlParams = String.format("?enableStatementLoadBalance=true&autoBalance=roundrobin&writeDataSourceAddress=%s", getMasterHostSpec());
        String url = initURL(hostSpecs) + urlParams;
        Properties properties = new Properties();
        properties.setProperty("user", TestUtil.getUser());
        properties.setProperty("password", TestUtil.getPassword());
        try (Connection connection = getConnection(url, properties)) {
            Assert.assertThat(connection, instanceOf(ReadWriteSplittingPgConnection.class));
            ReadWriteSplittingPgConnection readWriteSplittingPgConnection = (ReadWriteSplittingPgConnection)connection;
            ReadWriteSplittingHostSpec readWriteSplittingHostSpec = readWriteSplittingPgConnection.getReadWriteSplittingHostSpec();
            Assert.assertEquals(readWriteSplittingHostSpec.getWriteHostSpec(), getMasterHostSpec());
            Assert.assertEquals(readWriteSplittingHostSpec.getReadHostSpecs(), getReadHostSpecs());
            try (Statement statement = connection.createStatement()) {
                Assert.assertTrue(statement.execute("select * from account where id = 1"));
                HostSpec actual = getRoutedReadHostSpec(readWriteSplittingPgConnection);
                for (int i = 0; i < readHostSpecs.length; i++) {
                    HostSpec firstExpected = getNextExpectedRoundRobinSpec();
                    if (firstExpected.equals(actual)) {
                        break;
                    }
                }
                Assert.assertTrue(isRoutedToReadHostSpecs(readWriteSplittingPgConnection));
            }
            for (int i = 0; i < 10; i++) {
                try (Statement statement = connection.createStatement()) {
                    Assert.assertTrue(statement.execute("select * from account where id = 1"));
                    HostSpec actual = getRoutedReadHostSpec(readWriteSplittingPgConnection);
                    Assert.assertEquals(getNextExpectedRoundRobinSpec(), actual);
                }    
            }
            for (int i = 0; i < 10; i++) {
                try (PreparedStatement statement = connection.prepareStatement("select * from account where id = ?")) {
                    statement.setString(1, "1");
                    Assert.assertTrue(statement.execute());
                    HostSpec actual = getRoutedReadHostSpec(readWriteSplittingPgConnection);
                    Assert.assertEquals(getNextExpectedRoundRobinSpec(), actual);
                }
            }
            for (int i = 0; i < 3; i++) {
                try (Statement statement = connection.createStatement()) {
                    statement.execute("update account set balance = 11 where id = 1");
                    Assert.assertTrue(isRoutedToWriteHostSpecs(readWriteSplittingPgConnection));
                }    
            }
        }
    }

@Test
    public void shuffleLoadBalanceTest() throws SQLException {
        String urlParams = String.format("?enableStatementLoadBalance=true&autoBalance=shuffle&writeDataSourceAddress=%s", getMasterHostSpec());
        String url = initURL(hostSpecs) + urlParams;
        Properties properties = new Properties();
        properties.setProperty("user", TestUtil.getUser());
        properties.setProperty("password", TestUtil.getPassword());
        try (Connection connection = getConnection(url, properties)) {
            Assert.assertThat(connection, instanceOf(ReadWriteSplittingPgConnection.class));
            ReadWriteSplittingPgConnection readWriteSplittingPgConnection = (ReadWriteSplittingPgConnection)connection;
            ReadWriteSplittingHostSpec readWriteSplittingHostSpec = readWriteSplittingPgConnection.getReadWriteSplittingHostSpec();
            Assert.assertEquals(readWriteSplittingHostSpec.getWriteHostSpec(), getMasterHostSpec());
            Assert.assertEquals(readWriteSplittingHostSpec.getReadHostSpecs(), getReadHostSpecs());
            for (int i = 0; i < 10; i++) {
                try (Statement statement = connection.createStatement()) {
                    Assert.assertTrue(statement.execute("select * from account where id = 1"));
                    Assert.assertTrue(isRoutedToReadHostSpecs(readWriteSplittingPgConnection));
                }
            }
            for (int i = 0; i < 10; i++) {
                try (PreparedStatement statement = connection.prepareStatement("select * from account where id = ?")) {
                    statement.setString(1, "1");
                    Assert.assertTrue(statement.execute());
                    Assert.assertTrue(isRoutedToReadHostSpecs(readWriteSplittingPgConnection));
                }
            }
            for (int i = 0; i < 3; i++) {
                try (Statement statement = connection.createStatement()) {
                    statement.execute("update account set balance = 11 where id = 1");
                    Assert.assertTrue(isRoutedToWriteHostSpecs(readWriteSplittingPgConnection));
                }
            }
        }
    }

@Test
    public void priorityLoadBalanceTest() throws SQLException {
        String urlParams = String.format("?enableStatementLoadBalance=true&autoBalance=priority2&writeDataSourceAddress=%s", getMasterHostSpec());
        String url = initURL(hostSpecs) + urlParams;
        Properties properties = new Properties();
        properties.setProperty("user", TestUtil.getUser());
        properties.setProperty("password", TestUtil.getPassword());
        try (Connection connection = getConnection(url, properties)) {
            Assert.assertThat(connection, instanceOf(ReadWriteSplittingPgConnection.class));
            ReadWriteSplittingPgConnection readWriteSplittingPgConnection = (ReadWriteSplittingPgConnection)connection;
            ReadWriteSplittingHostSpec readWriteSplittingHostSpec = readWriteSplittingPgConnection.getReadWriteSplittingHostSpec();
            Assert.assertEquals(readWriteSplittingHostSpec.getWriteHostSpec(), getMasterHostSpec());
            Assert.assertEquals(readWriteSplittingHostSpec.getReadHostSpecs(), getReadHostSpecs());
            HostSpec firstActual;
            try (Statement statement = connection.createStatement()) {
                Assert.assertTrue(statement.execute("select * from account where id = 1"));
                firstActual = getRoutedReadHostSpec(readWriteSplittingPgConnection);
            }
            for (int i = 0; i < 10; i++) {
                try (Statement statement = connection.createStatement()) {
                    Assert.assertTrue(statement.execute("select * from account where id = 1"));
                    HostSpec actual = getRoutedReadHostSpec(readWriteSplittingPgConnection);
                    Assert.assertEquals(firstActual, actual);
                }
            }
            for (int i = 0; i < 10; i++) {
                try (PreparedStatement statement = connection.prepareStatement("select * from account where id = ?")) {
                    statement.setString(1, "1");
                    Assert.assertTrue(statement.execute());
                    HostSpec actual = getRoutedReadHostSpec(readWriteSplittingPgConnection);
                    Assert.assertEquals(firstActual, actual);
                }
            }
            for (int i = 0; i < 3; i++) {
                try (Statement statement = connection.createStatement()) {
                    statement.execute("update account set balance = 11 where id = 1");
                    Assert.assertTrue(isRoutedToWriteHostSpecs(readWriteSplittingPgConnection));
                }
            }
        }
    }

@Test
    public void targetServerTypeOfMasterTest() throws SQLException {
        String urlParams = String.format("?enableStatementLoadBalance=true&autoBalance=shuffle&writeDataSourceAddress=%s&targetServerType=master", getMasterHostSpec());
        String url = initURL(hostSpecs) + urlParams;
        Properties properties = new Properties();
        properties.setProperty("user", TestUtil.getUser());
        properties.setProperty("password", TestUtil.getPassword());
        try (Connection connection = getConnection(url, properties)) {
            Assert.assertThat(connection, instanceOf(ReadWriteSplittingPgConnection.class));
            ReadWriteSplittingPgConnection readWriteSplittingPgConnection = (ReadWriteSplittingPgConnection)connection;
            ReadWriteSplittingHostSpec readWriteSplittingHostSpec = readWriteSplittingPgConnection.getReadWriteSplittingHostSpec();
            Assert.assertEquals(readWriteSplittingHostSpec.getWriteHostSpec(), getMasterHostSpec());
            Assert.assertEquals(readWriteSplittingHostSpec.getReadHostSpecs(), getReadHostSpecs());
            for (int i = 0; i < 10; i++) {
                try (Statement statement = connection.createStatement()) {
                    Assert.assertTrue(statement.execute("select * from account where id = 1"));
                    Assert.assertTrue(isRoutedToWriteHostSpecs(readWriteSplittingPgConnection));
                }
            }
            for (int i = 0; i < 10; i++) {
                try (PreparedStatement statement = connection.prepareStatement("select * from account where id = ?")) {
                    statement.setString(1, "1");
                    Assert.assertTrue(statement.execute());
                    Assert.assertTrue(isRoutedToWriteHostSpecs(readWriteSplittingPgConnection));
                }
            }
            for (int i = 0; i < 3; i++) {
                try (Statement statement = connection.createStatement()) {
                    statement.execute("update account set balance = 11 where id = 1");
                    Assert.assertTrue(isRoutedToWriteHostSpecs(readWriteSplittingPgConnection));
                }
            }
        }
    }

@Test
    public void targetServerTypeOfSlaveTest() throws SQLException {
        String urlParams = String.format("?enableStatementLoadBalance=true&autoBalance=shuffle&writeDataSourceAddress=%s&targetServerType=slave", getMasterHostSpec());
        String url = initURL(hostSpecs) + urlParams;
        Properties properties = new Properties();
        properties.setProperty("user", TestUtil.getUser());
        properties.setProperty("password", TestUtil.getPassword());
        try (Connection connection = getConnection(url, properties)) {
            Assert.assertThat(connection, instanceOf(ReadWriteSplittingPgConnection.class));
            ReadWriteSplittingPgConnection readWriteSplittingPgConnection = (ReadWriteSplittingPgConnection)connection;
            ReadWriteSplittingHostSpec readWriteSplittingHostSpec = readWriteSplittingPgConnection.getReadWriteSplittingHostSpec();
            Assert.assertEquals(readWriteSplittingHostSpec.getWriteHostSpec(), getMasterHostSpec());
            Assert.assertEquals(readWriteSplittingHostSpec.getReadHostSpecs(), getReadHostSpecs());
            for (int i = 0; i < 10; i++) {
                try (Statement statement = connection.createStatement()) {
                    Assert.assertTrue(statement.execute("select * from account where id = 1"));
                    Assert.assertTrue(isRoutedToReadHostSpecs(readWriteSplittingPgConnection));
                }
            }
            for (int i = 0; i < 10; i++) {
                try (PreparedStatement statement = connection.prepareStatement("select * from account where id = ?")) {
                    statement.setString(1, "1");
                    Assert.assertTrue(statement.execute());
                    Assert.assertTrue(isRoutedToReadHostSpecs(readWriteSplittingPgConnection));
                }
            }
            for (int i = 0; i < 3; i++) {
                try (Statement statement = connection.createStatement()) {
                    try {
                        statement.execute("update account set balance = 11 where id = 1");
                    } catch (SQLException e) {
                        Assert.assertTrue(e.getMessage().contains("ERROR: cannot execute UPDATE in a read-only transaction"));
                    }
                }
            }
        }
    }
```

openGauss/openGauss-connector-jdbc .gitee-modal { width: 500px !important; }

搜索帮助

openGauss/openGauss-connector-jdbc