ClientSample

直接使用canal.example工程

a. 首先启动Canal Server，可参见[[QuickStart]] b.

  1. 可以在eclipse里，直接打开com.alibaba.otter.canal.example.SimpleCanalClientTest，直接运行
  2. 在工程的example目录下运行命令行：

java 1. mvn exec:java -Dexec.mainClass=”com.alibaba.otter.canal.example.SimpleCanalClientTest” 3. 下载example包: <https://github.com/alibaba/canal/releases，解压缩后，直接运行sh> startup.sh脚本

c. 触发数据变更 d. 在控制台或者logs中查看，可以看到如下信息 ：

1. ================> binlog[mysql-bin.002579:508882822] , name[retl,xdual] , eventType : UPDATE , executeTime : 1368607728000 , delay : 4270ms
2. ———-> before
3. ID : 1    update=false
4. X : 2013-05-15 11:43:42    update=false
5. ———-> after
6. ID : 1    update=false
7. X : 2013-05-15 16:48:48    update=true

从头创建工程

依赖配置：

1. <dependency>
2. <groupId>com.alibaba.otter</groupId>
3. <artifactId>canal.client</artifactId>
4. <version>1.1.0</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，添加依赖

  2. ClientSample代码

1. package com.alibaba.otter.canal.sample;
2. import java.net.InetSocketAddress;
3. import java.util.List;
4. import com.alibaba.otter.canal.client.CanalConnectors;
5. import com.alibaba.otter.canal.client.CanalConnector;
6. import com.alibaba.otter.canal.common.utils.AddressUtils;
7. import com.alibaba.otter.canal.protocol.Message;
8. import com.alibaba.otter.canal.protocol.CanalEntry.Column;
9. import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
10. import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
11. import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
12. import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
13. import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
14. public class SimpleCanalClientExample {
15. ```
    public static void main(String args[]) {
    ```
16. ```
        // 创建链接
    ```
17. ```
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),
    ```
18. ```
                                                                                            11111), "example", "", "");
    ```
19. ```
        int batchSize = 1000;
    ```
20. ```
        int emptyCount = 0;
    ```
21. ```
        try {
    ```
22. ```
            connector.connect();
    ```
23. ```
            connector.subscribe(".*\\..*");
    ```
24. ```
            connector.rollback();
    ```
25. ```
            int totalEmptyCount = 120;
    ```
26. ```
            while (emptyCount < totalEmptyCount) {
    ```
27. ```
                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
    ```
28. ```
                long batchId = message.getId();
    ```
29. ```
                int size = message.getEntries().size();
    ```
30. ```
                if (batchId == -1 || size == 0) {
    ```
31. ```
                    emptyCount++;
    ```
32. ```
                    System.out.println("empty count : " + emptyCount);
    ```
33. ```
                    try {
    ```
34. ```
                        Thread.sleep(1000);
    ```
35. ```
                    } catch (InterruptedException e) {
    ```
36. ```
                    }
    ```
37. ```
                } else {
    ```
38. ```
                    emptyCount = 0;
    ```
39. ```
                    // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
    ```
40. ```
                    printEntry(message.getEntries());
    ```
41. ```
                }
    ```

43. ```
                connector.ack(batchId); // 提交确认
    ```
44. ```
                // connector.rollback(batchId); // 处理失败, 回滚数据
    ```
45. ```
            }
    ```

47. ```
            System.out.println("empty too many times, exit");
    ```
48. ```
        } finally {
    ```
49. ```
            connector.disconnect();
    ```
50. ```
        }
    ```
51. ```
    }
    ```

53. ```
    private static void printEntry(List&lt;Entry> entrys) {
    ```
54. ```
        for (Entry entry : entrys) {
    ```
55. ```
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
    ```
56. ```
                continue;
    ```
57. ```
            }
    ```

59. ```
            RowChange rowChage = null;
    ```
60. ```
            try {
    ```
61. ```
                rowChage = RowChange.parseFrom(entry.getStoreValue());
    ```
62. ```
            } catch (Exception e) {
    ```
63. ```
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
    ```
64. ```
                                           e);
    ```
65. ```
            }
    ```

67. ```
            EventType eventType = rowChage.getEventType();
    ```
68. ```
            System.out.println(String.format("================&gt; binlog[%s:%s] , name[%s,%s] , eventType : %s",
    ```
69. ```
                                             entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
    ```
70. ```
                                             entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
    ```
71. ```
                                             eventType));
    ```

73. ```
            for (RowData rowData : rowChage.getRowDatasList()) {
    ```
74. ```
                if (eventType == EventType.DELETE) {
    ```
75. ```
                    printColumn(rowData.getBeforeColumnsList());
    ```
76. ```
                } else if (eventType == EventType.INSERT) {
    ```
77. ```
                    printColumn(rowData.getAfterColumnsList());
    ```
78. ```
                } else {
    ```
79. ```
                    System.out.println("-------&gt; before");
    ```
80. ```
                    printColumn(rowData.getBeforeColumnsList());
    ```
81. ```
                    System.out.println("-------&gt; after");
    ```
82. ```
                    printColumn(rowData.getAfterColumnsList());
    ```
83. ```
                }
    ```
84. ```
            }
    ```
85. ```
        }
    ```
86. ```
    }
    ```

88. ```
    private static void printColumn(List&lt;Column> columns) {
    ```
89. ```
        for (Column column : columns) {
    ```
90. ```
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
    ```
91. ```
        }
    ```
92. ```
    }
    ```
93. }
  1. 运行Client

首先启动Canal Server，可参见[[QuickStart]]

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

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

此时代表当前数据库无变更数据

  1. 触发数据库变更
1. mysql> use test;
2. Database changed
3. mysql> CREATE TABLE xdual (
4. ->   ID int(11) NOT NULL AUTO_INCREMENT,
5. ->   X timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
6. ->   PRIMARY KEY (ID)
7. -> ) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8 ;
8. Query OK, 0 rows affected (0.06 sec)
9. mysql> insert into xdual(id,x) values(null,now());Query OK, 1 row affected (0.06 sec)

可以从控制台中看到：

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工程，谢谢.