Canal介绍

      基于mysql数据库binlog的增量订阅&消费

ClientExample

依赖配置：(目前暂未正式发布到mvn仓库，所以需要各位下载canal源码后手工执行下mvn clean install -Dmaven.test.skip)

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

1. 创建mvn标准工程：

1.mvn archetype:create -DgroupId=com.alibaba.otter -DartifactId=canal.sample  

2.  修改pom.xml，添加依赖

3.  ClientSample代码

1.package com.alibaba.otter.canal.sample;
2.
3.import java.net.InetSocketAddress;
4.import java.util.List;
5.
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.
15.public class SimpleCanalClientExample {
16.
17.    public static void main(String args[]) {
18.        // 创建链接
19.        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),
20.                                                                                            11111), "example", "", "");
21.        int batchSize = 1000;
22.        int emptyCount = 0;
23.        try {
24.            connector.connect();
25.            connector.subscribe(".*\\..*");
26.            connector.rollback();
27.            int totalEmtryCount = 120;
28.            while (emptyCount < totalEmtryCount) {
29.                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
30.                long batchId = message.getId();
31.                int size = message.getEntries().size();
32.                if (batchId == -1 || size == 0) {
33.                    emptyCount++;
34.                    System.out.println("empty count : " + emptyCount);
35.                    try {
36.                        Thread.sleep(1000);
37.                    } catch (InterruptedException e) {
38.                    }
39.                } else {
40.                    emptyCount = 0;
41.                    // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
42.                    printEntry(message.getEntries());
43.                }
44.
45.                connector.ack(batchId); // 提交确认
46.                // connector.rollback(batchId); // 处理失败, 回滚数据
47.            }
48.
49.            System.out.println("empty too many times, exit");
50.        } finally {
51.            connector.disconnect();
52.        }
53.    }
54.
55.    private static void printEntry(List<Entry> entrys) {
56.        for (Entry entry : entrys) {
57.            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
58.                continue;
59.            }
60.
61.            RowChange rowChage = null;
62.            try {
63.                rowChage = RowChange.parseFrom(entry.getStoreValue());
64.            } catch (Exception e) {
65.                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
66.                                           e);
67.            }
68.
69.            EventType eventType = rowChage.getEventType();
70.            System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
71.                                             entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
72.                                             entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
73.                                             eventType));
74.
75.            for (RowData rowData : rowChage.getRowDatasList()) {
76.                if (eventType == EventType.DELETE) {
77.                    printColumn(rowData.getBeforeColumnsList());
78.                } else if (eventType == EventType.INSERT) {
79.                    printColumn(rowData.getAfterColumnsList());
80.                } else {
81.                    System.out.println("-------> before");
82.                    printColumn(rowData.getBeforeColumnsList());
83.                    System.out.println("-------> after");
84.                    printColumn(rowData.getAfterColumnsList());
85.                }
86.            }
87.        }
88.    }
89.
90.    private static void printColumn(List<Column> columns) {
91.        for (Column column : columns) {
92.            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
93.        }
94.    }
95.}

4. 运行Client

首先启动Canal Server，可参加QuickStart : http://agapple.iteye.com/blogs/1796070

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

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

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

5.  触发数据库变更

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.
10.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

最后：

  整个代码在附件中可以下载，如有问题可及时联系。