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]