作者:小凱
沉淀、分享、成長,讓自己和他人都能有所收獲!
本文的宗旨在于通過簡單干凈實踐的方式教會讀者,配置出一套 Canal 工具服務,來同步分庫分表的數據到 Elasticsearch 文件夾系統中。同時在 SpringBoot 工程中,配置出兩套數據源,一套是 MySQL + MyBatis,一套是 Elasticsearch + MyBatis。【這是非常重要的設計手段】
雖然現在有 TiDB 這樣的分布式數據庫,但對于分庫分表 + 數據同步ES,依然是非常主流的方案。同時也有一部分是把分庫分表的數據同步到 TiDB 使用。
Github:https://github.com/alibaba/canal(opens new window)
#一、組件介紹
canal ,譯為水道/管道/溝渠,主要用途是基于 MySQL 數據庫增量日志解析,提供增量數據訂閱和消費。
早期阿里巴巴因為杭州和美國雙機房部署,存在跨機房同步的業務需求,實現方式主要是基于業務 trigger 獲取增量變更。從 2010 年開始,業務逐步嘗試數據庫日志解析獲取增量變更進行同步,由此衍生出了大量的數據庫增量訂閱和消費業務。
它的工作原理是,canal 模擬 MySQL slave 的交互協議,偽裝自己為 MySQL slave ,向 MySQL master 發送dump 協議。在 MySQL master 收到 dump 請求,開始推送 binary log 給 slave (即 canal ) 這樣 canal 再解析 binary log (binlog)進行配置分發,同步到 Elasticsearch 等系統中進行使用。
那么有了 canal 就可以把分庫分表的數據同步到 Elasticsearch,提供匯總查詢和聚合操作,也就不需要把輪訓每個分庫分表數據了。
二、測試預期
本文的案例會把MySQL,2庫4表的數據,通過 Sharding 分庫分表寫入數據后,同步到 Elasticsearch。分庫分表如下(環境安裝中會自動安裝數據庫和設置庫表);
三、環境安裝
為了讓讀者伙伴更加簡單的學習到這一項方案技能,這里把所需的環境都配置成一整套的 docker compose 腳本文件(ARM、AMD),你只要執行安裝即可。安全前注意,無論是本機還是云服務器都需要安裝 docker-ce(opens new window)
1. 環境腳本
- 打開 xfg-dev-tech-canal 工程,下面就是 docker compose 的執行腳本。
- mac/windows 如果安裝了 docker 可以直接點擊如圖的三角號安裝。如果是在 Linux 安裝了 docker 可以把 dev-ops 整個文件夾都上傳到云服務器,之后通過腳本;docker-compose -f xfg-dev-tech-canal-docker-compose.yml up -d 進行安裝。
1.1 開啟 binlog
mysql 數據同步需要創建一個 canal 的賬戶,之后還需要開啟 binlog 日志
- 在 mysql 配置文件夾中,設置了初始化授權的賬戶、導入的庫表,以及開啟 mysql-bin 和配置要采集的庫。
- 如果你有配置自己其他的庫要同步也可以如此配置。
1.2 庫表采集配置
- 本文選擇的是 es 同步方式,所以需要在 canal-adapter 中 es7 文件夾添加同步的庫表 yml 配置。
- 以及在 application.yml 中配置出需要鏈接的庫表以及同步的目標地址,也就是 es 的地址。【因為本文的案例是在同一個 docker compose 下安裝,所以直接用名稱 elsticsearch 即可訪問】
2. 運行狀態
- 安裝完成后可以進入 portainer 查看各個組件的運行,如果有哪個運行失敗了,可以點擊那個小文件的圖標,它可以查看日志。
3. 創建索引
在 doc/dev-ops/curl 下提供了創建 Elasticsearch 的腳本;你可以點擊執行或者直接復制執行,也可以復制導入到 ApiPost 里執行。
以上這些腳本是為了創建出數據庫表同步到 Elasticsearch 后對應的索引和映射的字段。文章下面會用到。
3.1 創建
curl -X PUT "127.0.0.1:9200/xfg_dev_tech.user_order" -H 'Content-Type: application/json' -d'
{"mappings": {"properties": {"_user_id":{"type": "text"},"_user_name":{"type": "text"},"_order_id":{"type": "text"},"_uuid":{"type": "text"},"_create_time":{"type": "date"},"_update_time":{"type": "date"}}}
}'
3.2 添加
curl -X PUT "127.0.0.1:9200/xfg_dev_tech.user_order/_mapping" -H 'Content-Type: application/json' -d'
{"properties": {"_sku_name": {"type": "text"}}
}'
3.3 刪除
curl -X DELETE "127.0.0.1:9200/xfg_dev_tech.user_order"
4. 創建索引(Kibana)
4.1 索引管理
地址:http://127.0.0.1:5601/app/management/kibana/indexPatterns(opens new window)
- 填寫完,點擊創建索引模式即可。
4.2 數據頁面
地址:http://127.0.0.1:5601/app/discover(opens new window)
- 等后面同步數據了以后,直接在這里點刷新就可以看到了。
5. 許可證
kibana 提供了免費30天的試用許可,安裝后可以使用 x-pack-sql-jdbc。它的好處是可以讓我們通過 MyBatis 的方式查詢 Elasticsearch 數據。
地址:http://127.0.0.1:5601/app/management/stack/license_management(opens new window)
Elasticsearch 提供了 x-pack-sql-jdbc,讓對 Elasticsearch 的查詢也可以像使用 MySQL 數據庫一樣通過 MyBatis 進行查詢。但這個 x-pack-sql-jdbc 是付費的,免費可以使用 30 天。之后你可以選擇使用重新安裝,破解,或者使用 Elasticsearch 的查詢方式。還可以自己開發一個 Elasticsearch JDBC,GitHub 上也有類似的組件。
使用時需要引入 POM 配置;
<!-- https://mvnrepository.com/artifact/org.elasticsearch.plugin/x-pack-sql-jdbc -->
<dependency><groupId>org.elasticsearch.plugin</groupId><artifactId>x-pack-sql-jdbc</artifactId><version>7.17.14</version>
</dependency>
三、工程配置
本節涉及到了簡明教程中所講解的 Sharding 分庫分表 (opens new window)的使用,因為我們需要把分庫分表的數據通過 canal 同步到 Elasticsearch。(也可以使用其他分庫分表組件)
在工程中配置一套 Sharding 分庫分表映射的 MyBatis MyBatis,在配置一套 Elasticsearch x-pack-sql-jdbc 數據源映射的 MyBatis Mapper。這樣可以讀寫分別走自己設定好的 Mapper 對象了。
1. 創建數據源
@Configuration
public class DataSourceConfig {@Configuration@MapperScan(basePackages = "cn.bugstack.xfg.dev.tech.infrastructure.dao.elasticsearch", sqlSessionFactoryRef = "elasticsearchSqlSessionFactory")static class ElasticsearchMyBatisConfig {@Bean("elasticsearchDataSource")@ConfigurationProperties(prefix = "spring.elasticsearch.datasource")public DataSource igniteDataSource(Environment environment) {return new EsDataSource();}@Bean("elasticsearchSqlSessionFactory")public SqlSessionFactory elasticsearchSqlSessionFactory(DataSource elasticsearchDataSource) throws Exception {SqlSessionFactoryBean factoryBean = new SqlSessionFactoryBean();factoryBean.setDataSource(elasticsearchDataSource);factoryBean.setMapperLocations(new PathMatchingResourcePatternResolver().getResources("classpath:/mybatis/mapper/elasticsearch/*.xml"));return factoryBean.getObject();}}@Configuration@MapperScan(basePackages = "cn.bugstack.xfg.dev.tech.infrastructure.dao.mysql", sqlSessionFactoryRef = "mysqlSqlSessionFactory")static class MysqlMyBatisConfig {@Bean("mysqlDataSource")@ConfigurationProperties(prefix = "spring.mysql.datasource")public DataSource mysqlDataSource(Environment environment) {return DataSourceBuilder.create().url(environment.getProperty("spring.mysql.datasource.url")).driverClassName(environment.getProperty("spring.mysql.datasource.driver-class-name")).build();}@Bean("mysqlSqlSessionFactory")public SqlSessionFactory mysqlSqlSessionFactory(DataSource mysqlDataSource) throws Exception {SqlSessionFactoryBean factoryBean = new SqlSessionFactoryBean();factoryBean.setDataSource(mysqlDataSource);factoryBean.setMapperLocations(new PathMatchingResourcePatternResolver().getResources("classpath:/mybatis/mapper/mysql/*.xml"));return factoryBean.getObject();}}}
- ElasticsearchMyBatisConfig 使用 EsDataSource 創建數據源和映射 MyBatis Mapper 文件。
- MysqlMyBatisConfig 使用 DataSourceBuilder 創建 Sharding 提供的數據源和映射 MyBatis Mapper 文件。
2. Mapper 映射
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="cn.bugstack.xfg.dev.tech.infrastructure.dao.elasticsearch.IElasticsearchUserOrderDao"><resultMap id="dataMap" type="cn.bugstack.xfg.dev.tech.infrastructure.po.UserOrderPO"><result column="_user_id" property="userId"/><result column="_user_name" property="userName"/><result column="_order_id" property="orderId"/><result column="_sku_name" property="skuName"/><result column="_update_time" property="updateTime"/><result column="_create_time" property="createTime"/></resultMap><select id="selectByUserId" resultMap="dataMap">select _user_id, _user_name, _order_id, _sku_namefrom "xfg_dev_tech.user_order"order by _update_timelimit 10</select></mapper>
- 這個是 Elasticsearch 映射的 Mapper 文件,映射的字段就是前面安裝環境的時候設置的索引和字段。現在你使用 Elasticsearch 就不用在工程中硬編碼查詢語句了,變得非常方便。
四、工程測試
1. 寫入數據
@Test
public void test_insert() throws InterruptedException {for (int i = 0; i < 3; i++) {UserOrderPO userOrderPO = UserOrderPO.builder().userName("小哥哥").userId("xfg_" + RandomStringUtils.randomAlphabetic(6)).userMobile("+86 13521408***").sku("13811216").skuName("《手寫MyBatis:漸進式源碼實踐》").orderId(RandomStringUtils.randomNumeric(11)).quantity(1).unitPrice(BigDecimal.valueOf(128)).discountAmount(BigDecimal.valueOf(50)).tax(BigDecimal.ZERO).totalAmount(BigDecimal.valueOf(78)).orderDate(new Date()).orderStatus(0).isDelete(0).uuid(UUID.randomUUID().toString().replace("-", "")).ipv4("127.0.0.1").ipv6("2001:0db8:85a3:0000:0000:8a2e:0370:7334".getBytes()).extData("{\"device\": {\"machine\": \"IPhone 14 Pro\", \"location\": \"shanghai\"}}").build();userOrderDao.insert(userOrderPO);Thread.sleep(100);}
}
- 循環插入3條數據,按需你可以設置更多條數據。
- 這里的用戶編號 userId 是隨機的,也是切分鍵的 ID,所以會在不同的庫表寫入數據。
2. 數據驗證
MySQL:http://127.0.0.1:8899/ (opens new window)docker compose 配置的管理后臺,可以 root/123456 登錄
Kibana:http://127.0.0.1:5601/app/discover (opens new window)查詢寫入的數據。
3. 查詢數據
@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest
public class UserOrderDaoTest {@Resourceprivate IElasticsearchUserOrderDao userOrderDao;@Testpublic void test() {List<UserOrderPO> userOrderPOS = userOrderDao.selectByUserId();log.info("測試結果:{}", JSON.toJSONString(userOrderPOS));}}
- 通過 Elasticsearch 走 x-pack-sql-jdbc 的方式再把數據查詢出來