一 架构设计
网络异常，图片无法展示
|
简单架构设计
说明：
- 两个mysql库中均创建有canal/canal的账户;
- 这里A、B两个mysql库是用来模拟t_dept进行分库分表;
- 另外，在A、B两种表中都创建有表t_canal.
- canal原理: 可查看文章【了解canal，看这个就够了】
- 安装与搭建流程:可参考文章【canal应用-1个server+2个instance+2个client+2个mysql】
处理分表分库的场景，主要是要使用配置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(), e); } 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
四 高可用架构设计
网络异常，图片无法展示
|
高可用架构设计
