Canal的使用

@Test
public void testReadUpdateData() throws Exception {
  //1.获取 canal 连接对象, destination:刚创建的实例名称
  CanalConnector canalConnector = CanalConnectors.newClusterConnector(
    "192.168.22.161:2181,192.168.22.162:2181,192.168.22.163:2181",
    "test-canal-tcp", "", "");
  //2、连接
  canalConnector.connect();
  //3、订阅数据库
  canalConnector.subscribe("canal-monitor.*");

  while (true) {
    // 每次获取1000条增量数据
    Message message = canalConnector.getWithoutAck(1000);
    // 获取Entry集合
    List<CanalEntry.Entry> entries = message.getEntries();
    // 判断集合是否为空,如果为空,则等待一会继续拉取数据
    if (entries.size() <= 0) {
      System.out.println("当次抓取没有数据，休息一会。。。。。。");
      Thread.sleep(1000);
    } else {
      // 遍历entries，单条解析
      for (CanalEntry.Entry entry : entries) {
        //1.获取表名
        String tableName = entry.getHeader().getTableName();
        //2.获取类型
        CanalEntry.EntryType entryType = entry.getEntryType();
        //3.获取序列化后的数据
        ByteString storeValue = entry.getStoreValue();
        //4.判断当前entryType类型是否为ROWDATA
        if (CanalEntry.EntryType.ROWDATA.equals(entryType)) {
          //5.反序列化数据
          CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);
          //6.获取当前事件的操作类型
          CanalEntry.EventType eventType = rowChange.getEventType();
          //7.获取数据集
          List<CanalEntry.RowData> rowDataList = rowChange.getRowDatasList();
          //8.遍历rowDataList，并打印数据集
          for (CanalEntry.RowData rowData : rowDataList) {
            JSONObject beforeData = new JSONObject();
            List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
            for (CanalEntry.Column column : beforeColumnsList) {
              beforeData.put(column.getName(), column.getValue());
            }

            JSONObject afterData = new JSONObject();
            List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
            for (CanalEntry.Column column : afterColumnsList) {
              afterData.put(column.getName(), column.getValue());
            }

            //数据打印
            System.out.println("Table:" + tableName +
                               ",EventType:" + eventType +
                               ",Before:" + beforeData +
                               ",After:" + afterData);
          }
        } else {
          System.out.println("当前操作类型为：" + entryType);
        }
      }
    }
  }
}