ClientSample

    • 可以在eclipse里，直接打开com.alibaba.otter.canal.example.SimpleCanalClientTest，直接运行

    • 在工程的example目录下运行命令行：
      mvn exec:java -Dexec.mainClass="com.alibaba.otter.canal.example.SimpleCanalClientTest"

    • 下载example包: https://github.com/alibaba/canal/releases，解压缩后，直接运行sh startup.sh脚本c. 触发数据变更d. 在控制台或者logs中查看，可以看到如下信息 ：

    依赖配置：

    1. <dependency>
    2.     <groupId>com.alibaba.otter</groupId>
    3.     <artifactId>canal.client</artifactId>
    4.     <version>1.0.12</version>
    5. </dependency>
    1. 创建mvn标准工程：
    1. mvn archetype:create -DgroupId=com.alibaba.otter -DartifactId=canal.sample
    maven3.0.5以上版本舍弃了create，使用generate生成项目 
    1. mvn archetype:generate -DgroupId=com.alibaba.otter -DartifactId=canal.sample
    1. 修改pom.xml，添加依赖
    1. ClientSample代码
    1. package com.alibaba.otter.canal.sample;
    2.  
    3. import java.net.InetSocketAddress;
    4. import java.util.List;
    5.  
    6.  
    7. import com.alibaba.otter.canal.client.CanalConnectors;
    8. import com.alibaba.otter.canal.client.CanalConnector;
    9. import com.alibaba.otter.canal.common.utils.AddressUtils;
    10. import com.alibaba.otter.canal.protocol.Message;
    11. import com.alibaba.otter.canal.protocol.CanalEntry.Column;
    12. import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
    13. import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
    14. import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
    15. import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
    16. import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
    17.  
    18.  
    19. public class SimpleCanalClientExample {
    20.  
    21.  
    23.     // 创建链接
    24.     CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),
    25.                                                                                         11111), "example", "", "");
    26.     int batchSize = 1000;
    27.     int emptyCount = 0;
    28.     try {
    29.         connector.subscribe(".*\\..*");
    30.         connector.rollback();
    31.         int totalEmptyCount = 120;
    32.         while (emptyCount < totalEmptyCount) {
    33.             Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
    34.             long batchId = message.getId();
    35.             int size = message.getEntries().size();
    36.             if (batchId == -1 || size == 0) {
    37.                 emptyCount++;
    38.                 System.out.println("empty count : " + emptyCount);
    39.                 try {
    40.                     Thread.sleep(1000);
    41.                 } catch (InterruptedException e) {
    42.                 }
    43.             } else {
    45.                 // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
    46.                 printEntry(message.getEntries());
    49.             connector.ack(batchId); // 提交确认
    50.             // connector.rollback(batchId); // 处理失败, 回滚数据
    51.         }
    53.         System.out.println("empty too many times, exit");
    54.     } finally {
    55.         connector.disconnect();
    56.     }
    57. }
    59. private static void printEntry(List<Entry> entrys) {
    60.     for (Entry entry : entrys) {
    61.         if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
    62.             continue;
    63.         }
    65.         RowChange rowChage = null;
    66.         try {
    67.             rowChage = RowChange.parseFrom(entry.getStoreValue());
    68.         } catch (Exception e) {
    69.             throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
    70.                                        e);
    71.         }
    73.         System.out.println(String.format("================&gt; binlog[%s:%s] , name[%s,%s] , eventType : %s",
    74.                                          entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
    75.                                          entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
    76.                                          eventType));
    78.         for (RowData rowData : rowChage.getRowDatasList()) {
    79.             if (eventType == EventType.DELETE) {
    80.                 printColumn(rowData.getBeforeColumnsList());
    81.             } else if (eventType == EventType.INSERT) {
    82.                 printColumn(rowData.getAfterColumnsList());
    83.             } else {
    84.                 System.out.println("-------&gt; before");
    85.                 printColumn(rowData.getBeforeColumnsList());
    86.                 System.out.println("-------&gt; after");
    87.                 printColumn(rowData.getAfterColumnsList());
    88.             }
    89.         }
    90.     }
    91. }
    93. private static void printColumn(List<Column> columns) {
    94.     for (Column column : columns) {
    95.         System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
    96.     }
    98.  
    99.  
    100.  
    

      

    1. 运行Client
    

    首先启动Canal Server，可参见
    

    启动Canal Client后，可以从控制台从看到类似消息：
    

    1. empty count : 1
    2. empty count : 2
    3. empty count : 3
    4. empty count : 4
    

      

    1. 触发数据库变更
    

    可以从控制台中看到：
    

    1. empty count : 1
    2. empty count : 2
    3. empty count : 3
    4. empty count : 4
    5. ================> binlog[mysql-bin.001946:313661577] , name[test,xdual] , eventType : INSERT
    6. ID : 4    update=true
    7. X : 2013-02-05 23:29:46    update=true
    

     如果需要更详细的exmpale例子，请下载canal当前最新源码包，里面有个example工程，谢谢.