处理分表分库的场景，主要是要使用配置 group-instance.xml 。group-instance主要针对需要进行多库合并时，可以将多个物理instance合并为一个逻辑instance，提供客户端访问。

比如产品数据拆分了4个库，每个库会有一个instance，如果不用group，业务上要消费数据时，需要启动4个客户端，分别链接4个instance实例。使用group后，可以在canal server上合并为一个逻辑instance，只需要启动1个客户端，链接这个逻辑instance即可。

二 关键实现流程

2.1 canal.properties配置文件

canal.properties是对应一个canal server的全局配置，保存位置： /usr/local/hadoop/app/canal_group/conf/canal.properties 。配置修改内容如下：

canal.id = 1 #唯一标识
canal.ip =192.168.175.20 # client访问canal server的ip地址
canal.port = 11111  # client访问canal server的端口
#canal.instance.global.spring.xml = classpath:spring/file-instance.xml   #原来是这个
canal.instance.global.spring.xml = classpath:spring/group-instance.xml  #启动这个
#其他配置保持默认即可.
2.2 instance.properties配置文件
使用如下命令复制出两个代表canal instance的文件夹:
cp -R example t_dept;
cp -R example t_canal;
rm -rf example;
调整配置文件/usr/local/hadoop/app/canal/conf/t_dept/instance.properties如下:
#canal.instance.master.address=192.168.175.21:3306 #原来的
canal.instance.master1.address=192.168.175.21:3306 #新增,与group-instance.xml的对应
canal.instance.master2.address=192.168.175.22:3306 #新增,与group-instance.xml的对应
# username/password
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
# mq config
canal.mq.topic=t_dept
调整配置文件/usr/local/hadoop/app/canal/conf/t_canal/instance.properties如下:
#canal.instance.master.address=192.168.175.21:3306 #原来的
canal.instance.master1.address=192.168.175.21:3306 #新增,与group-instance.xml的对应
# username/password
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
# mq config
canal.mq.topic=t_canal
2.3 group-instance.xml配置文件
配置文件/usr/local/hadoop/app/canal/conf/spring/group-instance.xml不需要做调整。
2.4 启动canal server
进入文件夹/usr/local/hadoop/app/canal/bin执行如下启动命令:
./startup.sh
查看日志/usr/local/hadoop/app/canal/logs/canal/canal.log,出现如下内容,即表示启动成功:
2019-06-07 21:15:03.372 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## load canal configurations
2019-06-07 21:15:03.427 [main] INFO  c.a.o.c.d.monitor.remote.RemoteConfigLoaderFactory - ## load local canal configurations
2019-06-07 21:15:03.529 [main] INFO  com.alibaba.otter.canal.deployer.CanalStater - ## start the canal server.
2019-06-07 21:15:06.251 [main] INFO  com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[192.168.175.20:11111]
2019-06-07 21:15:22.245 [main] INFO  com.alibaba.otter.canal.deployer.CanalStater - ## the canal server is running now ......
2.5 使用canal client连接canal server
注意运行canal客户端代码时,一定要先启动canal server!!!
(1) 添加pom依赖
    <!--canal-->
    <dependency>
      <groupId>com.alibaba.otter</groupId>
      <artifactId>canal.client</artifactId>
      <version>1.1.3</version>
    </dependency>
(2) canal client代码:
package com.xgh.canal;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;
import java.net.InetSocketAddress;
import java.util.List;
public class CanalClientGroupTest1 {
    public static void main(String args[]) {
        //String zkHost="192.168.175.20:2181,192.168.175.21:2181,192.168.175.22:2181";
        // 创建链接
        //CanalConnector connector = CanalConnectors.newClusterConnector(zkHost,"example","","");
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.175.21", 11111),
                "t_dept", "", "");
        int batchSize = 1000;
        int emptyCount = 0;
        long batchId = 0;
        //外层死循环:在canal节点宕机后,抛出异常,等待zk对canal处理切换,切换完后,继续创建连接处理数据
        while(true) {
            try {
                connector.connect();
                connector.subscribe(".*\\..*");//订阅所有库下面的所有表
                //connector.subscribe("canal.t_canal");//订阅库canal库下的表t_canal
                connector.rollback();
                //内层死循环:按频率实时监听数据变化,一旦收到变化数据,立即做消费处理,并ack,考虑消费速度,可以做异步处理并ack.
                while (true) {
                    Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                    batchId = message.getId();
                    int size = message.getEntries().size();
                    //// 偏移量不等于-1 或者 获取的数据条数不为0 时，认为拿到消息，并处理
                    if (batchId == -1 || size == 0) {
                        emptyCount++;
                        System.out.println("empty count : " + emptyCount);//此時代表當前數據庫無遍更數據
                        Thread.sleep(1000); //1000ms拉一次变动数据
                    } else {
                        emptyCount = 0;
                        System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
                        printEntry(message.getEntries());
                    connector.ack(batchId); // 提交确认
            }catch(Exception e){
                e.printStackTrace();
                connector.rollback(batchId); // 处理失败, 回滚数据
            } finally {
                connector.disconnect();
    private static void printEntry(List<Entry> entrys) {
        for (Entry entry : entrys) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN
                    || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            RowChange rowChage = null;
            try {
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
            System.out.println("rowChare ======>"+rowChage.toString());
            EventType eventType = rowChage.getEventType(); //事件類型,比如insert,update,delete
            System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    entry.getHeader().getLogfileName(),//mysql的my.cnf配置中的log-bin名稱
                    entry.getHeader().getLogfileOffset(), //偏移量
                    entry.getHeader().getSchemaName(),//庫名
                    entry.getHeader().getTableName(), //表名
                    eventType));//事件名
            for (RowData rowData : rowChage.getRowDatasList()) {
                if (eventType == EventType.DELETE) {
                    printColumn(rowData.getBeforeColumnsList());
                } else if (eventType == EventType.INSERT) {
                    printColumn(rowData.getAfterColumnsList());
                } else {
                    System.out.println("-------> before");
                    printColumn(rowData.getBeforeColumnsList());
                    System.out.println("-------> after");
                    printColumn(rowData.getAfterColumnsList());
    private static void printColumn(List<Column> columns) {
        for (Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
2.6 其他
将canal client代码CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.175.21", 11111), "t_dept", "", "");中的队列名t_dept换成t_canal再执行,就可以监听t_canal对应数据变化了.
三 运行测试及总结
1. 监听t_dept的canal client可以接收到数据库A和B的数据变化
2. 监听t_canal的canal client只能接收到数据库B的数据变化
3. 数据过滤的设置问题
当在instance.properties和canal client中对设置filter时,canal client的设置会覆盖instance.properties中的配置。所以不如干脆保持instance.properties为默认状态,也即是不过滤，然后过滤全部设置在canal client中，如下：
connector.connect();
connector.subscribe(".*\\..*");//订阅所有库下面的所有表
//connector.subscribe("canal.t_canal");//订阅库canal库下的表t_canal
四 高可用架构设计