2. 什么？你想跨数据库关联查询？

fdfv232

1. 简介

我们平时开发中可能会遇到这样的问题，现在分布式环境下每个服务对应的数据库都是独立的，每个应用使用的都是自己的数据库，或者项目现场我们的服务需要使用第三方的提供的数据，但是第三方直接把数据库信息扔给我们，让我们自己去查询，像这种情况我们一般就两种做法

• 在我们的服务中添加一个数据源然后添加持久层进行操作
  
• 另起一个服务，然后这个服务去连接第三方数据库最后提供服务
  

其实这两种方法本质是一样的，就是添加对应的数据源，然后添加一堆持久层的对象，最后在service中对多个数据源的结果各种组装，实现起来顶多就是麻烦点，难度到不高。这个过程中其实最麻烦的点是数据的组装过程，如果业务复杂，组装起来感觉写一堆毫无意义的代码，也没什么重复利用的价值。那么有没有一种办法能直接跨库查询，比如将A库和B库的表直接进行连表查询，最好的是A库和B库即使不是同一种数据库，也能进行关联查询。
ok，calcite它来了。
2. 实现思路

我们在上一篇文章中讲了calcite如何建立元数据，在测试代码中其实已经实现了跨库的关联查询，本章就利用calcite的特性简单封装一个小demo，让其能提供跨库查询能力。

• 声明两个类
  
  
  
  • DataSourceProperty: 配置jdbc连接信息
    
  • DataSourceManager: 管理jdbc的DataSource对象
    
  ok，有了以上两个类，就可以做基本的jdbc的数据源管理了，其实只要能拿到DataSource，我们就可以做基本的数据库操作了，但是现在还不具备跨库查询的能力。
  
  
• 声明SuperDataSourceManager，将DataSource对象注册给calcite。
  其实到这里，我们就可以做跨库查询了，但是不好用，因为直接使用Statement，不管是参数封装还是执行结果的解析，都太原生了（不好用）
  
• 声明SuperDataSourceTemplate，类似于spring的JdbcTemplate，用来简化参数替换和结果解析（这里只是添加一个示例的模板代码）
  

3. Maven

<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0"         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">    <modelVersion>4.0.0</modelVersion>    <groupId>org.example</groupId>    <artifactId>super-query</artifactId>    <version>1.0-SNAPSHOT</version>    <parent>        <groupId>org.springframework.boot</groupId>        <artifactId>spring-boot-starter-parent</artifactId>        <version>2.7.3</version>        <relativePath/>    </parent>    <dependencies>        <dependency>            <groupId>org.springframework.boot</groupId>            <artifactId>spring-boot-starter-web</artifactId>        </dependency>        <dependency>            <groupId>org.springframework.boot</groupId>            <artifactId>spring-boot-starter-jdbc</artifactId>        </dependency>        <dependency>            <groupId>org.springframework.boot</groupId>            <artifactId>spring-boot-starter-test</artifactId>            <scope>test</scope>        </dependency>        <dependency>            <groupId>com.google.guava</groupId>            <artifactId>guava</artifactId>            <version>31.1-jre</version>        </dependency>        <dependency>            <groupId>org.projectlombok</groupId>            <artifactId>lombok</artifactId>            <version>1.18.24</version>        </dependency>        <dependency>            <groupId>org.apache.commons</groupId>            <artifactId>commons-lang3</artifactId>            <version>3.12.0</version>        </dependency>        <dependency>            <groupId>org.apache.commons</groupId>            <artifactId>commons-collections4</artifactId>            <version>4.4</version>        </dependency>        <dependency>            <groupId>org.apache.calcite</groupId>            <artifactId>calcite-core</artifactId>            <version>1.36.0</version>        </dependency>        <dependency>            <groupId>mysql</groupId>            <artifactId>mysql-connector-java</artifactId>        </dependency>        <dependency>            <groupId>org.postgresql</groupId>            <artifactId>postgresql</artifactId>            <version>42.2.23</version>        </dependency>    </dependencies></project>4. 核心代码


4.1 数据源管理器

DataSourceProperty：配置jdbc的连接信息
package com.ldx.superquery.datasource;import com.zaxxer.hikari.HikariDataSource;import lombok.Data;import org.apache.commons.lang3.StringUtils;import org.springframework.boot.jdbc.DatabaseDriver;import javax.sql.DataSource;/** * 数据源连接信息 */@Datapublic class DataSourceProperty {        /**         * URL         */        private String url;        /**         * 用户名         */        private String username;        /**         * 密码         */        private String password;        /**         * 数据源key         */        private String key;        /**         * 最多返回条数         */        private int maxRows = -1;        /**         * 驱动类         */        private String driverClassName;        /**         * 连接池类型         */        private String type;        public String getDriverClassName() {                if (StringUtils.isNotBlank(driverClassName)) {                        return driverClassName;                }                return DatabaseDriver.fromJdbcUrl(url).getDriverClassName();        }        public Class<? extends DataSource> getTypeClass() {                try {                        //noinspection unchecked                        return (Class<? extends DataSource>) Class.forName(type);                }                catch (Exception e) {                        return HikariDataSource.class;                }        }}DataSourceManager：管理jdbc的连接信息，并且最后注册给calcite
package com.ldx.superquery.datasource;import com.google.common.collect.Maps;import lombok.Data;import org.springframework.boot.jdbc.DataSourceBuilder;import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;import org.springframework.jdbc.datasource.DataSourceTransactionManager;import org.springframework.transaction.PlatformTransactionManager;import javax.sql.DataSource;import java.util.Map;/** * 数据源管理器 */public class DataSourceManager {    private final Map<String, DataSourceNode> DATA_SOURCE_MAP = Maps.newConcurrentMap();    public void register(String dataSourceKey, DataSource dataSource) {        this.register(new DataSourceNode(dataSourceKey, dataSource));    }    public void register(DataSourceProperty dataSourceProperty) {        final DataSourceNode dataSourceNode = new DataSourceNode(dataSourceProperty);        this.register(dataSourceNode);    }    public void register(DataSourceNode dataSourceNode) {        final String dataSourceKey = dataSourceNode.getDataSourceKey();        DATA_SOURCE_MAP.put(dataSourceKey, dataSourceNode);        SuperDataSourceManager.register(dataSourceKey, dataSourceNode.getDataSource());    }    public DataSourceNode getDataSource(String dataSourceKey) {        return DATA_SOURCE_MAP.get(dataSourceKey);    }    public void unregister(String dataSourceKey) {        DATA_SOURCE_MAP.remove(dataSourceKey);    }    /**     * 用来二次封装 datasource     */    @Data    public static class DataSourceNode {        private String dataSourceKey;        private DataSource dataSource;                // spring内置的named template        private NamedParameterJdbcTemplate jdbcTemplate;                // 事务管理器        private PlatformTransactionManager platformTransactionManager;        public DataSourceNode(String dataSourceKey, DataSource dataSource) {            this.dataSourceKey = dataSourceKey;            this.dataSource = dataSource;            this.jdbcTemplate = new NamedParameterJdbcTemplate(dataSource);            this.platformTransactionManager = new DataSourceTransactionManager(dataSource);        }        public DataSourceNode(DataSourceProperty dataSourceProperty) {            this(dataSourceProperty.getKey(), DataSourceBuilder                    .create()                    .url(dataSourceProperty.getUrl())                    .username(dataSourceProperty.getUsername())                    .password(dataSourceProperty.getPassword())                    .driverClassName(dataSourceProperty.getDriverClassName())                    .type(dataSourceProperty.getTypeClass())                    .build());        }    }}4.2 超级数据源管理器

SuperDataSourceManager： 用来管理calcite相关的连接信息
package com.ldx.superquery.datasource;import org.apache.calcite.adapter.jdbc.JdbcSchema;import org.apache.calcite.jdbc.CalciteConnection;import org.apache.calcite.schema.Schema;import org.apache.calcite.schema.SchemaPlus;import javax.sql.DataSource;import java.sql.Connection;import java.sql.DriverManager;import java.sql.SQLException;import java.sql.Statement;import java.util.Properties;/** * 超级数据源管理器 */public class SuperDataSourceManager {    private static final SchemaPlus ROOT_SCHEMA;    private static final CalciteConnection CALCITE_CONNECTION;    static {        // see CalciteConnectionProperty        Properties info = new Properties();        info.setProperty("lex", "JAVA");        // 不区分大小写        info.setProperty("caseSensitive", "false");        Connection connection = null;        try {            connection = DriverManager.getConnection("jdbc:calcite:", info);            CALCITE_CONNECTION = connection.unwrap(CalciteConnection.class);        }        catch (SQLException e) {            throw new RuntimeException("create calcite connection failed", e);        }        ROOT_SCHEMA = CALCITE_CONNECTION.getRootSchema();    }    public static void register(String dataSourceKey, DataSource dataSource) {        Schema schema = JdbcSchema.create(ROOT_SCHEMA, dataSourceKey, dataSource, null, null);        ROOT_SCHEMA.add(dataSourceKey, schema);    }    public static Statement getStatement() {        try {            return CALCITE_CONNECTION.createStatement();        }         catch (SQLException e) {            throw new RuntimeException("create calcite statement failed", e);        }    }}4.3 JdbcTemplate

SuperJdbcTemplate : 一个简单的门面来提供一些常用的jdbc相关方法
package com.ldx.superquery.datasource;import org.springframework.jdbc.core.ColumnMapRowMapper;import org.springframework.jdbc.core.ResultSetExtractor;import org.springframework.jdbc.core.RowMapperResultSetExtractor;import java.sql.ResultSet;import java.sql.SQLException;import java.sql.Statement;import java.util.List;import java.util.Map;/** * 超级数据源template */public class SuperJdbcTemplate {    private final Statement statement;    public SuperJdbcTemplate(Statement statement) {        this.statement = statement;    }    public List<Map<String, Object>> queryForList(String sql) throws SQLException {        final ColumnMapRowMapper columnMapRowMapper = new ColumnMapRowMapper();        return this.query(sql, new RowMapperResultSetExtractor<>(columnMapRowMapper));    }    public <T> T query(String sql, ResultSetExtractor<T> rse) throws SQLException {        try (ResultSet resultSet = statement.executeQuery(sql)) {            return rse.extractData(resultSet);        }    }}5. 测试用例

5.1 数据库

数据库用到的还是我们的老演员：

5.2 测试用例代码

这里分别测试了单库的查询，也测试了跨库的连表查询
package com.ldx.superquery.datasource;import com.google.common.collect.Maps;import lombok.SneakyThrows;import lombok.extern.slf4j.Slf4j;import org.junit.jupiter.api.BeforeAll;import org.junit.jupiter.api.Test;import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;import java.util.List;import java.util.Map;/** * 测试超级查询 */@Slf4jpublic class SuperQueryTest {    private static final String MYSQL_KEY = "mysql_test";    private static final String POSTGRES_KEY = "postgres_test";    private static DataSourceManager dsm;    @BeforeAll    public static void given_datasource_manager() {        final DataSourceManager dataSourceManager = new DataSourceManager();        final DataSourceProperty mysqlDataSourceProperty = buildMysqlDataSourceProperty();        final DataSourceProperty postgresDataSourceProperty = buildPostgresDataSourceProperty();        dataSourceManager.register(mysqlDataSourceProperty);        dataSourceManager.register(postgresDataSourceProperty);        dsm = dataSourceManager;    }    @Test    public void should_return_records_when_use_spring_jdbc_for_mysql() {        final DataSourceManager.DataSourceNode ds = dsm.getDataSource(MYSQL_KEY);        final NamedParameterJdbcTemplate jdbcTemplate = ds.getJdbcTemplate();        final Map<String, Object> params = Maps.newHashMap();        params.put("id", 1);        final List<Map<String, Object>> result = jdbcTemplate.queryForList("select * from `user` where id = :id", params);        log.info("execute query for mysql datasource results: {}", result);    }    @Test    public void should_return_records_when_use_spring_jdbc_for_postgres() {        final DataSourceManager.DataSourceNode ds = dsm.getDataSource(POSTGRES_KEY);        final NamedParameterJdbcTemplate jdbcTemplate = ds.getJdbcTemplate();        final Map<String, Object> params = Maps.newHashMap();        params.put("role_key", 1);        final List<Map<String, Object>> result = jdbcTemplate.queryForList("select * from role where role_key = :role_key", params);        log.info("execute query for postgres datasource results: {}", result);    }    @Test    @SneakyThrows    public void should_return_records_when_use_super_jdbc_for_postgres() {        final SuperJdbcTemplate SuperJdbcTemplate = new SuperJdbcTemplate(SuperDataSourceManager.getStatement());        final List<Map<String, Object>> result = SuperJdbcTemplate.queryForList("select * from "+ POSTGRES_KEY +".role");        log.info("execute super query for postgres datasource results: {}", result);    }    @Test    @SneakyThrows    public void should_return_records_when_use_super_jdbc() {        final SuperJdbcTemplate SuperJdbcTemplate = new SuperJdbcTemplate(SuperDataSourceManager.getStatement());        final List<Map<String, Object>> result = SuperJdbcTemplate.queryForList("select * from "+ POSTGRES_KEY +".role r right join "+ MYSQL_KEY + ".`user` u on r.role_key = u.role_key");        log.info("execute super query for postgres datasource results: ");        result.forEach(item -> log.info(item.toString()));    }    private static DataSourceProperty buildMysqlDataSourceProperty() {        final DataSourceProperty dataSourceProperty = new DataSourceProperty();        dataSourceProperty.setDriverClassName("com.mysql.cj.jdbc.Driver");        dataSourceProperty.setKey(MYSQL_KEY);        dataSourceProperty.setUrl("jdbc:mysql://localhost:3306/test");        dataSourceProperty.setUsername("root");        dataSourceProperty.setPassword("123456");        return dataSourceProperty;    }    private static DataSourceProperty buildPostgresDataSourceProperty() {        final DataSourceProperty dataSourceProperty = new DataSourceProperty();        dataSourceProperty.setDriverClassName("org.postgresql.Driver");        dataSourceProperty.setKey(POSTGRES_KEY);        dataSourceProperty.setUrl("jdbc:postgresql://localhost:5432/test");        dataSourceProperty.setUsername("root");        dataSourceProperty.setPassword("123456");        return dataSourceProperty;    }}5.3 测试结果展示

这里只展示一下should_return_records_when_use_super_jdbc用例的执行