1. 程式人生 > >[Spring cloud 一步步實現廣告系統] 15. 使用開源元件監聽Binlog 實現增量索引準備

[Spring cloud 一步步實現廣告系統] 15. 使用開源元件監聽Binlog 實現增量索引準備

MySQL Binlog簡介
  • 什麼是binlog?

一個二進位制日誌,用來記錄對資料發生或潛在發生更改的SQL語句,並以而進行的形式儲存在磁碟中。

  • binlog 的作用?

最主要有3個用途:

  • 資料複製(主從同步)

    Mysql 的Master-Slave協議,讓Slave可以通過監聽binlog實現資料複製,達到資料一致性目的

  • 資料恢復

    通過mysqlbinlog工具恢復資料

  • 增量備份

  • Binlog 變數

    • log_bin (Binlog 開關,使用show variables like 'log_bin';檢視)

    • binlog_format (Binlog 日誌格式,使用show variables like 'binlog_format';

      檢視)

      日誌格式總共有三種:

      • ROW, 僅儲存記錄被修改的細節,不記錄SQL語句上下文相關資訊。(能清晰的記錄下每行資料的修改細節,不需要記錄上下文相關資訊,因此不會發生某些特定情況下的procedure、function以及trigger 的呼叫無法被準確複製的問題,任何情況下都可以被複制,且能加快從庫重放日誌的效率,保證從庫資料的一致性)
      • STATEMENT,每一條修改資料的SQL都會被記錄。(只記錄執行語句的細節和上下文環境,避免了記錄每一行的變化,在一些修改記錄較多的情況下,相比ROW型別能大大減少binlog的日誌量,節約IO,提高效能。還可以用於實時的還原,同時主從版本可以不一樣,從伺服器版本可以比主伺服器版本高)
      • MIXED, 上述2種的混合使用
  • Binlog 管理

    • show master logs; 檢視所有binlog的日誌列表
    • show master status; 檢視最後一個binlog日誌編號名稱,以及最後一個事件技術的位置(position)
    • Flush logs; 重新整理binlog,此刻開始產生一個新編號的binlog日誌檔案
    • reset master; 清空所有的binlog日誌
  • Binlog 相關SQL show binlog events[in 'log_name'][from position][limit [offset,]row_count]

    UTOOLS1565224352749.png

    UTOOLS1565224799877.png

  • 常用的Binlog event

    • QUERY - 與資料無關的操作,begin、drop table、truncate table等等
    • TABLE_MAP - 記錄下一個操作所對應的表資訊,儲存了資料庫名稱和表名稱
    • XID - 標記事務提交
    • WRITE_ROWS 插入資料,即insert操作
    • UPDATE_ROWS 更新資料,即update操作
    • DELETE_ROWS 刪除資料,即delete操作

Event包含header和data兩部分,header提供了event的建立時間,哪個伺服器等資訊,data部分提供的是針對該event的具體資訊,如具體資料的修改。

Tip: binlog不會記錄資料表的列名

在接下來的實現中,我們會將自己的系統包裝成一個假的Mysql Slave,通過開源工具mysql-binlog-connector-java來實現監聽binlog。

開源工具mysql-binlog-connector-java
  • 工具原始碼:Github傳送門

  • 元件使用

    1.加依賴

    <!-- binlog 日誌監聽,解析開源工具類庫 -->
    <dependency>
        <groupId>com.github.shyiko</groupId>
        <artifactId>mysql-binlog-connector-java</artifactId>
        <version>0.18.1</version>
    </dependency>
    

    2.建立一個測試介面

    package com.sxzhongf.ad.service;
    
    import com.github.shyiko.mysql.binlog.BinaryLogClient;
    import com.github.shyiko.mysql.binlog.event.DeleteRowsEventData;
    import com.github.shyiko.mysql.binlog.event.EventData;
    import com.github.shyiko.mysql.binlog.event.UpdateRowsEventData;
    import com.github.shyiko.mysql.binlog.event.WriteRowsEventData;
    
    import java.io.IOException;
    
    /**
     * BinlogServiceTest for 測試Mysql binlog 監控
     * {@code
     * Mysql8 連線提示 Client does not support authentication protocol requested by server; consider upgrading MySQL client 解決方法
     * USE mysql;
     * ALTER USER 'root'@'localhost' IDENTIFIED WITH mysql_native_password BY 'password';
     * FLUSH PRIVILEGES;
     * }
     *
     * @author <a href="mailto:[email protected]">Isaac.Zhang | 若初</a>
     */
    public class BinlogServiceTest {
    
        /**
         * --------Update-----------
         * UpdateRowsEventData{tableId=90, includedColumnsBeforeUpdate={0, 1, 2, 3, 4, 5, 6, 7}, includedColumns={0, 1, 2, 3, 4, 5, 6, 7}, rows=[
         *     {before=[11, 10, Test Bin Log, 1, Tue Jun 25 08:00:00 CST 2019, Tue Jun 25 08:00:00 CST 2019, Tue Jun 25 08:00:00 CST 2019, Tue Jun 25 08:00:00 CST 2019], after=[11, 10, zhangpan test Binlog, 1, Tue Jun 25 08:00:00 CST 2019, Tue Jun 25 08:00:00 CST 2019, Tue Jun 25 08:00:00 CST 2019, Tue Jun 25 08:00:00 CST 2019]}
         * ]}
         *
         * --------Insert-----------
         * WriteRowsEventData{tableId=91, includedColumns={0, 1, 2, 3, 4, 5, 6, 7}, rows=[
         *     [10, 11, ad unit test binlog, 1, 0, 1236.7655, Thu Jun 27 08:00:00 CST 2019, Thu Jun 27 08:00:00 CST 2019]
         * ]}
         */
    
        public static void main(String[] args) throws IOException {
    
    //        //構造BinaryLogClient,填充mysql連結資訊
            BinaryLogClient client = new BinaryLogClient("127.0.0.1", 3306,
                    "root", "12345678"
            );
    
            //設定需要讀取的Binlog的檔案以及位置,否則,client會從"頭"開始讀取Binlog並監聽
    //        client.setBinlogFilename("binlog.000035");
    //        client.setBinlogPosition();
    
            //給客戶端註冊監聽器,實現對Binlog的監聽和解析
            //event 就是監聽到的Binlog變化資訊,event包含header & data 兩部分
            client.registerEventListener(event -> {
                EventData data = event.getData();
                if (data instanceof UpdateRowsEventData) {
                    System.out.println("--------Update-----------");
                    System.out.println(data.toString());
                } else if (data instanceof WriteRowsEventData) {
                    System.out.println("--------Insert-----------");
                    System.out.println(data.toString());
                } else if (data instanceof DeleteRowsEventData) {
                    System.out.println("--------Delete-----------");
                    System.out.println(data.toString());
                }
            });
    
            client.connect();
        }
    }
    

    執行:

    八月 08, 2019 9:13:32 上午 com.github.shyiko.mysql.binlog.BinaryLogClient connect
    資訊: Connected to 127.0.0.1:3306 at binlog.000038/951 (sid:65535, cid:336)
    ...
    

    執行sql update ad_user set user_status=1 where user_id=10;

UTOOLS1565227012106.png

我們需要知道的是,我們的目的是實現對Mysql資料表的變更實現監聽,並解析成我們想要的格式,也就是我們的java物件。根據上面我們看到的監聽結果,我們知道了返回資訊的大概內容,既然我們已經學會了簡單的使用BinaryLogClient 來監聽binlog,接下來,我們需要定義一個監聽器,來實現我們自己的業務內容。

因為我們只需要Event中的內容,那麼我們也就只需要通過實現com.github.shyiko.mysql.binlog.BinaryLogClient.EventListener介面,來自定義一個監聽器實現我們的業務即可。通過Event的內容,來判定是否需要處理當前event以及如何處理。

構造解析binlog的模版檔案

我們監聽binlog來構造增量資料的根本原因,是為了將我們的廣告投放系統廣告檢索系統 業務解耦,由於我們的檢索系統中沒有定義資料庫以及資料表的相關,所以,我們通過定義一份模版檔案,通過解析模版檔案來得到我們需要的資料庫和表資訊,因為binlog的監聽是不區分是哪個資料庫和哪個資料表資訊的,我們可以通過模版來指定我們想要監聽的部分。

{
  "database": "advertisement",
  "tableList": [
    {
      "tableName": "ad_plan",
      "level": 2,
      "insert": [
        {
          "column": "plan_id"
        },
        {
          "column": "user_id"
        },
        {
          "column": "plan_status"
        },
        {
          "column": "start_date"
        },
        {
          "column": "end_date"
        }
      ],
      "update": [
        {
          "column": "plan_id"
        },
        {
          "column": "user_id"
        },
        {
          "column": "plan_status"
        },
        {
          "column": "start_date"
        },
        {
          "column": "end_date"
        }
      ],
      "delete": [
        {
          "column": "plan_id"
        }
      ]
    },
    {
      "tableName": "ad_unit",
      "level": 3,
      "insert": [
        {
          "column": "unit_id"
        },
        {
          "column": "unit_status"
        },
        {
          "column": "position_type"
        },
        {
          "column": "plan_id"
        }
      ],
      "update": [
        {
          "column": "unit_id"
        },
        {
          "column": "unit_status"
        },
        {
          "column": "position_type"
        },
        {
          "column": "plan_id"
        }
      ],
      "delete": [
        {
          "column": "unit_id"
        }
      ]
    },
    {
      "tableName": "ad_creative",
      "level": 2,
      "insert": [
        {
          "column": "creative_id"
        },
        {
          "column": "type"
        },
        {
          "column": "material_type"
        },
        {
          "column": "height"
        },
        {
          "column": "width"
        },
        {
          "column": "audit_status"
        },
        {
          "column": "url"
        }
      ],
      "update": [
        {
          "column": "creative_id"
        },
        {
          "column": "type"
        },
        {
          "column": "material_type"
        },
        {
          "column": "height"
        },
        {
          "column": "width"
        },
        {
          "column": "audit_status"
        },
        {
          "column": "url"
        }
      ],
      "delete": [
        {
          "column": "creative_id"
        }
      ]
    },
    {
      "tableName": "relationship_creative_unit",
      "level": 3,
      "insert": [
        {
          "column": "creative_id"
        },
        {
          "column": "unit_id"
        }
      ],
      "update": [
      ],
      "delete": [
        {
          "column": "creative_id"
        },
        {
          "column": "unit_id"
        }
      ]
    },
    {
      "tableName": "ad_unit_district",
      "level": 4,
      "insert": [
        {
          "column": "unit_id"
        },
        {
          "column": "province"
        },
        {
          "column": "city"
        }
      ],
      "update": [
      ],
      "delete": [
        {
          "column": "unit_id"
        },
        {
          "column": "province"
        },
        {
          "column": "city"
        }
      ]
    },
    {
      "tableName": "ad_unit_hobby",
      "level": 4,
      "insert": [
        {
          "column": "unit_id"
        },
        {
          "column": "hobby_tag"
        }
      ],
      "update": [
      ],
      "delete": [
        {
          "column": "unit_id"
        },
        {
          "column": "hobby_tag"
        }
      ]
    },
    {
      "tableName": "ad_unit_keyword",
      "level": 4,
      "insert": [
        {
          "column": "unit_id"
        },
        {
          "column": "keyword"
        }
      ],
      "update": [
      ],
      "delete": [
        {
          "column": "unit_id"
        },
        {
          "column": "keyword"
        }
      ]
    }
  ]
}

上面的模版檔案中,指定了一個數據庫為advertisement,大家可以方便新增多個監聽庫。在資料庫下面,我們監聽了幾個表的CUD操作以及每個操作所需要的欄位資訊。

  • 實現模版 —> Java Entity

    • 定義模版檔案對應的實體
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public class BinlogTemplate {
    		//單資料庫對應
        private String database;
      	//多表
        private List<JsonTable> tableList;
    }
    
    • 對應的json 中 table資訊
    /**
     * JsonTable for 用於表示template.json中對應的表資訊
     *
     * @author <a href="mailto:[email protected]">Isaac.Zhang | 若初</a>
     */
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public class JsonTable {
        private String tableName;
        private Integer level;
    
        private List<Column> insert;
        private List<Column> update;
        private List<Column> delete;
    
        @Data
        @AllArgsConstructor
        @NoArgsConstructor
        public static class Column {
            private String columnName;
        }
    }
    
    • 讀取的對應表資訊物件(最主要目的就是為了能將欄位索引 對映到 欄位名稱
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public class TableTemplate {
        private String tableName;
        private String level;
    
      	//操作型別 -> 多列
        private Map<OperationTypeEnum, List<String>> opTypeFieldSetMap = new HashMap<>();
    
        /**
         * Binlog日誌中 欄位索引 -> 欄位名稱 的一個轉換對映
         * 因為binlog中不會顯示更新的列名是什麼,它只會展示欄位的索引,因此我們需要實現一次轉換
         */
        private Map<Integer, String> posMap = new HashMap<>();
    }
    
    • 解析模版檔案到java物件
    @Data
    public class ParseCustomTemplate {
    
        private String database;
    
        /**
         * key -> TableName
         * value -> {@link TableTemplate}
         */
        private Map<String, TableTemplate> tableTemplateMap;
    
        public static ParseCustomTemplate parse(BinlogTemplate _template) {
            ParseCustomTemplate template = new ParseCustomTemplate();
            template.setDatabase(_template.getDatabase());
    
            for (JsonTable jsonTable : _template.getTableList()) {
                String name = jsonTable.getTableName();
                Integer level = jsonTable.getLevel();
    
                TableTemplate tableTemplate = new TableTemplate();
                tableTemplate.setTableName(name);
                tableTemplate.setLevel(level.toString());
                template.tableTemplateMap.put(name, tableTemplate);
    
                //遍歷操作型別對應的列資訊
                Map<OperationTypeEnum, List<String>> operationTypeListMap = tableTemplate.getOpTypeFieldSetMap();
    
                for (JsonTable.Column column : jsonTable.getInsert()) {
                    getAndCreateIfNeed(
                            OperationTypeEnum.ADD,
                            operationTypeListMap,
                            ArrayList::new
                    ).add(column.getColumnName());
                }
    
                for (JsonTable.Column column : jsonTable.getUpdate()) {
                    getAndCreateIfNeed(
                            OperationTypeEnum.UPDATE,
                            operationTypeListMap,
                            ArrayList::new
                    ).add(column.getColumnName());
                }
    
                for (JsonTable.Column column : jsonTable.getDelete()) {
                    getAndCreateIfNeed(
                            OperationTypeEnum.DELETE,
                            operationTypeListMap,
                            ArrayList::new
                    ).add(column.getColumnName());
                }
            }
    
            return template;
        }
    
        /**
         * 從Map中獲取物件,如果不存在,建立一個
         */
        private static <T, R> R getAndCreateIfNeed(T key, Map<T, R> map, Supplier<R> factory) {
            return map.computeIfAbsent(key, k -> factory.get());
        }
    }
    
    • 解析 欄位索引 -> 欄位名稱 的一個轉換對映

    首先,我們來看一下binlog的具體日誌資訊:

    --------Insert-----------
    WriteRowsEventData{tableId=91, includedColumns={0, 1, 2, 3, 4, 5, 6, 7}, rows=[
    [10, 11, ad unit test binlog, 1, 0, 1236.7655, Thu Jun 27 08:00:00 CST 2019, Thu Jun 27 08:00:00 CST 2019]
    --------Update-----------
    UpdateRowsEventData{tableId=81, includedColumnsBeforeUpdate={0, 1, 2, 3, 4, 5}, includedColumns={0, 1, 2, 3, 4, 5}, rows=[
        {before=[10, Isaac Zhang, 2D3ABB6F2434109A105170FB21D00453, 0, Fri Jun 21 15:07:53 CST 2019, Fri Jun 21 15:07:53 CST 2019], after=[10, Isaac Zhang, 2D3ABB6F2434109A105170FB21D00453, 1, Fri Jun 21 15:07:53 CST 2019, Fri Jun 21 15:07:53 CST 2019]}
    
    

    可以看到,在日誌中includedColumns只包含了{0, 1, 2, 3, 4, 5}位置資訊,那麼我們怎麼能知道它具體代表的是哪個欄位呢,接下來我們來實現這步對映關係,在實現之前,我們先來查詢一下資料庫中我們的表中欄位所處的具體位置:

    sql> SELECT table_schema,table_name,column_name,ordinal_position FROM information_schema.COLUMNS
    WHERE TABLE_SCHEMA = 'advertisement' AND TABLE_NAME='ad_user'
    

    UTOOLS1565311337937.png

    我們可以看到ordinal_position對應的是1-6,可是上面監聽到的binlog日誌索引是0-5,所以我們就可以看出來之間的對應關係。

    我們開始編碼實現,我們使用JdbcTemplate進行查詢資料庫資訊:

    @Slf4j
    @Component
    public class TemplateHolder {
        private ParseCustomTemplate template;
    
        private final JdbcTemplate jdbcTemplate;
    
        private String SQL_SCHEMA = "SELECT TABLE_SCHEMA,TABLE_NAME,COLUMN_NAME,ORDINAL_POSITION FROM information_schema.COLUMNS " +
                "WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ?";
    
        @Autowired
        public TemplateHolder(JdbcTemplate jdbcTemplate) {
            this.jdbcTemplate = jdbcTemplate;
        }
    
        /**
         * 需要在容器載入的時候,就載入資料資訊
         */
        @PostConstruct
        private void init() {
            loadJSON("template.json");
        }
    
        /**
         * 對外提供載入服務
         */
        public TableTemplate getTable(String tableName) {
            return template.getTableTemplateMap().get(tableName);
        }
    
        /**
         * 載入需要監聽的binlog json檔案
         */
        private void loadJSON(String path) {
            ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
            InputStream inputStream = classLoader.getResourceAsStream(path);
    
            try {
                BinlogTemplate binlogTemplate = JSON.parseObject(
                        inputStream,
                        Charset.defaultCharset(),
                        BinlogTemplate.class
                );
    
                this.template = ParseCustomTemplate.parse(binlogTemplate);
                loadMeta();
            } catch (IOException ex) {
                log.error((ex.getMessage()));
                throw new RuntimeException("fail to parse json file");
            }
        }
    
        /**
         * 載入元資訊
         * 使用表索引到列名稱的對映關係
         */
        private void loadMeta() {
            for (Map.Entry<String, TableTemplate> entry : template.getTableTemplateMap().entrySet()) {
                TableTemplate table = entry.getValue();
    
                List<String> updateFields = table.getOpTypeFieldSetMap().get(
                        OperationTypeEnum.UPDATE
                );
                List<String> insertFields = table.getOpTypeFieldSetMap().get(
                        OperationTypeEnum.ADD
                );
                List<String> deleteFields = table.getOpTypeFieldSetMap().get(
                        OperationTypeEnum.DELETE
                );
    
                jdbcTemplate.query(SQL_SCHEMA, new Object[]{
                                template.getDatabase(), table.getTableName()
                        }, (rs, i) -> {
                            int pos = rs.getInt("ORDINAL_POSITION");
                            String colName = rs.getString("COLUMN_NAME");
    
                            if ((null != updateFields && updateFields.contains(colName))
                                || (null != insertFields && insertFields.contains(colName))
                                || (null != deleteFields && deleteFields.contains(colName))) {
                                         table.getPosMap().put(pos - 1, colName);
                            }
                            return null;
                        }
                );
            }
        }
    }
    
    • 監聽binlog實現

      • 定義Event 解析所需要轉換的java物件
      @Data
      public class BinlogRowData {
      
          private TableTemplate tableTemplate;
      
          private EventType eventType;
      
          private List<Map<String, String>> before;
      
          private List<Map<String, String>> after;
      
      }
      
      • 定義binlog client BinaryLogClient
      /**
       * CustomBinlogClient for 自定義Binlog Client
       *
       * @author <a href="mailto:[email protected]">Isaac.Zhang | 若初</a>
       * @since 2019/6/27
       */
      @Slf4j
      @Component
      public class CustomBinlogClient {
      
          private BinaryLogClient client;
      
          private final BinlogConfig config;
          private final AggregationListener listener;
      
          @Autowired
          public CustomBinlogClient(BinlogConfig config, AggregationListener listener) {
              this.config = config;
              this.listener = listener;
          }
      
          public void connect() {
              new Thread(() -> {
                  client = new BinaryLogClient(
                          config.getHost(),
                          config.getPort(),
                          config.getUsername(),
                          config.getPassword()
                  );
      
                  if (!StringUtils.isEmpty(config.getBinlogName()) && !config.getPosition().equals(-1L)) {
                      client.setBinlogFilename(config.getBinlogName());
                      client.setBinlogPosition(config.getPosition());
                  }
      
                  try {
                      log.info("connecting to mysql start...");
                      client.connect();
                      log.info("connecting to mysql done!");
                  } catch (IOException e) {
                      e.printStackTrace();
                  }
              }).start();
          }
      
          public void disconnect() {
              try {
                  log.info("disconnect to mysql start...");
                  client.disconnect();
                  log.info("disconnect to mysql done!");
              } catch (IOException e) {
                  e.printStackTrace();
              }
          }
      }
      
      • 使用client註冊事件監聽器com.github.shyiko.mysql.binlog.BinaryLogClient.EventListener
      /**
       * Ilistener for 為了後續擴充套件不同的實現
       *
       * @author <a href="mailto:[email protected]">Isaac.Zhang | 若初</a>
       */
      public interface Ilistener {
      
          void register();
      
          void onEvent(BinlogRowData eventData);
      }
      
      • 監聽Binlog, 收集mysql binlog datas
      @Slf4j
      @Component
      public class AggregationListener implements BinaryLogClient.EventListener {
      
          private String dbName;
          private String tbName;
      
          private Map<String, Ilistener> listenerMap = new HashMap<>();
      
          @Autowired
          private TemplateHolder templateHolder;
      
          private String genKey(String dbName, String tbName) {
              return dbName + ":" + tbName;
          }
      
          /**
           * 根據表實現註冊資訊
           */
          public void register(String dbName, String tbName, Ilistener listener) {
              log.info("register : {}-{}", dbName, tbName);
              this.listenerMap.put(genKey(dbName, tbName), listener);
          }
      
          @Override
          public void onEvent(Event event) {
      
              EventType type = event.getHeader().getEventType();
              log.info("Event type: {}", type);
      
              //資料庫增刪改之前,肯定有一個table_map event 的binlog
              if (type == EventType.TABLE_MAP) {
                  TableMapEventData data = event.getData();
                  this.tbName = data.getTable();
                  this.dbName = data.getDatabase();
                  return;
              }
      
              //EXT_UPDATE_ROWS 是Mysql 8以上的type
              if (type != EventType.EXT_UPDATE_ROWS
                      && type != EventType.EXT_WRITE_ROWS
                      && type != EventType.EXT_DELETE_ROWS
                      ) {
                  return;
              }
      
              // 檢查表名和資料庫名是否已經正確填充
              if (StringUtils.isEmpty(dbName) || StringUtils.isEmpty(tbName)) {
                  log.error("Meta data got error. tablename:{},database:{}", tbName, dbName);
                  return;
              }
      
              //找出對應資料表敏感的監聽器
              String key = genKey(this.dbName, this.tbName);
              Ilistener ilistener = this.listenerMap.get(key);
              if (null == ilistener) {
                  log.debug("skip {}", key);
              }
      
              log.info("trigger event:{}", type.name());
      
              try {
                  BinlogRowData rowData = convertEventData2BinlogRowData(event.getData());
                  if (null == rowData) {
                      return;
                  }
                  rowData.setEventType(type);
                  ilistener.onEvent(rowData);
      
              } catch (Exception e) {
                  e.printStackTrace();
                  log.error(e.getMessage());
              } finally {
                  this.dbName = "";
                  this.tbName = "";
              }
          }
      
          /**
           * 解析Binlog資料到Java實體物件的對映
           *
           * @param data binlog
           * @return java 物件
           */
          private BinlogRowData convertEventData2BinlogRowData(EventData data) {
              TableTemplate tableTemplate = templateHolder.getTable(tbName);
              if (null == tableTemplate) {
                  log.warn("table {} not found.", tbName);
                  return null;
              }
      
              List<Map<String, String>> afterMapList = new ArrayList<>();
      
              for (Serializable[] after : getAfterValues(data)) {
                  Map<String, String> afterMap = new HashMap<>();
      
                  int columnLength = after.length;
                  for (int i = 0; i < columnLength; ++i) {
                      //取出當前位置對應的列名
                      String colName = tableTemplate.getPosMap().get(i);
                      //如果沒有,則說明不需要該列
                      if (null == colName) {
                          log.debug("ignore position: {}", i);
                          continue;
                      }
      
                      String colValue = after[i].toString();
                      afterMap.put(colName, colValue);
                  }
      
                  afterMapList.add(afterMap);
              }
      
              BinlogRowData binlogRowData = new BinlogRowData();
              binlogRowData.setAfter(afterMapList);
              binlogRowData.setTableTemplate(tableTemplate);
      
              return binlogRowData;
          }
      
          /**
           * 獲取不同事件的變更後資料
           * Add & Delete變更前資料假定為空
           */
          private List<Serializable[]> getAfterValues(EventData eventData) {
      
              if (eventData instanceof WriteRowsEventData) {
                  return ((WriteRowsEventData) eventData).getRows();
              }
      
              if (eventData instanceof UpdateRowsEventData) {
                  return ((UpdateRowsEventData) eventData).getRows()
                                                          .stream()
                                                          .map(Map.Entry::getValue)
                                                          .collect(Collectors.toList()
                                                          );
              }
      
              if (eventData instanceof DeleteRowsEventData) {
                  return ((DeleteRowsEventData) eventData).getRows();
              }
      
              return Collections.emptyList();
          }
      }
      
      • 解析binlog 資料物件BinlogRowData ,用於增量索引的後續處理
      /**
       * MysqlRowData for 簡化{@link BinlogRowData} 以方便實現增量索引的實現
       *
       * @author <a href="mailto:[email protected]">Isaac.Zhang | 若初</a>
       */
      @Data
      @AllArgsConstructor
      @NoArgsConstructor
      public class MysqlRowData {
      
          //實現多資料的時候,需要傳遞資料庫名稱
          //private String database;
          private String tableName;
          private String level;
          private OperationTypeEnum operationTypeEnum;
          private List<Map<String, String>> fieldValueMap = new ArrayList<>();
      }
      

      因為我們需要將Binlog EventType轉換為我們的操作型別OperationTypeEnum,所以,我們在OperationTypeEnum中新增一個轉換方法:

      public enum OperationTypeEnum {
      ...
          public static OperationTypeEnum convert(EventType type) {
              switch (type) {
                  case EXT_WRITE_ROWS:
                      return ADD;
                  case EXT_UPDATE_ROWS:
                      return UPDATE;
                  case EXT_DELETE_ROWS:
                      return DELETE;
                  default:
                      return OTHER;
              }
          }
      }
      

      我們還需要定義一個表包含的各個列名稱的java類,方便我們後期對資料表的CUD操作:

      package com.sxzhongf.ad.mysql.constant;
      
      import java.util.HashMap;
      import java.util.Map;
      
      /**
       * Constant for 各個列名稱的java類,方便我們後期對資料表的CUD操作
       *
       * @author <a href="mailto:[email protected]">Isaac.Zhang | 若初</a>
       */
      public class Constant {
      
          private static final String DATABASE_NAME = "advertisement";
      
          public static class AD_PLAN_TABLE_INFO {
      
              public static final String TABLE_NAME = "ad_plan";
      
              public static final String COLUMN_PLAN_ID = "plan_id";
              public static final String COLUMN_USER_ID = "user_id";
              public static final String COLUMN_PLAN_STATUS = "plan_status";
              public static final String COLUMN_START_DATE = "start_date";
              public static final String COLUMN_END_DATE = "end_date";
          }
      
          public static class AD_CREATIVE_TABLE_INFO {
      
              public static final String TABLE_NAME = "ad_creative";
      
              public static final String COLUMN_CREATIVE_ID = "creative_id";
              public static final String COLUMN_TYPE = "type";
              public static final String COLUMN_MATERIAL_TYPE = "material_type";
              public static final String COLUMN_HEIGHT = "height";
              public static final String COLUMN_WIDTH = "width";
              public static final String COLUMN_AUDIT_STATUS = "audit_status";
              public static final String COLUMN_URL = "url";
          }
      
          public static class AD_UNIT_TABLE_INFO {
      
              public static final String TABLE_NAME = "ad_unit";
      
              public static final String COLUMN_UNIT_ID = "unit_id";
              public static final String COLUMN_UNIT_STATUS = "unit_status";
              public static final String COLUNN_POSITION_TYPE = "position_type";
              public static final String COLUNN_PLAN_ID = "plan_id";
          }
      
          public static class RELATIONSHIP_CREATIVE_UNIT_TABLE_INFO {
      
              public static final String TABLE_NAME = "relationship_creative_unit";
      
              public static final String COLUMN_CREATIVE_ID = "creative_id";
              public static final String COLUMN_UNIT_ID = "unit_id";
          }
      
          public static class AD_UNIT_DISTRICT_TABLE_INFO {
      
              public static final String TABLE_NAME = "ad_unit_district";
      
              public static final String COLUMN_UNIT_ID = "unit_id";
              public static final String COLUMN_PROVINCE = "province";
              public static final String COLUMN_CITY = "city";
          }
      
          public static class AD_UNIT_KEYWORD_TABLE_INFO {
      
              public static final String TABLE_NAME = "ad_unit_keyword";
      
              public static final String COLUMN_UNIT_ID = "unit_id";
              public static final String COLUMN_KEYWORD = "keyword";
          }
      
          public static class AD_UNIT_HOBBY_TABLE_INFO {
      
              public static final String TABLE_NAME = "ad_unit_hobby";
      
              public static final String COLUMN_UNIT_ID = "unit_id";
              public static final String COLUMN_HOBBY_TAG = "hobby_tag";
          }
      
          //key -> 表名
          //value -> 資料庫名
          public static Map<String, String> table2db;
      
          static {
              table2db = new HashMap<>();
              table2db.put(AD_PLAN_TABLE_INFO.TABLE_NAME, DATABASE_NAME);
              table2db.put(AD_CREATIVE_TABLE_INFO.TABLE_NAME, DATABASE_NAME);
              table2db.put(AD_UNIT_TABLE_INFO.TABLE_NAME, DATABASE_NAME);
              table2db.put(RELATIONSHIP_CREATIVE_UNIT_TABLE_INFO.TABLE_NAME, DATABASE_NAME);
              table2db.put(AD_UNIT_DISTRICT_TABLE_INFO.TABLE_NAME, DATABASE_NAME);
              table2db.put(AD_UNIT_HOBBY_TABLE_INFO.TABLE_NAME, DATABASE_NAME);
              table2db.put(AD_UNIT_KEYWORD_TABLE_INFO.TABLE_NAME, DATABASE_NAME);
          }
      }