调研背景
前段时间在调研实时计算的时候就在思考一个问题，目前的流计算是基于一个数据流的概念，而我们目前报名系统数据的同步是由集团将数据直接更新到数据库的，如果我们后期要集成这部分数据做实时报表分析，那么怎么监听这一部分数据呢，经调研借助阿里巴巴的Canal开源项目，我们可以非常便捷地利用Mysql的binlog日志将Mysql中的数据动态地抽取到任意存储中。下面我就将Canal的使用做一下简单的介绍。
Canal介绍
我们之前做过很多次数据库的主从配置，简单来说，Canal 就是将自己伪装成 MySQL 从节点（Slave），并从主节点（Master）获取 Binlog，解析和存储后供下游消费端使用。Canal 包含两个组成部分：服务端和客户端。服务端负责连接至不同的 MySQL 实例，并为每个实例维护一个事件消息队列；客户端则可以订阅这些队列中的数据变更事件，处理并存储到数据仓库中。下面我们来看如何快速搭建起一个 Canal 服务。
使用说明
配置 MySQL 主节点
MySQL 默认没有开启 Binlog，因此我们需要对 my.cnf 文件做以下修改：
server-id = 1
log_bin = /usr/local/mysql/mysql-bin.log
binlog_format = ROW
注意 binlog_format 必须设置为 ROW, 因为在 STATEMENT 或 MIXED 模式下, Binlog 只会记录和传输 SQL 语句（以减少日志大小），而不包含具体数据，我们也就无法保存了。
从节点通过一个专门的账号连接主节点，这个账号需要拥有全局的 REPLICATION 权限。我们可以使用 GRANT 命令创建这样的账号：
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT
ON *.* TO 'canal'@'%' IDENTIFIED BY 'canal'
Canal 服务端
下载 Canal 安装后，能看到以下配置文件
canal.deployer/conf/canal.properties
canal.deployer/conf/instanceA/instance.properties
canal.deployer/conf/instanceB/instance.properties
conf/canal.properties 是主配置文件，如其中的 canal.port 用以指定服务端监听的端口。instanceA/instance.properties 则是各个实例的配置文件，主要的配置项有：
# slaveId 不能与 my.cnf 中的 server-id 项重复
canal.instance.mysql.slaveId = 1234
canal.instance.master.address = 127.0.0.1:3306
canal.instance.dbUsername = canal
canal.instance.dbPassword = canal
canal.instance.connectionCharset = UTF-8
# 订阅实例中所有的数据库和表
canal.instance.filter.regex = .*\\..*
执行 sh bin/startup.sh 命令开启服务端
编写 Canal 客户端
从服务端消费变更消息时，我们需要创建一个 Canal 客户端，指定需要订阅的数据库和表，并开启轮询。
首先，在项目中添加 com.alibaba.otter:canal.client 依赖项，构建 CanalConnector 实例：
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("IP", 端口号), "example", "", "");
connector.connect();
connector.subscribe(".*\\..*");//需要监听的库、表
while (true) {
Message message = connector.getWithoutAck(100);
long batchId = message.getId();
if (batchId == -1 || message.getEntries().isEmpty()) {
Thread.sleep(3000);
} else {
printEntries(message.getEntries());
connector.ack(batchId);
}
}
这段代码和连接消息系统很相似。变更事件会批量发送过来，待处理完毕后我们可以 ACK 这一批次，从而避免消息丢失。
// printEntries
RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
for (RowData rowData : rowChange.getRowDatasList()) {
if (rowChange.getEventType() == EventType.INSERT) {
printColumns(rowData.getAfterCollumnList());
}
}
每一个 Entry 代表一组具有相同变更类型的数据列表，如 INSERT 类型、UPDATE、DELETE 等。每一行数据我们都可以获取到各个字段的信息：
// printColumns
String line = columns.stream()
.map(column -> column.getName() + "=" + column.getValue())
.collect(Collectors.joining(","));
System.out.println(line);
延申思考
由于我们做报表需要统计所有数据，那么如果采用以上方案，那么势必存在一部分动态数据、一部分固定的数据（监听前），目前已与集团cdc团队沟通过关闭数据库同步，我们先做好原始数据的统计，待统计完成后，再开启数据库同步，然后结合我们 的实时计算进行可以完成我们的需求。经过这几天的思考个人感觉其实也可以采取另一种方式，比如我们要同步user表，那么我们在不关闭数据同步的情况下怎么处理呢，可以新建一张表user_new然后将user表数据全部插入user_new，这样我们就可以将所有数据通过canal转移到消息队列后输入到数据分析平台中，理论上可行，当然这还需要测试。
