輕量級binlog同步工具包binlogportal

使用binlog的緣由

近期須要重構一個老系統，須要從幾個服務中實時同步訂單的修改到重構表裏。
這裏就面臨兩個選擇，

1. 在每一個服務的mysql操做前埋點，發送修改信息到隊列或服務上。這種方案須要修改多個服務的代碼而且測試對原系統的影響，有額外開發和測試成本。
2. 同步mysql的binlog，根據表的insert和update更新新表。可是須要維護一個binlog同步的服務

本次選擇了binlog同步的方式。搭建的binlog服務也能夠用在以後新系統的緩存更新和同步ES索引上，相對於埋點這種只有一次性做用的方式，性價比較高。

工具調研：canal和mysql-binlog-connector-java

1.canal

要實現binlog同步服務，使用較多的開源方案是canal，運行比較穩定，並且功能也很豐富。

可是在實際部署服務的時候，遇到了一些問題：

• canal採用了client-server的模式，至少須要部署兩個集羣。咱們的項目都是使用私有云部署，爲了穩定運行，就有額外的資源和維護開銷。

  • 後來發現canal server能夠投遞信息到kafka。可是咱們的消息隊列是自研的，只能嘗試去改源碼。
• canal的server是完整獨立的包，沒法直接用springboot嵌套。而咱們的基礎組件都依賴於springboot，好比監控，配置中心等。
• canal的HA部署使用的是zookeeper，很惋惜咱們並無可用的zookeper集羣，也沒有資源從新部署一個。
• 單機部署的時候，已經處理的binlog postion是保存在文件裏面的，咱們用的私有云docker，重啓後全丟失。

2.mysql-binlog-connector-java

調研同時也發現了另外一個binlog同步工具，mysql-binlog-connector-java。

這是一個開源的binlog同步工具，功能很簡單，就是接收binlog信息。做爲一個依賴jar能夠很容易在springboot中使用。

可是沒有對binlog的內容作格式化處理，使用很不方便。固然更沒有保存信息和分佈式部署功能。

自研工具包binlogportal

基於這些問題，咱們須要一個具備如下特性的binlog同步工具：

• 可使用springboot加載運行，具備較好的擴展性

  • 說白了就是做爲一個jar包，開放出接口能夠自定義處理binlog信息的方式
• 可使用redis實現binlog position的保存和分佈式部署

爲了知足這些條件，經過對mysql-binlog-connector-java封裝後，實現了自研的工具binlogportal。

• 提供了binlogportal-spring-boot-starter包，可以使用spring boot快速部署
• 使用redis保存binlog position信息，重啓後可從上次position位置開始
• 當前支持insert和update的結構化
• 提供默認的http事件處理器。可經過實現IEventHandler接口，自定義事件處理器
• 使用redis做爲分佈式協調器，可多機部署實現高可用

使用說明

Mysql配置

• Mysql須要開啓binlog並設置爲row模式
• 同步binlog使用的mysql帳號，須要添加REPLICATION權限，示例以下：

CREATE USER binlogportal IDENTIFIED BY '123456';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'binlogportal'@'%';
GRANT ALL PRIVILEGES ON *.* TO 'binlogportal'@'%';
FLUSH PRIVILEGES;

經過spring boot構建項目

• 直接依賴binlogportal-spring-boot-starter

<dependency>
  <groupId>com.insistingon.binlogportal</groupId>
  <artifactId>binlogportal-spring-boot-starter</artifactId>
  <version>1.0.5</version>
</dependency>

• 經過spring boot的application.yml配置啓動器

binlogportal:
  enable: true # 是否啓用autoconfig
  distributed-enable: true # 是否啓用分佈式部署
  distributed-redis: # distributed-enable爲true時，要提供一個redis做爲分佈式協調器
    host: 127.0.0.1
    port: 6379
    auth:
  position-redis: # 保存binlog position的redis，必須配置
    host: 127.0.0.1
    port: 6379
    auth:
  db-config: # 數據庫配置，能夠有多個，key自定義便可
    d1:
      host: 0.0.0.0
      port: 3306
      user-name: binlogportal
      password: 123456
      handler-list: [logEventHandler] # 該數據庫使用的事件處理器，名稱爲spring的bean name
  http-handler: # 啓用自帶的http事件處理器，可發送請求
    url-list: [http://127.0.0.1:8988/testit] # 要發送的url列表，http參數爲統一的格式
    result-callback: httpCallBack # 配置自定義的結果處理器，須要實現IHttpCallback接口，值爲bean name

• Starter啓動

  • spring boot autoconfig啓動成功後，會把BinlogPortalStarter的實例注入到IOC中
  • 項目中經過注入的方式獲取binlogPortalStarter使用
  • binlogPortalStarter.start()會爲每一個mysql庫建立一個線程處理binlog
  • 下面是使用CommandLineRunner啓動starter的一個例子

@Slf4j
@Component
public class BinlogSync implements CommandLineRunner {
    @Resource
    BinlogPortalStarter binlogPortalStarter;

    public void run(String... args) throws Exception {
        try {
            binlogPortalStarter.start();
        } catch (BinlogPortalException e) {
            log.error(e.getMessage(), e);
        }
    }
}

非spring boot項目

• 非spring boot項目，可使用基礎包

<dependency>
  <groupId>com.insistingon.binlogportal</groupId>
  <artifactId>binlogportal</artifactId>
  <version>1.0.5</version>
</dependency>

• 依賴後實現配置類BinlogPortalConfig和SyncConfig，傳入Starter中運行便可

public class TestClass{
 public static void main(String[] args) {
        SyncConfig syncConfig = new SyncConfig();
        syncConfig.setHost("0.0.0.0");
        syncConfig.setPort(3306);
        syncConfig.setUserName("binlogportal");
        syncConfig.setPassword("123456");

        BinlogPortalConfig binlogPortalConfig = new BinlogPortalConfig();
        binlogPortalConfig.addSyncConfig(syncConfig);

        RedisConfig redisConfig = new RedisConfig("127.0.0.1", 6379);
        RedisPositionHandler redisPositionHandler = new RedisPositionHandler(redisConfig);
        binlogPortalConfig.setPositionHandler(redisPositionHandler);

        binlogPortalConfig.setDistributedHandler(new RedisDistributedHandler(redisConfig));

        BinlogPortalStarter binlogPortalStarter = new BinlogPortalStarter();
        binlogPortalStarter.setBinlogPortalConfig(binlogPortalConfig);
        try {
            binlogPortalStarter.start();
        } catch (BinlogPortalException e) {
            e.printStackTrace();
        }
    }
}

2.分佈式部署實現

項目中高可用實現是基於redis的分佈式鎖。

每一個實例都會加載所有數據庫的配置，在建立binlog鏈接以前，先要獲取redis鎖，獲取鎖後會定時刷新鎖的過時時間。全部實例會定時從新搶鎖。

同一個mysql庫的binlog文件和position會保存在redis裏，若是一個實例宕機。新搶到鎖的實例在初始化時，會使用上個實例已保存的binlog信息繼續獲取。

項目源碼可在Git上查看。
項目的Git地址：https://github.com/dothetrick...

以上內容屬我的學習總結，若有不當之處，歡迎在評論中指正