Canal实时监控案例
Canal实时监控案例
文章目录
- Canal实时监控案例
- 0. 写在前面
- 1. TCP 模式测试
- 1.1 IDEA创建项目canal-module
- 1.2 通用监视类——CanalClient
- 1.2.1 Canal 封装的数据结构
- 1.2.2 在 canal-module 模块下创建 cn.canal 包,并在该包下创建 CanalClient.java文件
- 2. Kafka 模式测试
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_canalcanalConnector.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类型是否为ROWDATAif (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
- 修改数据
- 删除数据
结束!
Canal实时监控案例相关推荐
- Canal实时监控mysql数据库
1. canal原理概述 1.1 mysql的主从复制原理1.1.1 mysql master将数据变更写入二进制日志(binlog,其中记录叫二进制日志事件,可通过show binlog event ...
- 看大众点评如何通过实时监控系统CAT打造7*24服务
为什么80%的码农都做不了架构师?>>> 看大众点评如何通过实时监控系统CAT打造7*24服务 2015-06-08 尤勇 高可用架构 https://github.com ...
- zabbix安装使用及监控案例(从头到尾)
前文提要:本文为整个zabbix的介绍及配置以及案例 目录 一.监控的概述 二.zabbix的概述 1.zabbix是什么 2.zabbix的作用 3.zabbix的特性 4.zabbix系统的组成 ...
- 实时监控:基于流计算 Oceanus ( Flink ) 实现系统和应用级实时监控
作者:吴云涛,腾讯 CSIG 高级工程师 本文描述了如何使用腾讯云大数据组件来完成实时监控系统的设计和实现,通过实时采集并分析云服务器(CVM)及其 App 应用的 CPU和内存等资源消耗数据,以短信 ...
- 服务器部署 配置jetty运行参数_Zookeeper+websocket实现对分布式服务器的实时监控...
Zookeeper简介 Zookeeper是Hadoop的一个子项目,它是分布式系统中的协调系统. 简单来说就是一个Zookeeper注册同步中心,内部结构为一个树形目录,每个节点上可以存放一定量(默 ...
- rsync+inotify实现实时同步案例详解
rsync+inotify实现实时同步案例详解 转自:http://chocolee.blog.51cto.com/8158455/1400596 随着应用系统规模的不断扩大,对数据的安全性和可靠性也 ...
- 流媒体服务器+终端(android,ios,web)来实现实时监控或着说是直播
简介说明: 流媒体服务器+终端(android,ios,web)来实现实时监控或着说是直播 以下几个因素建议考虑 采集.预处理.编码.传输.服务器转码.解码 采集采集是播放环节中的第一环,iOS 系统 ...
- 推荐一款清爽的实时监控大屏附安装教程
点击▲关注 "爪哇妹" 给公众号标星置顶 更多摄影技巧 第一时间直达 前言 前段时间分享了一个支付监控大屏,好多小伙伴在问撸主是怎么实现的,所以花几分钟时间跟大家分享一下. 预览 ...
- HTML+CSS+JavaScript+Ajax+ECharts实现疫情实时监控大屏-2设计与实现
疫情期间,各类应用系统.大数据系统为防疫抗疫带来了巨大助力.国家博物馆首次将代码列为藏品,那就是抗击疫情期间来自阿里程序员写的健康码第一行代码以及研发人员的签名.这展现了广大程序员在疫情面前,争当勇敢 ...
最新文章
- ajax bootstrap 进度条,Bootstrap进度条与AJAX后端数据传递结合使用实例详解
- python opencv 如何检测模糊图像?自动判断模糊程度 cv2.Laplacian().var()
- 技术人员写blog的好处
- Scalaz(44)- concurrency :scalaz Future,尚不完整的多线程类型
- 经典C语言程序100例之六三
- 使用.Net Core编写命令行工具(CLI)
- MVP公益活动:编程一小时
- 局部变量与全局变量同名时如何在局部变量的作用范围内访问全局变量?
- Microsoft ASP.NET 4 Step by Step
- Android原生PDF功能实现,掌握了这些Android高级工程师必备知识,
- logback配置详解
- JavaScript(JS)--基础2(基本语法、字面量和变量、标识符、强制类型转换)
- 金融衍生物英语名词的对应中文解析
- 测试9年,面试华为要薪1万,华为员工:公司没这么低工资的岗
- webrtc音频QOS方法四(音频接收端NACK流程实现)
- PAT A1143 Lowest Common Ancestor ——沉舟侧畔千帆过,病树前头万木春
- 项管行知04--项目章程
- Windows10更新后耳机没声音
- 幼儿园教师计算机运用论文,幼儿园信息化论文范文
- 阿里云短信验证服务详细基础教程