1. Canal接收来的binlog，可以发送到MQ，也可以使用客户端进行TCP连接后从Canal中取出。本节就演示一下如何通过客户端方式取出binlog。

启动Canal

按照官方网站的QuickStart快速启动Canal。

手动写一个Client

本节通过官方wiki来编写，大家可以下载github源码，到example目录中看相关最新Clinet使用方法。

总体步骤

  • 创建maven-quickstart工程
  • 导入canal依赖
  • 编写Demo程序

  • 启动程序并触发数据库变更

    操作细节

  • 创建maven-quickstart工程

image.png

  • 导入canal依赖

    1. <dependency>
    2.  <groupId>com.alibaba.otter</groupId>
    3.  <artifactId>canal.client</artifactId>
    4.  <version>1.1.0</version>
    5. </dependency>

    这里使用1.1.0的官方是这样用的，如果用1.1.5官方Demo好像有问题

  • 编写Demo程序 ```java

import java.net.InetSocketAddress; import java.util.List;

import com.alibaba.otter.canal.client.CanalConnectors; import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.Message;

public class SimpleCanalClientExample {

  1. public static void main(String args[]) {
  2.     // 创建链接,指定Canal地址
  3.     CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.2.5",
  4.             11111), "example",
  5.         "", "");
  7.     int batchSize = 1000;
  8.     int emptyCount = 0;
  9.     try {
  10.         connector.connect();
  11.         //订阅所有库所有表
  12.         connector.subscribe(".*\\..*");
  13.         connector.rollback();
  14.         int totalEmptyCount = 120;
  15.         while (emptyCount < totalEmptyCount) {
  16.             // 获取指定数量的数据
  17.             Message message = connector.getWithoutAck(batchSize);
  18.             long batchId = message.getId();
  19.             int size = message.getEntries().size();
  20.             if (batchId == -1 || size == 0) {
  21.                 emptyCount++;
  22.                 System.out.println("empty count : " + emptyCount);
  23.                 try {
  24.                     Thread.sleep(1000);
  25.                 } catch (InterruptedException e) {
  26.                 }
  27.             } else {
  28.                 //若有数据变动
  29.                 emptyCount = 0;
  30.                 // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
  31.                 //打印消息项
  32.                 printEntry(message.getEntries());
  33.             }
  35.             connector.ack(batchId); // 提交确认
  36.             // connector.rollback(batchId); // 处理失败, 回滚数据
  37.         }
  39.         System.out.println("empty too many times, exit");
  40.     } finally {
  41.         connector.disconnect();
  42.     }
  43. }
  45. /**
  46.  * 打印消息项
  47.  * @param entrys
  48.  */
  49. private static void printEntry(List<CanalEntry.Entry> entrys) {
  50.     for (CanalEntry.Entry entry : entrys) {
  51.         if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
  52.             continue;
  53.         }
  55.         CanalEntry.RowChange rowChage = null;
  56.         try {
  57.             //从entry的Stroe值中解析RowChange内容
  58.             rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
  59.         } catch (Exception e) {
  60.             throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
  61.                 e);
  62.         }
  64.         CanalEntry.EventType eventType = rowChage.getEventType();
  66.         //打印变化内容
  67.         System.out.println(String.format("================&gt; binlog[%s:%s] , name[%s,%s] , eventType : %s",
  68.             entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
  69.             entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
  70.             eventType));
  72.         for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
  73.             if (eventType == CanalEntry.EventType.DELETE) {
  74.                 printColumn(rowData.getBeforeColumnsList());
  75.             } else if (eventType == CanalEntry.EventType.INSERT) {
  76.                 printColumn(rowData.getAfterColumnsList());
  77.             } else {
  78.                 System.out.println("-------&gt; before");
  79.                 printColumn(rowData.getBeforeColumnsList());
  80.                 System.out.println("-------&gt; after");
  81.                 printColumn(rowData.getAfterColumnsList());
  82.             }
  83.         }
  84.     }
  85. }
  87. private static void printColumn(List<CanalEntry.Column> columns) {
  88.     for (CanalEntry.Column column : columns) {
  89.         System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
  90.     }
  91. }

} ```

  • 启动程序并触发数据库变更