开源基于Canal的开源增量数据订阅&消费中间件 - 好文

CanalSync

canal 是阿里巴巴开源的一款基于数据库增量日志解析,提供增量数据订阅&消费,目前主要支持了MySQL(也支持mariaDB)。

我开发的这个CanalSync项目 https://github.com/yuzd/CanalSync
<https://github.com/yuzd/CanalSync> ==>觉得不错帮忙给个star谢谢

是基于canal-server之上的数据库同步&消费中间件，

用于可快速搭建消费canal-server的项目。目前我已实现并开源了如下：

* 数据消费传输到redis组件
* 数据消费传输到rabbitmq组件
* 数据消费传输到mysql数据库组件
<https://github.com/yuzd/CanalSync#%E7%A4%BA%E6%84%8F%E5%9B%BE>示意图

<https://camo.githubusercontent.com/e081cd3bfba19c08dd972feb1733031515d818df/68747470733a2f2f696d61676573342e632d63747269702e636f6d2f7461726765742f7a623032313530303030303079326f6b34333646312e706e67>

<https://github.com/yuzd/CanalSync#nuget>Nuget：

<https://github.com/yuzd/CanalSync#1-%E6%8E%A5%E6%94%B6canal-server%E7%9A%84%E6%B6%88%E6%81%AF%E4%B8%AD%E9%97%B4%E4%BB%B6>
1. 接收canal-server的消息中间件：

Install-Package Canal.Server

<https://github.com/yuzd/CanalSync#2-%E8%A7%A3%E6%9E%90canal-server%E6%B6%88%E6%81%AF%E8%BD%AC%E5%87%BA%E5%8F%AF%E6%89%A7%E8%A1%8Csql%E7%9A%84%E4%B8%AD%E9%97%B4%E4%BB%B6>
2. 解析canal-server消息转出可执行sql的中间件：

Install-Package Canal.SqlParse

<https://github.com/yuzd/CanalSync#%E5%A6%82%E4%BD%95%E4%BD%BF%E7%94%A8>如何使用

如果你需要写一个数据消费传输到XXXMQ，用不到反解析成sql的话，只需要引用 Canal.Server中间件。
如果你需要写一个数据消费传输到XXXdb，得用到反解析sql中间件，需要同时引用Canal.Server 和 Canal.SqlParse 这2个中间件。

<https://github.com/yuzd/CanalSync#canalserver-%E5%A6%82%E4%BD%95%E4%BD%BF%E7%94%A8>
Canal.Server 如何使用

* 引用 Canal.Server 并appsettings.json 配置canal-server的参数.如下图：

<https://camo.githubusercontent.com/3d3f25b3bbcdafa507be65b761a0dcf7202b3412/68747470733a2f2f696d61676573342e632d63747269702e636f6d2f7461726765742f7a62303731353030303030307963617467354634372e706e67>

参数说明：
<https://camo.githubusercontent.com/b5424af7b9ec911f11ccd2354c626bb686fc6948/68747470733a2f2f696d61676573342e632d63747269702e636f6d2f7461726765742f7a623074313530303030303078766e7834443846322e706e67>

* 创建一个消费类必须要实现： INotificationHandler 接口，例如叫TestHandler public class
TestHandler:INotificationHandler<CanalBody>{ public Task Handle(CanalBody
notification) { //写消费逻辑 return Task.CompletedTask;; } }
* 在startUp 使用并注册该消费类 //注册了之后 canal-server有新的消息就会进入到TestHandler的Handle方法
services.AddCanalService(produce => produce.RegisterSingleton<TestHandler>());

<https://github.com/yuzd/CanalSync#canalsqlparse-%E5%A6%82%E4%BD%95%E4%BD%BF%E7%94%A8>
Canal.SqlParse 如何使用

目前只实现了解析mysql的逻辑，未来会加入sqlserver的解析逻辑!!
//注册使用 connectionString是mysql的数据库连接字符串
services.AddMysqlParseService(connectionString); // 计划中还未实现
//services.AddSqlserverParseService(connectionString);
在类的构造方法可注入：

<https://camo.githubusercontent.com/7bcf4720f85a7016db3c518e9e72d819a42f085d/68747470733a2f2f696d61676573342e632d63747269702e636f6d2f7461726765742f7a62306431353030303030307875357071303742332e706e67>

如上图，代表将canal-server的数据直接在另外的mysql库里面执行，等于2个mysql数据进行互相同步。

<https://github.com/yuzd/CanalSync#%E6%AC%A7%E6%B4%B2%E4%B8%8E%E4%B8%AD%E5%9B%BD%E7%9A%842%E4%B8%AAmysql%E5%BA%93-%E4%BD%BF%E7%94%A8%E4%B8%8A%E8%BF%B0%E6%96%B9%E6%B3%95%E8%BF%9B%E8%A1%8C%E5%90%8C%E6%AD%A5%E7%9A%84%E6%B5%8B%E8%AF%95>
欧洲与中国的2个mysql库使用上述方法进行同步的测试

结果：同步速度在100~200qps
<https://camo.githubusercontent.com/df1b154ed9ed3df1311740275c86ddff477b438e/68747470733a2f2f696d61676573342e632d63747269702e636f6d2f7461726765742f7a623075313530303030303078766d697a463531432e706e67>

热门工具换一换