SpringBoot+Canal을 사용하여 실시간 데이터베이스 모니터링을 구현하는 방법
Canal 工作原理
Canal 模拟 MySQL slave 的交互协议,伪装自己为 MySQL slave,向 MySQL master 发送 dump 协议
MySQL master 收到 dump 请求,开始推送 binary log 给 slave(也就是 Canal)
Canal 解析 binary log 对象(原始为 byte 流)
MySQL 打开 binlog 模式
在 MySQL 配置文件 my.cnf 设置如下信息:
[mysqld] # 打开binlog log-bin=mysql-bin # 选择ROW(行)模式 binlog-format=ROW # 配置MySQL replaction需要定义,不要和canal的slaveId重复 server_id=1
改了配置文件之后,重启 MySQL,使用命令查看是否打开 binlog 模式:
SHOW VARIABLES LIKE 'log_bin';
查看 binlog 日志文件列表:SHOW BINARY LOGS;
查看当前正在写入的 binlog 文件:SHOW MASTER STATUS;
记录文件名 File 和 Position 值。
启动 Canal 服务
详细配置可以参考:
conf\example\instance.properties ################################################# ## mysql serverId , v1.0.26+ will autoGen # canal.instance.mysql.slaveId=0 //每个instance都会伪装成一个mysql slave , 此id对于canal前端的Mysql实例而言,必须是唯一的,但是同一个Canal中相同的instance,此slaveld应该一样 # enable gtid use true/false canal.instance.gtidon=false # position info canal.instance.master.address=127.0.0.1:3306 //需要连接的数据库地址及端口 canal.instance.master.journal.name= //需要读取的起始的binlog文件 canal.instance.master.position= //需要读取的起始的binlog文件的偏移量 canal.instance.master.timestamp= //需要读取的起始的binlog的时间戳 canal.instance.master.gtid= # rds oss binlog canal.instance.rds.accesskey= canal.instance.rds.secretkey= canal.instance.rds.instanceId= # table meta tsdb info canal.instance.tsdb.enable=true //v1.0.25版本新增,是否开启table meta的时间序列版本记录功能 #canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb //v1.0.25版本新增,table meta的时间序列版本的本地存储路径,默认为instance目录 #canal.instance.tsdb.dbUsername=canal #canal.instance.tsdb.dbPassword=canal #canal.instance.standby.address = #canal.instance.standby.journal.name = #canal.instance.standby.position = #canal.instance.standby.timestamp = #canal.instance.standby.gtid= # username/password canal.instance.dbUsername=canal //数据库账号 canal.instance.dbPassword=canal //数据库密码 canal.instance.connectionCharset = UTF-8 //数据库解析编码格式 canal.instance.defaultDatabaseName =test //数据库连接时默认schema # enable druid Decrypt database password canal.instance.enableDruid=false #canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ== # table regex canal.instance.filter.regex=.*\\..* //mysql 数据解析关注的表,Perl正则表达式. # table black regex canal.instance.filter.black.regex= //canal将会过滤那些不符合要求的table,这些table的数据将不会被解析和传送 #################################################
conf\canal.properties ################################################# ######### common argument ############# ################################################# canal.id= 1 #每个canal server实例的唯一标识 canal.ip= #canal server绑定的本地IP信息,如果不配置,默认选择一个本机IP进行, canal.port=11111 #canal server提供socket tcp服务的端口 canal.metrics.pull.port=11112 canal.zkServers= #canal server链接zookeeper集群的链接信息 # flush data to zk canal.zookeeper.flush.period = 1000 #canal持久化数据到zookeeper上的更新频率,单位毫秒 canal.withoutNetty = false # tcp, kafka, RocketMQ canal.serverMode = tcp # flush meta cursor/parse position to file canal.file.data.dir = ${canal.conf.dir} #canal持久化数据到file上的目录 canal.file.flush.period = 1000 #canal持久化数据到file上的更新频率,单位毫秒 ## memory store RingBuffer size, should be Math.pow(2,n) canal.instance.memory.buffer.size = 16384 #canal内存store中可缓存buffer记录数,需要为2的指数 ## memory store RingBuffer used memory unit size , default 1kb canal.instance.memory.buffer.memunit = 1024 # 内存记录的单位大小,默认1KB,和buffer.size组合决定最终的内存使用大小 ## meory store gets mode used MEMSIZE or ITEMSIZE canal.instance.memory.batch.mode = MEMSIZE #canal内存store中数据缓存模式 1. ITEMSIZE : 根据buffer.size进行限制,只限制记录的数量 2. MEMSIZE : 根据buffer.size * buffer.memunit的大小,限制缓存记录的大小 canal.instance.memory.rawEntry = true ## detecing config canal.instance.detecting.enable = false #是否开启心跳检查 #canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now() canal.instance.detecting.sql = select 1 #心跳检查sql canal.instance.detecting.interval.time = 3 #心跳检查频率,单位秒 canal.instance.detecting.retry.threshold = 3 #心跳检查失败重试次数 ##非常注意:interval.time * retry.threshold值,应该参考既往DBA同学对数据库的故障恢复时间, ##“太短”会导致集群运行态角色“多跳”;“太长”失去了活性检测的意义,导致集群的敏感度降低,Consumer断路可能性增加。 canal.instance.detecting.heartbeatHaEnable = false #心跳检查失败后,是否开启自动mysql自动切换 #说明:比如心跳检查失败超过阀值后,如果该配置为true,canal就会自动链到mysql备库获取binlog数据 false # support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery canal.instance.transaction.size = 1024 # 最大事务完整解析的长度支持超过该长度后,一个事务可能会被拆分成多次提交到canal store中,无法保证事务的完整可见性 # mysql fallback connected to new master should fallback times canal.instance.fallbackIntervalInSeconds = 60 #canal发生mysql切换时,在新的mysql库上查找 binlog时需要往前查找的时间,单位秒 说明:mysql主备库可能存在解析延迟或者时钟不统一,需要回退一段时间,保证数据不丢 # network config canal.instance.network.receiveBufferSize = 16384 #网络链接参数,SocketOptions.SO_RCVBUF canal.instance.network.sendBufferSize = 16384 #网络链接参数,SocketOptions.SO_SNDBUF canal.instance.network.soTimeout = 30 #网络链接参数,SocketOptions.SO_TIMEOUT # binlog filter config canal.instance.filter.druid.ddl = true canal.instance.filter.query.dcl = false #ddl语句是否隔离发送,开启隔离可保证每次只返回发送一条ddl数据,不和其他dml语句混合返回.(otter ddl同步使用) canal.instance.filter.query.dml = false #是否忽略DML的query语句,比如insert/update/delete table.(mysql5.6的ROW模式可以包含statement模式的query记录) canal.instance.filter.query.ddl = false #是否忽略DDL的query语句,比如create table/alater table/drop table/rename table/create index/drop index. (目前支持的ddl类型主要为table级别的操作,create databases/trigger/procedure暂时划分为dcl类型) canal.instance.filter.table.error = false canal.instance.filter.rows = false canal.instance.filter.transaction.entry = false # binlog format/image check canal.instance.binlog.format = ROW,STATEMENT,MIXED canal.instance.binlog.image = FULL,MINIMAL,NOBLOB # binlog ddl isolation canal.instance.get.ddl.isolation = false # parallel parser config canal.instance.parser.parallel = true ## concurrent thread number, default 60% available processors, suggest not to exceed Runtime.getRuntime().availableProcessors() #canal.instance.parser.parallelThreadSize = 16 ## disruptor ringbuffer size, must be power of 2 canal.instance.parser.parallelBufferSize = 256 # table meta tsdb info //关于时间序列版本 canal.instance.tsdb.enable=true canal.instance.tsdb.dir=${canal.file.data.dir:../conf}/${canal.instance.destination:} canal.instance.tsdb.url=jdbc:h3:${canal.instance.tsdb.dir}/h3;CACHE_SIZE=1000;MODE=MYSQL; canal.instance.tsdb.dbUsername=canal canal.instance.tsdb.dbPassword=canal # dump snapshot interval, default 24 hour canal.instance.tsdb.snapshot.interval=24 # purge snapshot expire , default 360 hour(15 days) canal.instance.tsdb.snapshot.expire=360 # rds oss binlog account canal.instance.rds.accesskey = canal.instance.rds.secretkey = ################################################# ######### destinations ############# ################################################# canal.destinations= example # conf root dir canal.conf.dir = ../conf # auto scan instance dir add/remove and start/stop instance canal.auto.scan = true #开启instance自动扫描 如果配置为true,canal.conf.dir目录下的instance配置变化会自动触发: a. instance目录新增: 触发instance配置载入,lazy为true时则自动启动 b. instance目录删除:卸载对应instance配置,如已启动则进行关闭 c. instance.properties文件变化:reload instance配置,如已启动自动进行重启操作 canal.auto.scan.interval = 5 #instance自动扫描的间隔时间,单位秒 canal.instance.tsdb.spring.xml=classpath:spring/tsdb/h3-tsdb.xml #canal.instance.tsdb.spring.xml=classpath:spring/tsdb/mysql-tsdb.xml canal.instance.global.mode = spring #instance管理模式,Production级别我们要求使用spring canal.instance.global.lazy = false #全局lazy模式 #canal.instance.global.manager.address = 127.0.0.1:1099 #全局的manager配置方式的链接信息 #canal.instance.global.spring.xml = classpath:spring/memory-instance.xml canal.instance.global.spring.xml = classpath:spring/file-instance.xml #全局的spring配置方式的组件文件 #canal.instance.global.spring.xml = classpath:spring/default-instance.xml
1)canal.deployer-1.1.5\conf\canal.properties
进行全局配置,可以修改 IP、端口号或实例
# 默认有一个 example,需要增加实例的可以配置 canal.destinations = example
2)canal.deployer-1.1.5\conf\example\instance.properties
进行局部实例配置,可以修改数据库账号和密码、数据库表名、binlog 文件名和 position 等
# 没有改变的就没有贴出来,注意 MySQL 的用户名和密码 canal.instance.master.address=192.168.58.131:3306 # username/password canal.instance.dbUsername=test canal.instance.dbPassword=liubihao canal.instance.connectionCharset = UTF-8 # enable druid Decrypt database password canal.instance.enableDruid=false # table regex canal.instance.filter.regex=.*\\..* # table black regex canal.instance.filter.black.regex=
ysql 数据解析m关注的表,Perl正则表达式.
多个正则之间以逗号(,)分隔,转义符需要双斜杠(\\)
常见例子:
1. 所有表:.* or .*\\..*
2. canal schema下所有表: canal\\..*
3. canal下的以canal打头的表:canal\\.canal.*
4. canal schema下的一张表:canal.test1
5. 多个规则组合使用:canal\\..*,mysql.test1,mysql.test2 (逗号分隔)
bin/start.sh
启动服务(Windows 系统为bin/start.bat
)
后端进行相应配置
修改 pom.xml
<dependency> <groupId>com.alibaba.otter</groupId> <artifactId>canal.client</artifactId> <version>1.1.4</version> </dependency>
新建 CanalClient.java
package org.nwpu.atcss.util; 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 org.springframework.stereotype.Component; import java.util.List; import java.net.InetSocketAddress; @Component public class CanalClient { private static void printEntries(List<Entry> entries) throws Exception { for (Entry entry : entries) { if (entry.getEntryType() != EntryType.ROWDATA) { continue; } RowChange rowChange = RowChange.parseFrom(entry.getStoreValue()); EventType eventType = rowChange.getEventType(); System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s", entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(), entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), eventType)); for (RowData rowData : rowChange.getRowDatasList()) { switch (rowChange.getEventType()) { case INSERT: System.out.println("INSERT "); printColumns(rowData.getAfterColumnsList()); break; case UPDATE: System.out.println("UPDATE "); printColumns(rowData.getAfterColumnsList()); break; case DELETE: System.out.println("DELETE "); printColumns(rowData.getBeforeColumnsList()); break; default: break; } } } } private static void printColumns(List<Column> columns) { for (Column column : columns) { System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated()); } } public static void main(String[] args) throws Exception { // hostname, port, destination, username, password CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1", 11111), "example", "", ""); try { connector.connect(); // 监听的表,格式为数据库.表名,数据库.表名 connector.subscribe(".*\\..*"); connector.rollback(); while (true) { Message message = connector.getWithoutAck(100); // 获取指定数量的数据 long batchId = message.getId(); if (batchId == -1 || message.getEntries().isEmpty()) { Thread.sleep(1000); continue; } // System.out.println(message.getEntries()); printEntries(message.getEntries()); connector.ack(batchId); // 提交确认,消费成功,通知server删除数据 // connector.rollback(batchId);// 处理失败, 回滚数据,后续重新获取数据 } } catch (Exception e) { System.out.println("Something Error."); } finally { connector.disconnect(); } } }
测试
启动 CanalClient.java
;
修改本地数据库内容之后,控制台成功监听并报告更新信息。
위 내용은 SpringBoot+Canal을 사용하여 실시간 데이터베이스 모니터링을 구현하는 방법의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!

핫 AI 도구

Undresser.AI Undress
사실적인 누드 사진을 만들기 위한 AI 기반 앱

AI Clothes Remover
사진에서 옷을 제거하는 온라인 AI 도구입니다.

Undress AI Tool
무료로 이미지를 벗다

Clothoff.io
AI 옷 제거제

AI Hentai Generator
AI Hentai를 무료로 생성하십시오.

인기 기사

뜨거운 도구

메모장++7.3.1
사용하기 쉬운 무료 코드 편집기

SublimeText3 중국어 버전
중국어 버전, 사용하기 매우 쉽습니다.

스튜디오 13.0.1 보내기
강력한 PHP 통합 개발 환경

드림위버 CS6
시각적 웹 개발 도구

SublimeText3 Mac 버전
신 수준의 코드 편집 소프트웨어(SublimeText3)

뜨거운 주제











Jasypt 소개 Jasypt는 개발자가 최소한의 노력으로 프로젝트에 기본 암호화 기능을 추가할 수 있게 해주며 암호화 작동 방식에 대한 깊은 이해가 필요하지 않은 단방향 및 양방향 암호화에 대한 높은 보안을 제공합니다. 표준 기반 암호화 기술. 비밀번호, 텍스트, 숫자, 바이너리 암호화... Spring 기반 애플리케이션, 개방형 API와의 통합에 적합하며 모든 JCE 공급자와 함께 사용할 수 있습니다... 다음 종속성을 추가합니다: com.github.ulisesbocchiojasypt-spring-boot-starter2. Jasypt의 이점은 코드가 유출되더라도 데이터 소스를 보장할 수 있어 시스템 보안을 보호합니다.

1. Redis는 분산 잠금 원칙과 분산 잠금이 필요한 이유를 구현합니다. 분산 잠금에 대해 이야기하기 전에 분산 잠금이 필요한 이유를 설명해야 합니다. 분산 잠금의 반대는 독립형 잠금입니다. 다중 스레드 프로그램을 작성할 때 공유 변수를 동시에 작동하여 발생하는 데이터 문제를 방지하기 위해 일반적으로 잠금을 사용하여 공유 변수를 상호 제외합니다. 공유 변수의 사용 범위는 동일한 프로세스에 있습니다. 동시에 공유 리소스를 운영해야 하는 여러 프로세스가 있는 경우 어떻게 상호 배타적일 수 있습니까? 오늘날의 비즈니스 애플리케이션은 일반적으로 마이크로서비스 아키텍처입니다. 이는 하나의 애플리케이션이 여러 프로세스를 배포한다는 의미이기도 합니다. 여러 프로세스가 MySQL에서 동일한 레코드 행을 수정해야 하는 경우 잘못된 작업으로 인해 발생하는 더티 데이터를 방지하려면 배포가 필요합니다. 현재 소개할 스타일은 잠겨 있습니다. 포인트를 얻고 싶다

사용 시나리오 1. 주문이 성공적으로 이루어졌으나 30분 이내에 결제가 이루어지지 않았습니다. 결제 시간이 초과되어 주문이 자동으로 취소되었습니다. 2. 주문이 서명되었으며 서명 후 7일 동안 평가가 수행되지 않았습니다. 주문 시간이 초과되어 평가되지 않으면 시스템은 기본적으로 긍정적 평가로 설정됩니다. 3. 판매자가 5분 동안 주문을 받지 않으면 주문이 취소됩니다. 문자 메시지 알림이 전송됩니다... 지연이 길고 실시간 성능이 낮은 시나리오의 경우 작업 예약을 사용하여 정기적인 폴링 처리를 수행할 수 있습니다. 예: xxl-job 오늘은 다음을 선택하겠습니다.

SpringBoot와 SpringMVC는 모두 Java 개발에서 일반적으로 사용되는 프레임워크이지만 둘 사이에는 몇 가지 분명한 차이점이 있습니다. 이 기사에서는 이 두 프레임워크의 기능과 용도를 살펴보고 차이점을 비교할 것입니다. 먼저 SpringBoot에 대해 알아봅시다. SpringBoot는 Spring 프레임워크를 기반으로 하는 애플리케이션의 생성 및 배포를 단순화하기 위해 Pivotal 팀에서 개발되었습니다. 독립 실행형 실행 파일을 구축하는 빠르고 가벼운 방법을 제공합니다.

Springboot가 파일을 읽지만 jar 패키지로 패키징한 후 최신 개발에 액세스할 수 없습니다. springboot가 파일을 jar 패키지로 패키징한 후 파일을 읽을 수 없는 상황이 발생합니다. 그 이유는 패키징 후 파일의 가상 경로 때문입니다. 유효하지 않으며 읽기를 통해서만 액세스할 수 있습니다. 파일은 리소스 publicvoidtest(){Listnames=newArrayList();InputStreamReaderread=null;try{ClassPathResourceresource=newClassPathResource("name.txt");Input 아래에 있습니다.

1. RedisAPI 기본 직렬화 메커니즘인 RedisTemplate1.1을 사용자 정의합니다. API 기반 Redis 캐시 구현은 데이터 캐싱 작업에 RedisTemplate 템플릿을 사용합니다. 여기서 RedisTemplate 클래스를 열고 클래스의 소스 코드 정보를 봅니다. 키 선언, 값의 다양한 직렬화 방법, 초기 값은 비어 있음 @NullableprivateRedisSe

Springboot+Mybatis-plus가 다중 테이블 추가 작업을 수행하기 위해 SQL 문을 사용하지 않을 때 내가 직면한 문제는 테스트 환경에서 생각을 시뮬레이션하여 분해됩니다. 매개 변수가 있는 BrandDTO 개체를 생성하여 배경으로 매개 변수 전달을 시뮬레이션합니다. Mybatis-plus에서 다중 테이블 작업을 수행하는 것은 매우 어렵다는 것을 Mybatis-plus-join과 같은 도구를 사용하지 않으면 해당 Mapper.xml 파일을 구성하고 냄새나고 긴 ResultMap만 구성하면 됩니다. 해당 SQL 문을 작성합니다. 이 방법은 번거로워 보이지만 매우 유연하며 다음을 수행할 수 있습니다.

프로젝트에서는 일부 구성 정보가 필요한 경우가 많습니다. 이 정보는 테스트 환경과 프로덕션 환경에서 구성이 다를 수 있으며 실제 비즈니스 상황에 따라 나중에 수정해야 할 수도 있습니다. 이러한 구성은 코드에 하드 코딩할 수 없습니다. 예를 들어 이 정보를 application.yml 파일에 작성할 수 있습니다. 그렇다면 코드에서 이 주소를 어떻게 얻거나 사용합니까? 2가지 방법이 있습니다. 방법 1: @Value 주석이 달린 ${key}를 통해 구성 파일(application.yml)의 키에 해당하는 값을 가져올 수 있습니다. 이 방법은 마이크로서비스가 상대적으로 적은 상황에 적합합니다. 프로젝트, 업무가 복잡할 때는 논리
