Canal实时监控案例

Canal实时监控案例

---

0. 写在前面

• Canal版本：Canal-1.1.5
• Kafka版本：Kafka-2.4.1
• Zookeeper版本：Zookeeper-3.5.7

解压安装canal的tar.gz包之前，提前创建一个目录canal-x.x.x作为canal的安装目录，因为canal解压后是分散的

1. TCP 模式测试

1.1 IDEA创建项目canal-module

编辑pom.xml文件：添加以下依赖

<dependencies>
    <dependency>
        <groupId>com.alibaba.otter</groupId>
        <artifactId>canal.client</artifactId>
        <version>1.1.2</version>
    </dependency>

    <dependency>
        <groupId>com.alibaba.otter</groupId>
        <artifactId>canal.protocol</artifactId>
        <version>1.1.5</version>
    </dependency>

    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>2.4.1</version>
    </dependency>
</dependencies>

V1.1.5版本需要多添加canal.protocol这个依赖，如果是V1.1.2就不需要

1.2 通用监视类——CanalClient

1.2.1 Canal 封装的数据结构

Message：一次canal从日志中抓取的信息,一个message可以包含多个sql执行的结果

1.2.2 在 canal-module 模块下创建 cn.canal 包，并在该包下创建 CanalClient.java文件

代码如下：

import com.alibaba.fastjson.JSONObject;
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 com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;

import java.net.InetSocketAddress;
import java.util.List;


public class CanalClient {

    public static void main(String[] args) throws InvalidProtocolBufferException, InterruptedException {
        // TODO 获取连接
        CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress("node01", 11111),"example", "", "");

        while (true) {
            // TODO 连接
            canalConnector.connect();
            // TODO 订阅数据库test_canal
            canalConnector.subscribe("test_canal.*");
            // TODO 获取指定数量的数据
            Message message = canalConnector.get(100);
            // TODO 获取Entry集合
            List<CanalEntry.Entry> entries = message.getEntries();

            //TODO 判断集合是否为空,如果为空,则等待一会继续拉取数据
            if (entries.size() <= 0) {
                System.out.println("当次抓取没有数据，休息一会----------------");
                Thread.sleep(1000);
            } else {
                // TODO 遍历entries，单条解析
                for (CanalEntry.Entry entry : entries) {
                    // 1.获取表名
                    String tableName = entry.getHeader().getTableName();
                    // 2.获取类型
                    CanalEntry.EntryType entryType = entry.getEntryType();
                    / /3.获取序列化后的数据
                    ByteString storeValue = entry.getStoreValue();

                    //4.判断当前entryType类型是否为ROWDATA
                    if (CanalEntry.EntryType.ROWDATA.equals(entryType)) {
                        // 5.反序列化数据
                        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);
                        // 6.获取当前事件的操作类型
                        CanalEntry.EventType eventType = rowChange.getEventType();
                        // 7.获取数据集
                        List<CanalEntry.RowData> rowDataList = rowChange.getRowDatasList();

                        // 8.遍历rowDataList，并打印数据集
                        for (CanalEntry.RowData rowData : rowDataList) {
                            // 之前的数据
                            JSONObject beforeData = new JSONObject();
                            List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
                            for (CanalEntry.Column column : beforeColumnsList) {
                                beforeData.put(column.getName(), column.getValue());
                            }
                            // 之后的数据
                            JSONObject afterData = new JSONObject();
                            List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                            for (CanalEntry.Column column : afterColumnsList) {
                                afterData.put(column.getName(), column.getValue());
                            }
                            // 数据打印（控制台|Kafka）
                            System.out.println("Table:" + tableName +
                                    ",EventType:" + eventType +
                                    ",Before:" + beforeData +
                                    ",After:" + afterData);
                        }
                    } else {
                        System.out.println("当前操作类型为：" + entryType);
                    }
                }
            }
        }
    }
}

开启canal，运行CanalClient查程序，对订阅的数据库canal_test下的表进行增删改操作，同时观察控制台的输出情况

• 增加数据

单词插入一条数据

insert into user_info values('1001', 'zss', 'male');

一条sql影响多行

insert into user_info values('1002', 'lisi', 'female'),('1001', 'zss', 'male');

• 修改数据

• 删除数据

2. Kafka 模式测试

• 修改 canal.properties 中 canal 的输出 model，默认 tcp，改为输出到kafka

# tcp, kafka, rocketMQ, rabbitMQ
canal.serverMode = kafka

• 修改 Kafka 集群的地址

##################################################
#########                    Kafka                   #############
##################################################
kafka.bootstrap.servers = node01:9092,node02:9092,node03:9092

• 修改 instance.properties 输出到 Kafka 的主题(canal_test)以及分区数

# mq config
canal.mq.topic=canal_test
# dynamic topic route by schema or table regex
#canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..*
canal.mq.partition=0
# hash partition config
#canal.mq.partitionsNum=3
#canal.mq.partitionHash=test.table:id^name,.*\\..*
#canal.mq.dynamicTopicPartitionNum=test.*:4,mycanal:6

注意：默认还是输出到指定 Kafka 主题的一个 kafka 分区，因为多个分区并行可能会打乱 binlog 的顺序， 如果要提高并行度， 首先设置 kafka 的分区数>1, 然后设置 canal.mq.partitionHash 属性

• 启动Canal

[zhangsan@node01 example]$ cd /opt/module/canal/ 
[zhangsan@node01 example]$  bin/startup.sh

• 看到 CanalLauncher 你表示启动成功，同时会创建 canal_test 主题

[zhangsan@node01 example]$ jps 
2269 Jps
2253 CanalLauncher

• 启动 Kafka 消费客户端测试，查看消费情况

	[zhangsan@node01 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server node01:9092 --topic canal_test

• 向 MySQL 中插入|修改|删除数据后查看消费者控制台

Kafka 消费者控制台

• 增加数据

单词插入一条数据

insert into user_info values('1001', 'zss', 'male');

一条sql影响多行

insert into user_info values('1002', 'lisi', 'female'),('1001', 'zss', 'male');
update user_info 

• 修改数据

• 删除数据

结束！