• Redis - 保证数据库与缓存数据一致性 - 如何保证两步都执行成功？

    缓存和数据库一致性问题，看这篇就够了,中写道:无论是更新缓存还是删除缓存，只要第二步发生失败，那么就会导致数据库和缓存不一致。

    1.重试

    2.异步重试

    3.订阅binlog日志

            具体来讲就是，我们的业务应用在修改数据时，「只需」修改数据库，无需操作缓存。

    那什么时候操作缓存呢？这就和数据库的「变更日志」有关了。

    拿 MySQL 举例，当一条数据发生修改时，MySQL 就会产生一条变更日志（Binlog），我们可以订阅这个日志，拿到具体操作的数据，然后再根据这条数据，去删除对应的缓存。

    使用canal中间件,订阅数据库变更日志,将变更传递到下游消息队列.

    环境准备:

    mysql数据库:   1.开启binlog,row模式,server-id = 1不与canal的slaveId重复

                            2.账号授权

    canal下载安装配置:

                            1.下载安装

                            2.配置 - rocketMq为例

                                    conf/canal.properties

    1. #################################################
    2. ######### 		common argument		#############
    3. #################################################
    4. # tcp bind ip
    5. canal.ip = 192.168.43.231
    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. #需要修改的地方
    30. canal.serverMode = rocketMQ
    31. # flush meta cursor/parse position to file
    32. canal.file.data.dir = ${canal.conf.dir}
    33. canal.file.flush.period = 1000
    34. ## memory store RingBuffer size, should be Math.pow(2,n)
    35. canal.instance.memory.buffer.size = 16384
    36. ## memory store RingBuffer used memory unit size , default 1kb
    37. canal.instance.memory.buffer.memunit = 1024 
    38. ## meory store gets mode used MEMSIZE or ITEMSIZE
    39. canal.instance.memory.batch.mode = MEMSIZE
    40. canal.instance.memory.rawEntry = true
    41.  
    42. ## detecing config
    43. canal.instance.detecting.enable = false
    44. #canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now()
    45. canal.instance.detecting.sql = select 1
    46. canal.instance.detecting.interval.time = 3
    47. canal.instance.detecting.retry.threshold = 3
    48. canal.instance.detecting.heartbeatHaEnable = false
    49.  
    50. # support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery
    51. canal.instance.transaction.size =  1024
    52. # mysql fallback connected to new master should fallback times
    53. canal.instance.fallbackIntervalInSeconds = 60
    54.  
    55. # network config
    56. canal.instance.network.receiveBufferSize = 16384
    57. canal.instance.network.sendBufferSize = 16384
    58. canal.instance.network.soTimeout = 30
    59.  
    60. # binlog filter config
    61. canal.instance.filter.druid.ddl = true
    62. canal.instance.filter.query.dcl = false
    63. canal.instance.filter.query.dml = false
    64. canal.instance.filter.query.ddl = false
    65. canal.instance.filter.table.error = false
    66. canal.instance.filter.rows = false
    67. canal.instance.filter.transaction.entry = false
    68. canal.instance.filter.dml.insert = false
    69. canal.instance.filter.dml.update = false
    70. canal.instance.filter.dml.delete = false
    71.  
    72. # binlog format/image check
    73. canal.instance.binlog.format = ROW,STATEMENT,MIXED 
    74. canal.instance.binlog.image = FULL,MINIMAL,NOBLOB
    75.  
    76. # binlog ddl isolation
    77. canal.instance.get.ddl.isolation = false
    78.  
    79. # parallel parser config
    80. canal.instance.parser.parallel = true
    81. ## concurrent thread number, default 60% available processors, suggest not to exceed Runtime.getRuntime().availableProcessors()
    82. #canal.instance.parser.parallelThreadSize = 16
    83. ## disruptor ringbuffer size, must be power of 2
    84. canal.instance.parser.parallelBufferSize = 256
    85.  
    86. # table meta tsdb info
    87. canal.instance.tsdb.enable = true
    88. canal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:}
    89. canal.instance.tsdb.url = jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;
    90. canal.instance.tsdb.dbUsername = canal
    91. canal.instance.tsdb.dbPassword = canal
    92. # dump snapshot interval, default 24 hour
    93. canal.instance.tsdb.snapshot.interval = 24
    94. # purge snapshot expire , default 360 hour(15 days)
    95. canal.instance.tsdb.snapshot.expire = 360
    96.  
    97. #################################################
    98. ######### 		destinations		#############
    99. #################################################
    100. #canal.destinations = example
    101. canal.destinations = example
    102. # conf root dir
    103. canal.conf.dir = ../conf
    104. # auto scan instance dir add/remove and start/stop instance
    105. canal.auto.scan = true
    106. canal.auto.scan.interval = 5
    107. # set this value to 'true' means that when binlog pos not found, skip to latest.
    108. # WARN: pls keep 'false' in production env, or if you know what you want.
    109. canal.auto.reset.latest.pos.mode = false
    110.  
    111. canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml
    112. #canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml
    113.  
    114. canal.instance.global.mode = spring
    115. canal.instance.global.lazy = false
    116. canal.instance.global.manager.address = ${canal.admin.manager}
    117. #canal.instance.global.spring.xml = classpath:spring/memory-instance.xml
    118. canal.instance.global.spring.xml = classpath:spring/file-instance.xml
    119. #canal.instance.global.spring.xml = classpath:spring/default-instance.xml
    120.  
    121. ##################################################
    122. ######### 	      MQ Properties      #############
    123. ##################################################
    124. # aliyun ak/sk , support rds/mq  要修改的
    125. canal.aliyun.accessKey = LTAIs4e56kBVE9
    126. canal.aliyun.secretKey = pJPGAtvJKGWsS
    127. canal.aliyun.uid=
    128. canal.mq.servers = 192.168.56.1:9876
    129. canal.mq.flatMessage = true
    130. canal.mq.canalBatchSize = 50
    131. canal.mq.canalGetTimeout = 100
    132. # Set this value to "cloud", if you want open message trace feature in aliyun.
    133. canal.mq.accessChannel = local
    134. canal.mq.database.hash = true
    135. canal.mq.send.thread.size = 30
    136. canal.mq.build.thread.size = 8
    137.  
    138. ##################################################
    139. ######### 		     Kafka 		     #############
    140. ##################################################
    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 = redis-test-producer
    159. rocketmq.enable.message.trace = false
    160. rocketmq.customized.trace.topic =
    161. rocketmq.namespace =
    162. #需要修改的地方，多个值用分号隔开
    163. rocketmq.namesrv.addr = 192.168.56.1:9876
    164. rocketmq.retry.times.when.send.failed = 0
    165. rocketmq.vip.channel.enabled = false
    166. rocketmq.tag = canal_tag
    167.  
    168. ##################################################
    169. ######### 		    RabbitMQ	     #############
    170. ##################################################
    171. # rabbitmq.host =
    172. # rabbitmq.virtual.host =
    173. # rabbitmq.exchange =
    174. # rabbitmq.username =
    175. # rabbitmq.password =
    176. # rabbitmq.deliveryMode =

    instance.properties

    1. ## mysql serverId
    2. canal.instance.mysql.slaveId = 5
    3. #position info，需要改成自己的数据库信息
    4. canal.instance.master.address = 127.0.0.1:3306 
    5. canal.instance.master.journal.name = 
    6. canal.instance.master.position = 
    7. canal.instance.master.timestamp = 
    8. #canal.instance.standby.address = 
    9. #canal.instance.standby.journal.name =
    10. #canal.instance.standby.position = 
    11. #canal.instance.standby.timestamp = 
    12. #username/password，需要改成自己的数据库信息
    13. canal.instance.dbUsername = canal  
    14. canal.instance.dbPassword = canal
    15. canal.instance.defaultDatabaseName =
    16. canal.instance.connectionCharset = UTF-8
    17. #table regex
    18. canal.instance.filter.regex =.*\\..*
    19. canal.mq.topic = canal_topic
    20. canal.mq.partition = 0

                                    

            3.启动

        rocket相关

    测试:

            
        
            com.alibaba.otter
            canal.client
            1.1.4
        
    package com.cloud.redis.consumer.binlog;

rocket监听
@Slf4j
@Component
@RocketMQMessageListener(topic = "canal_topic",
                        selectorExpression = "*",
                        consumerGroup = "canal_consumer",
messageModel = MessageModel.CLUSTERING, //广播...
consumeMode = ConsumeMode.ORDERLY) //同时,顺序
public class MyRocketMQListener implements RocketMQListener {
    @Override
    public void onMessage(Object message) {
        System.out.println("MQListener1..."+message);
    }
}

问题
1.mysql服务报错:MySQL启动失败提示:本地计算机上的Mysql服务启动后停止 - 查看data下.err日志解决2.连接问题:D:\software\canal\canal\logs\user_date - caching_sha2_password Auth failed -select host, user, authentication_string, plugin from userALTER USER 'canal'@'%' IDENTIFIED WITH mysql_native_password BY 'canal';
3.c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - ---> begin to find start position, it will be long time for reset or first position
4. The specified topic is blank - 设置canal.mq.topic
5.Connection refused - canal.serverMode = tcp
6.NoClassDefFoundError:org/apache/rocketmq/client/consumer/DefaultLitePullConsumer -  导包jar
7.Error[UPDATE command denied to user 'canal'@'localhost' for table 'canal_node_server'] - 1. GRANT SELECT, INSERT, UPDATE, DELETE, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%'; 2.flush privileges; 3.show grants for 'canal'@'%';

    QuickStart · alibaba/canal Wiki · GitHub

  • 相关阅读:
    Java预习43
    上传镜像到 docker hub 中
    【计网】链路层
    智能优化算法Matlab源码大礼包领取
    门控循环单元（GRU）【动手学深度学习v2】
    Android 判断当前线程是否是主线程的方法
    Python之第六章 内置容器 --- 字符串
    java-net-php-python-ssm儿童演出礼服租赁网站计算机毕业设计程序
    C语言中常见的逻辑错误
    ZigBee 3.0理论教程-通用-1-11：安全加密-网络层（NWK）安全
  • 原文地址：https://blog.csdn.net/xiaoxiamiqianqian/article/details/126555516