fdfv232 发表于 2025-2-6 12:46:15

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

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用例的执行
页: [1]
查看完整版本: 2. 什么?你想跨数据库关联查询?