一 架构设计
简单架构设计
说明：
- 两个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
四 高可用架构设计
高可用架构设计
