• Canal + Kafka 同步 MySQL 数据到 Redis

    解决缓存和数据库一致性问题

    一般来说，缓存中的数据没什么问题，但是数据库更新后，就容易出现缓存（Redis）和数据库（MySQL）间的数据一致性问题。由于写和读是并发的，没法保证顺序，就会出现缓存和数据库的数据不一致的问题

    Canal工作原理
    canal模拟mysql slave的交互协议，伪装自己为mysql slave，向mysql master发送dump协议
    mysql master收到dump请求，开始推送binary log给slave(也就是canal)
    canal解析binary log对象(原始为byte流)

    可参考官方文档

    QuickStart · alibaba/canal Wiki · GitHub

    1、Kafka，ZK安装略，本博主其它文章有

            安装好后，先启动ZK，在启动Kafka

    2、win10编辑my.ini

    1. [mysqld]
    2. #开启 binlog
    3. log-bin=mysql-bin
    4. #选择 ROW 模式
    5. binlog-format=ROW 
    6. # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复
    7. server_id=1

            

    1. 开启授权
    2. CREATE USER canal IDENTIFIED BY 'canal';  
    3. GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
    4. -- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
    5. FLUSH PRIVILEGES;

    3、下载canal

            下载地址：Releases · alibaba/canal · GitHub

    解压后会有5文件夹

    进入conf

    编辑canal.properties

    1. #################################################
    2. ######### 		common argument		#############
    3. #################################################
    4. # tcp bind ip
    5. canal.ip =
    6. # register ip to zookeeper
    7. canal.register.ip =
    8. canal.port = 11111
    9. canal.metrics.pull.port = 11112
    10. # canal instance user/passwd
    11. # canal.user = canal
    12. # canal.passwd = E3619321C1A937C46A0D8BD1DAC39F93B27D4458
    13.  
    14. # canal admin config
    15. #canal.admin.manager = 127.0.0.1:8089
    16. canal.admin.port = 11110
    17. canal.admin.user = admin
    18. canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
    19. # admin auto register
    20. #canal.admin.register.auto = true
    21. #canal.admin.register.cluster =
    22. #canal.admin.register.name =
    23.  
    24. canal.zkServers =
    25. # flush data to zk
    26. canal.zookeeper.flush.period = 1000
    27. canal.withoutNetty = false
    28. # tcp, kafka, rocketMQ, rabbitMQ
    29. canal.serverMode = kafka
    30. # flush meta cursor/parse position to file
    31. canal.file.data.dir = ${canal.conf.dir}
    32. canal.file.flush.period = 1000
    33. ## memory store RingBuffer size, should be Math.pow(2,n)
    34. canal.instance.memory.buffer.size = 16384
    35. ## memory store RingBuffer used memory unit size , default 1kb
    36. canal.instance.memory.buffer.memunit = 1024 
    37. ## meory store gets mode used MEMSIZE or ITEMSIZE
    38. canal.instance.memory.batch.mode = MEMSIZE
    39. canal.instance.memory.rawEntry = true
    40.  
    41. ## detecing config
    42. canal.instance.detecting.enable = false
    43. #canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now()
    44. canal.instance.detecting.sql = select 1
    45. canal.instance.detecting.interval.time = 3
    46. canal.instance.detecting.retry.threshold = 3
    47. canal.instance.detecting.heartbeatHaEnable = false
    48.  
    49. # support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery
    50. canal.instance.transaction.size =  1024
    51. # mysql fallback connected to new master should fallback times
    52. canal.instance.fallbackIntervalInSeconds = 60
    53.  
    54. # network config
    55. canal.instance.network.receiveBufferSize = 16384
    56. canal.instance.network.sendBufferSize = 16384
    57. canal.instance.network.soTimeout = 30
    58.  
    59. # binlog filter config
    60. canal.instance.filter.druid.ddl = true
    61. canal.instance.filter.query.dcl = false
    62. canal.instance.filter.query.dml = false
    63. canal.instance.filter.query.ddl = false
    64. canal.instance.filter.table.error = false
    65. canal.instance.filter.rows = false
    66. canal.instance.filter.transaction.entry = false
    67. canal.instance.filter.dml.insert = false
    68. canal.instance.filter.dml.update = false
    69. canal.instance.filter.dml.delete = false
    70.  
    71. # binlog format/image check
    72. canal.instance.binlog.format = ROW,STATEMENT,MIXED 
    73. canal.instance.binlog.image = FULL,MINIMAL,NOBLOB
    74.  
    75. # binlog ddl isolation
    76. canal.instance.get.ddl.isolation = false
    77.  
    78. # parallel parser config
    79. canal.instance.parser.parallel = true
    80. ## concurrent thread number, default 60% available processors, suggest not to exceed Runtime.getRuntime().availableProcessors()
    81. #canal.instance.parser.parallelThreadSize = 16
    82. ## disruptor ringbuffer size, must be power of 2
    83. canal.instance.parser.parallelBufferSize = 256
    84.  
    85. # table meta tsdb info
    86. canal.instance.tsdb.enable = true
    87. canal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:}
    88. canal.instance.tsdb.url = jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;
    89. canal.instance.tsdb.dbUsername = canal
    90. canal.instance.tsdb.dbPassword = canal
    91. # dump snapshot interval, default 24 hour
    92. canal.instance.tsdb.snapshot.interval = 24
    93. # purge snapshot expire , default 360 hour(15 days)
    94. canal.instance.tsdb.snapshot.expire = 360
    95.  
    96. #################################################
    97. ######### 		destinations		#############
    98. #################################################
    99. canal.destinations = example
    100. # conf root dir
    101. canal.conf.dir = ../conf
    102. # auto scan instance dir add/remove and start/stop instance
    103. canal.auto.scan = true
    104. canal.auto.scan.interval = 5
    105. # set this value to 'true' means that when binlog pos not found, skip to latest.
    106. # WARN: pls keep 'false' in production env, or if you know what you want.
    107. canal.auto.reset.latest.pos.mode = false
    108.  
    109. canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml
    110. #canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml
    111.  
    112. canal.instance.global.mode = spring
    113. canal.instance.global.lazy = false
    114. canal.instance.global.manager.address = ${canal.admin.manager}
    115. #canal.instance.global.spring.xml = classpath:spring/memory-instance.xml
    116. canal.instance.global.spring.xml = classpath:spring/file-instance.xml
    117. #canal.instance.global.spring.xml = classpath:spring/default-instance.xml
    118.  
    119. ##################################################
    120. ######### 	      MQ Properties      #############
    121. ##################################################
    122. # aliyun ak/sk , support rds/mq
    123. canal.aliyun.accessKey =
    124. canal.aliyun.secretKey =
    125. canal.aliyun.uid=
    126.  
    127. canal.mq.flatMessage = true
    128. canal.mq.canalBatchSize = 50
    129. canal.mq.canalGetTimeout = 100
    130. # Set this value to "cloud", if you want open message trace feature in aliyun.
    131. canal.mq.accessChannel = local
    132.  
    133. canal.mq.database.hash = true
    134. canal.mq.send.thread.size = 30
    135. canal.mq.build.thread.size = 8
    136.  
    137. ##################################################
    138. ######### 		     Kafka 		     #############
    139. ##################################################
    140. ##kafka地址
    141. kafka.bootstrap.servers = 127.0.0.1:9092
    142. kafka.acks = all
    143. kafka.compression.type = none
    144. kafka.batch.size = 16384
    145. kafka.linger.ms = 1
    146. kafka.max.request.size = 1048576
    147. kafka.buffer.memory = 33554432
    148. kafka.max.in.flight.requests.per.connection = 1
    149. kafka.retries = 0
    150.  
    151. kafka.kerberos.enable = false
    152. kafka.kerberos.krb5.file = "../conf/kerberos/krb5.conf"
    153. kafka.kerberos.jaas.file = "../conf/kerberos/jaas.conf"
    154.  
    155. ##################################################
    156. ######### 		    RocketMQ	     #############
    157. ##################################################
    158. rocketmq.producer.group = test
    159. rocketmq.enable.message.trace = false
    160. rocketmq.customized.trace.topic =
    161. rocketmq.namespace =
    162. rocketmq.namesrv.addr = 127.0.0.1:9876
    163. rocketmq.retry.times.when.send.failed = 0
    164. rocketmq.vip.channel.enabled = false
    165. rocketmq.tag = 
    166.  
    167. ##################################################
    168. ######### 		    RabbitMQ	     #############
    169. ##################################################
    170. rabbitmq.host =
    171. rabbitmq.virtual.host =
    172. rabbitmq.exchange =
    173. rabbitmq.username =
    174. rabbitmq.password =
    175. rabbitmq.deliveryMode =

    进入canal\conf\example

    编辑instance.properties

    1. #################################################
    2. ## mysql serverId , v1.0.26+ will autoGen
    3. # canal.instance.mysql.slaveId=0
    4.  
    5. # enable gtid use true/false
    6. canal.instance.gtidon=false
    7.  
    8. # position info
    9. canal.instance.master.address=127.0.0.1:3306
    10. canal.instance.master.journal.name=
    11. canal.instance.master.position=
    12. canal.instance.master.timestamp=
    13. canal.instance.master.gtid=
    14.  
    15. # rds oss binlog
    16. canal.instance.rds.accesskey=
    17. canal.instance.rds.secretkey=
    18. canal.instance.rds.instanceId=
    19.  
    20. # table meta tsdb info
    21. canal.instance.tsdb.enable=true
    22. #canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
    23. #canal.instance.tsdb.dbUsername=canal
    24. #canal.instance.tsdb.dbPassword=canal
    25.  
    26. #canal.instance.standby.address =
    27. #canal.instance.standby.journal.name =
    28. #canal.instance.standby.position =
    29. #canal.instance.standby.timestamp =
    30. #canal.instance.standby.gtid=
    31.  
    32. # username/password 这里是你的数据库
    33. canal.instance.dbUsername=root
    34. canal.instance.dbPassword=12345678
    35. canal.instance.connectionCharset = UTF-8
    36. # enable druid Decrypt database password
    37. canal.instance.enableDruid=false
    38. #canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==
    39.  
    40. # table regex 这里是你要监听的表，多个用，隔开
    41. canal.instance.filter.regex=test.user
    42. # table black regex
    43. canal.instance.filter.black.regex=mysql\\.slave_.*
    44. # table field filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
    45. #canal.instance.filter.field=test1.t_product:id/subject/keywords,test2.t_company:id/name/contact/ch
    46. # table field black filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
    47. #canal.instance.filter.black.field=test1.t_product:subject/product_image,test2.t_company:id/name/contact/ch
    48.  
    49. # mq config 这里是你要用哪个topic监听
    50. canal.mq.topic=myTestTopic
    51. # dynamic topic route by schema or table regex
    52. #canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..*
    53. canal.mq.partition=0
    54. # hash partition config
    55. #canal.mq.partitionsNum=3
    56. #canal.mq.partitionHash=test.table:id^name,.*\\..*
    57. #canal.mq.dynamicTopicPartitionNum=test.*:4,mycanal:6
    58. #################################################
    59.  
    60.

    启动canal

    双击bin目录的 startup.bat

    4、在pom.xml引入Kafka、canal的依赖

    1. <dependency>
    2.     <groupId>org.springframework.kafka</groupId>
    3.     <artifactId>spring-kafka</artifactId>
    4. </dependency>
    5. <dependency>
    6.     <groupId>org.springframework.kafka</groupId>
    7.     <artifactId>spring-kafka-test</artifactId>
    8.     <scope>test</scope>
    9. </dependency>
    10. <dependency>
    11.     <groupId>com.alibaba.otter</groupId>
    12.     <artifactId>canal.client</artifactId>
    13.     <version>1.1.4</version>
    14. </dependency>

    yml添加Kafka的配置

    1.   kafka:
    2.     #kafka配置
    3.     bootstrap-servers: 127.0.0.1:9092
    4.     producer: #生产者
    5.       retries: 3 #设置大于0的值，则客户端会将发送失败的记录重新发送的次数
    6.       # 每次批量发送消息的数量
    7.       batch-size: 16384
    8.       buffer-memory: 33554432
    9.       # 指定消息key和消息体的编解码方式
    10.       key-serializer: org.apache.kafka.common.serialization.StringSerializer
    11.       value-serializer: org.apache.kafka.common.serialization.StringSerializer
    12.       acks: -1
    13.     consumer:
    14.       # 指定默认消费者group id
    15.       group-id: my-consumer-group
    16.       auto-offset-reset: earliest
    17.       enable-auto-commit: false
    18.       auto-commit-interval: 5000
    19.       # 指定消息key和消息体的编解码方式
    20.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    21.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    22.       
    23.     listener:
    24.       ack-mode: manual_immediate

    5、定义消费者

    1.  
    2. import com.fan.li.myspringboot.util.RedisClient;
    3. import lombok.extern.slf4j.Slf4j;
    4. import org.apache.kafka.clients.consumer.ConsumerRecord;
    5. import org.springframework.beans.factory.annotation.Autowired;
    6. import org.springframework.kafka.annotation.KafkaListener;
    7. import org.springframework.kafka.support.Acknowledgment;
    8. import org.springframework.stereotype.Component;
    9.  
    10. /**
    11.  * @ClassName CanalConsumer
    12.  * @Description TODO
    13.  * @Author fan
    14.  * @Date 2024/2/23 14:11
    15.  * @Version 1.0
    16.  */
    17. @Component
    18. @Slf4j
    19. public class CanalConsumer {
    20.  
    21.     @Autowired
    22.     RedisClient redisClient;//redis工具类，自己封装即可
    23.  
    24.     @KafkaListener(topics = "myTestTopic", groupId = "canalKafka-groupId")
    25.     public void canalListenerKafkaToRedis(ConsumerRecord<String, String> record, Acknowledgment ack) {
    26.         //数据表变动的值
    27.         String value = record.value();
    28.         log.info("数据表发生变动:{}", value);
    29.         //简单存入redis
    30.         redisClient.setKey("你的key",value);
    31.         //手动提交
    32.         ack.acknowledge();
    33.     }
    34. }
    35.

    6、测试：往数据库的表插入一条数据，观察后台日志

  • 相关阅读:
    Redis哨兵模式
    就是一整个爱住，你们大胆飞，我就是最坚强的后盾——Java面试突击宝典
    测试部门来了个00后卷王之王，老油条感叹真干不过，但是...
    ffmpeg音频重采样
    应用程序无法正常启动0xc000007b的解决策略，多种解决方法分享
    几种常见采样方法及原理
    less和scss语法详解
    js根据预设条件定义数组元素
    Ansible-playbook循环学习
    react是否支持给标签设置自定义的属性，比如给video标签设置webkit-playsinline？
  • 原文地址：https://blog.csdn.net/Zxdwr520/article/details/136256992