热门标签
热门文章
- 1AI大模型安全挑战和安全要求解读_人工智能大模型 安全
- 2UWB定位技术：璀璨的行业“新星”_uwb 园区 楼 楼层 场景
- 3基于Android平台开发，天气预报APP
- 4Windows AD 组策略 通过脚本修改管理员密码：以安全方式_如何通过ad修改计算机本地admin账号密码
- 5网络基础VlAN配置 Trunk技术（eNSP、Cisco）_不同部门不能通信代码
- 6Error connecting to node kafka1:9092 (id: 0 rack: null) ，marathon docker镜像添加hosts_error connecting to node node1:9092
- 7matlab ode45 初值,为什么设定的初值在ode45里无效？
- 8如何清理Mac上的系统数据，让你告别电脑空间不足的烦恼_mac 系统数据 是什么
- 9windows10家庭版安装docker教程，报Docker Desktop requires Windows 10 Pro or Enterprise version 15063 to run._dockerwin10家庭版
- 10python实例：面向对象学员管理系统_python好友管理系统面向对象
当前位置: article > 正文
Canal监控MySQL数据库实现数据同步_canal+java实现mysql数据库数据变化的监听
作者：盐析白兔 | 2024-07-15 16:35:34
赞
踩
canal+java实现mysql数据库数据变化的监听
canal是阿里巴巴旗下的一款开源项目，纯Java开发。基于数据库增量日志解析，提供增量数据订阅&消费，目前主要支持了MySQL。
canal简介
需求：将MySQL中某些表的数据实时的同步到Redis
设计：当MySQL数据库表发生变化时，通过canal解析数据库增量日志，将信息发送到RocketMQ，我们从MQ拿到数据同步到Redis。
一、准备工作
1、配置MySQL
我的系统是Ubuntu 16.04，MySQL配置文件路径为： /etc/mysql/mysql.conf.d
①修改mysqld.cnf文件内容：
[mysqld]
log-bin=mysql-bin # 添加这一行就ok，开启 Binlog 写入功能
binlog-format=ROW # 选择row模式
server_id=1 # 不能和canal的slaveId重复
- 1
- 2
- 3
- 4
重启MySQL，检查是否开启binlog，在MySQL终端输入以下命令：
show variables like 'log_bin';
- 1
出现如下图所示，则开启成功。
②创建账号，赋予权限
（我这里创建的账号密码为canal，canal）
CREATE USER canal IDENTIFIED BY 'canal';
GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%'; --开启所有权限
FLUSH PRIVILEGES;
- 1
- 2
- 3
2、配置Canal
①下载canal
我下载的版本是canal.deployer-1.1.4.tar.gz
canal下载
②解压
tar -xzvf canal.deployer-1.1.4.tar.gz
- 1
解压之后目录结构如下：
bin
conf
lib
logs
③配置canal
编辑 instance.properties 文件
vim conf/example/instance.properties
- 1
内容如下：
################################################# ## mysql serverId , v1.0.26+ will autoGen canal.instance.mysql.slaveId=1234 # enable gtid use true/false canal.instance.gtidon=false # position info 需要改成自己的数据库信息 canal.instance.master.address=192.168.0.110:3306 canal.instance.master.journal.name= canal.instance.master.position= canal.instance.master.timestamp= 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 #canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb #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 # 这里配置之前创建的MySQL账号 canal.instance.dbUsername=canal canal.instance.dbPassword=canal canal.instance.connectionCharset = UTF-8 # enable druid Decrypt database password #canal.instance.enableDruid=false #canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ== # table regex #canal.instance.filter.regex=.*\\..* ==> 所有库表 # 这里指定监听的数据库表 ==> 数据库名.表名 canal.instance.filter.regex=test.user # test库的user表 # table black regex canal.instance.filter.black.regex= # table field filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2) #canal.instance.filter.field=test1.t_product:id/subject/keywords,test2.t_company:id/name/contact/ch # table field black filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2) #canal.instance.filter.black.field=test1.t_product:subject/product_image,test2.t_company:id/name/contact/ch # mq config # MQ 配置 canal.mq.topic=canal-example # 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,.*\\..* #################################################
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
④启动canal
执行startup.sh脚本
./bin/startup.sh
- 1
查看logs/canal/canal.log
2020-03-24 19:50:41.724 [main] INFO com.alibaba.otter.canal.deployer.CanalLauncher - ## set default uncaught exception handler
2020-03-24 19:50:41.744 [main] INFO com.alibaba.otter.canal.deployer.CanalLauncher - ## load canal configurations
2020-03-24 19:50:41.778 [main] INFO com.alibaba.otter.canal.deployer.CanalStarter - ## start the canal server.
2020-03-24 19:50:41.838 [main] INFO com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[192.168.0.110(192.168.0.110):11111]
2020-03-24 19:50:43.489 [main] INFO com.alibaba.otter.canal.deployer.CanalStarter - ## the canal server is running now ......
- 1
- 2
- 3
- 4
- 5
- 6
查看logs/example/example.log
2020-03-24 19:50:42.182 [main] INFO c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [canal.properties]
2020-03-24 19:50:42.185 [main] INFO c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [example/instance.properties]
2020-03-24 19:50:42.366 [main] WARN o.s.beans.GenericTypeAwarePropertyDescriptor - Invalid JavaBean property 'connectionCharset' being accessed! Ambiguous write methods found next to actually used [public void com.alibaba.otter.canal.parse.inbound.mysql.AbstractMysqlEventParser.setConnectionCharset(java.nio.charset.Charset)]: [public void com.alibaba.otter.canal.parse.inbound.mysql.AbstractMysqlEventParser.setConnectionCharset(java.lang.String)]
2020-03-24 19:50:42.424 [main] INFO c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [canal.properties]
2020-03-24 19:50:42.425 [main] INFO c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [example/instance.properties]
2020-03-24 19:50:42.900 [main] INFO c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-example
2020-03-24 19:50:42.918 [main] WARN c.a.o.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table black filter :
2020-03-24 19:50:43.472 [main] INFO c.a.otter.canal.instance.core.AbstractCanalInstance - start successful....
2020-03-24 19:50:44.495 [destination = example , address = /192.168.0.110:3306 , EventParser] WARN c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - ---> begin to find start position, it will be long time for reset or first position
2020-03-24 19:50:44.504 [destination = example , address = /192.168.0.110:3306 , EventParser] WARN c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - prepare to find start position just last position
{"identity":{"slaveId":-1,"sourceAddress":{"address":"192.168.0.110","port":3306}},"postion":{"gtid":"","included":false,"journalName":"mysql-bin.000007","position":488,"serverId":1,"timestamp":1585041845000}}
2020-03-24 19:50:44.789 [destination = example , address = /192.168.0.110:3306 , EventParser] WARN c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - ---> find start position successfully, EntryPosition[included=false,journalName=mysql-bin.000007,position=488,serverId=1,gtid=,timestamp=1585041845000] cost : 288ms , the next step is binlog dump
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
canal成功启动，下一步进行数据同步
注意：
如果要从最新位置记录数据库的binlog，可以删除cong/example/meta.dat文件。
二、数据同步
canal.serverMode有三种模式： tcp(默认), kafka, RocketMQ，我这里使用的是RocketMQ。当监听的数据库表发生变化时，canal解析MySQL的增量日志后，将信息发送到MQ，我们通过监听MQ获取数据库表的更新信息，然后做相关的处理：同步到redis等操作。
1、修改canal配置文件
①首先确保配置了上面的conf/example/instance.properties配置文件的MQ配置：
# mq config
# MQ 配置
# 配置topic
canal.mq.topic=canal-example
# 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,.*\\..*
#################################################
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
②修改canal 配置文件canal.properties
vim conf/canal.properties
# 可选项: tcp(默认), kafka, RocketMQ canal.serverMode = RocketMQ # kafka/rocketmq 集群配置: 192.168.1.117:9092,192.168.1.118:9092,192.168.1.119:9092 # 指定RocketMQ地址 canal.mq.servers = 127.0.0.1:6667 canal.mq.retries = 0 canal.mq.batchSize = 16384 canal.mq.maxRequestSize = 1048576 canal.mq.lingerMs = 1 canal.mq.bufferMemory = 33554432 # Canal的batch size, 默认50K, 由于kafka最大消息体限制请勿超过1M(900K以下) canal.mq.canalBatchSize = 50 # Canal get数据的超时时间, 单位: 毫秒, 空为不限超时 canal.mq.canalGetTimeout = 100 # 是否为flat json格式对象 canal.mq.flatMessage = true canal.mq.compressionType = none canal.mq.acks = all
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
2、编写客户端代码
①在项目application.yml添加RocketMQ配置信息：
- topic和上面conf/example/instance.properties配置文件中的canal.mq.topic保持一致
- name-server和上面conf/canal.properties配置文件中的canal.mq.servers保持一致
# rocket mq
rocketmq:
name-server: 127.0.0.1:6667
canal:
topic: canal-example
group: canal-group
- 1
- 2
- 3
- 4
- 5
- 6
- 7
编写RocketMQConfig配置文件：
@Configuration
@ConfigurationProperties(prefix = "rocketmq")
@Data
public class RocketMQConfig {
private String nameServer;
private Map<String, String> canal;
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
②编写Consumer端代码：
@Slf4j @Component @Data public class MQConsumer { @Autowired private RocketMQConfig rocketMQConfig; @Autowired private MessageListenerConcurrently rmqMsgListener; @Autowired private RedisConnectionFactory redisConnectionFactory; private DefaultMQPushConsumer mqPushConsumer; @PostConstruct public void defaultConsumer() throws MQClientException { // 指定 RocketMQ 的 group，topic 和 nameServer String group = rocketMQConfig.getCanal().get("group"); String topic = rocketMQConfig.getCanal().get("topic"); String nameServer = rocketMQConfig.getNameServer(); mqPushConsumer = new DefaultMQPushConsumer(group); //指定NameServer地址，多个地址以 ; 隔开 mqPushConsumer.setNamesrvAddr(nameServer); mqPushConsumer.subscribe(topic, "*"); mqPushConsumer.setMessageModel(MessageModel.BROADCASTING); mqPushConsumer.registerMessageListener(rmqMsgListener); mqPushConsumer.start(); /* * connect redis server */ redisConnectionFactory.getConnection(); } @PreDestroy public void destroy() { mqPushConsumer.shutdown(); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
③编写Listener端代码：
@Slf4j @Data @Component public class MQMsgListener implements MessageListenerConcurrently { @Autowired private RedisTemplate redisTemplate; @Autowired private RedisService redisService; /**. * 处理canal发来的数据库更新消息, 同步到redis * * @param list messageList * @param consumeConcurrentlyContext consumeConcurrentlyContext * @return ConsumeConcurrentlyStatus */ @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) { Message rmqMsg = list.get(0); /* * Thread save for local variable */ String dbName; String tableName; String sqlType; String msg = new String(rmqMsg.getBody()); //parser canal message try { JsonNode rootNode = JacksonMapper.readTree(msg); JsonNode dataNode = rootNode.get("data"); if (dataNode.isNull()) { return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } HashOperations<String, String, String> hashOperations = redisTemplate.opsForHash(); ValueOperations<String, String> valueOperations = redisTemplate.opsForValue(); dbName = rootNode.get("database").asText(); tableName = rootNode.get("table").asText(); sqlType = rootNode.get("type").asText(); log.info("******DB Changed:" + tableName + msg); /** * 处理逻辑：通过判断数据库名，表名，操作类型等做相关的处理 * 这里举个例子，操作test数据库的user表 */ if ("user".equals((tableName)) && "test".equals(dbName)) { log.info( "table 【{}】 ,db【{}】", tableName, dbName); if ("DELETE".equals(sqlType)) { //删除操作 log.info( "sqlType【{}】", sqlType); } if ("INSERT".equals(sqlType)) { //插入操作 log.info( "sqlType【{}】", sqlType); } if ("UPDATE".equals(sqlType)) { //更新操作 log.info( "sqlType【{}】", sqlType); } return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } } catch (Exception e) { e.printStackTrace(); //for any exception should return success, otherwise MQ will resend always return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/盐析白兔/article/detail/830224
推荐阅读
相关标签
