大型企业如何实现MySQL到Redis的同步.docx

大型企业如何实现MySQL到Redis的同 步？
导读：本文将重点讨论在超大规模系统中缓存会面临什么样的问题，以及应该使用什么样的 策略来更新缓存。
MySQL > ? redis01缓存穿透
超大规模系统的不能承受之痛如何构建Redis集群？由于集群可以水平扩容，因此只要集群足够大，理论上支持 海量并发就不是问题。但是，如果并发请求数量的基数过大，那么即使只有很小比 率的请求穿透缓存，直接访问数据库的请求其绝对数量也仍然不小。再加上大促期 间的流量峰值，还是会存在因为缓存穿透而引发系统雪崩的风险。
那么，这个问题该如何解决呢？其实方法并不难想到，不让请求穿透缓存就行了。
如今内存存储的价格一路走低，只要能买得起足够多的服务器，Redis集群的容量 就是无限的。我们可以把全量数据都放在Redis集群中，处理读请求的时候，只需connector, ack(batchld) ; // 提交确认。} catch (Throwable t)
{connector, rollback(batchld) ; // 处理失败,回滚数据。}}这个程序的逻辑并不复杂，程序启动并连接到Canal服务后，就不停地拉取数据， 如果没有数据就休眠一会儿，如果有数据就调用processEntries方法处理并更新 缓存。每批数据更新成功之后，都会调用ack方法向Canal服务返回成功响应，如 果失败那么抛出异常之后再回滚。下面是processEntries方法的主要代码： for (CanalEntry. RowData rowData : rowChage. getRowDatasList()) { if (eventType
二二 CanalEntry. EventType. DELETE) { // 删除。jedis. del (row2Key (/zuser_id，z,rowData. getBeforeColumnsList ()));} else if (eventType 二
CanalEntry. EventType. INSERT) { // 插入。jedis. set (row2Key (/zuser_id，z,rowData. getAfterColumnsList()), row2Value(rowData. getAfterColumnsList()));}
else { // 更新。jedis. set (row2Key(〃user_id〃，rowData. getAfterColumnsList()),
row2Value(rowData. getAfterColumnsList()));}} 上述代码会根据事件类型分别进行处理，如果MySQL中的数据删除了，就删除Redis 中对应的数据。如果是更新和插入操作，就调用Redis的SET命令来写入数据。
下面就来启动这个账户缓存更新服务以进行验证。在账户余额表中插入一条记录，SQL命令如下： mysql> insert into account_balance values (888, 100, N0W(), 999);然后，我们再来
看一下Redis缓存,操作命令和输出结果如下:127. 0. 0. 1:6379> get888〃{\〃log_id\〃：\〃999\〃，\〃balance\〃：\〃100\〃，\〃user_id\〃：\〃888\", \〃timestamp\〃：\ "2020-03-08 16:18:10\〃}〃
从上述输出结果中我们可以看到，数据已经自动同步到Redis中了。GitHub上可以 下载该例如的完整代码，链接地址是： s://github /liyue2008/canal-to-redis-exampleo03总结
在处理超大规模并发的场景时，由于并发请求的数量非常大，即使只有少量的缓存 穿透，也有可能卡死数据库引发雪崩效应。对于这种情况，我们可以通过Redis缓 存全量数据来彻底防止缓存穿透的问题。对于缓存数据更新的方法，我们可以通过 订阅数据更新的消息队列来异步更新缓存，更通用的方法是，把缓存更新服务伪装 成一个MySQL从节点，订阅MySQL的Binlog,通过Binlog来更新Redis缓存。
需要特别注意的是，无论是通过消息队列还是Canal来异步更新缓存，系统对整个 更新服务的数据可靠性和实时性要求都比拟高，数据丧失或者更新慢了，都会造成 Redis中的数据与MySQL中的数据不同步的问题。在把这套方案应用到生产环境之 前，我们需要考虑一旦出现不同步的问题，应该采取什么样的降级或补偿方案。
要读取Redis,而不用访问数据库，这样就完全没有“缓存穿透”的风险了。实际 上，很多大型互联网公司都在使用这种方法。
不过，在Redis中缓存全量数据，又会引发一个新的问题。那就是，缓存中的数据 应该如何更新呢？因为我们取消了缓存穿透的机制，在这种情况下，如果能从缓存 中直接读到数据，那么可以直接返回，如果没能读到数据，那就只能返回错误了！所 以，当系统更新数据库的数据之后，必须及时更新缓存。
至此，我们又要面对一个老问题：如何保证Redis中的数据与数据库中的数据同步 更新？可以用分布式事务来解决数据一致性的问题，但是这些方法都不太适合用来 更新缓存。原因是，分布式事务对数据更新服务有很强的侵入性。这里仍以下单服 务为例来说明，如果为了更新缓存，增加一个分布式事务，那么无论我们使用哪种 分布式事务，下单服务的性能或多或少都会受到影响。还有一个问题是，如果Redis 本身出现了故障，写入数据失败，那么还会导致下单失败的问题，相当于是降低了下 单服务的性能和可用性，这样肯定是不行的。
对于像订单服务之类的核心业务，一个可行的方法是，启动一个更新订单缓存的服 务，接收订单变更的消息队列(Message Queue, MQ)中的消息，然后更新Redis 中缓存的订单数据。使用订单变更消息更新缓存的结构如图1所示。因为对于这类 核心的业务数据，使用方通常会非常多，服务本来就需要向外发送消息，增加一个 消费订阅，基本上不会增加额外的开发本钱，也不需要对订单服务本身做出任何更 改。
订单库
订单缓存
图1使用订单变更消息更新缓存
对于上述方法，我们唯一需要担忧的问题是，如果消息丧失了，应该怎么办？因为 现在消息是缓存数据的唯一来源，一旦出现消息丧失的问题，缓存里缺失的那条数 据就会永远也无法补上，所以，必须保证整个消息链条的可靠性。不过，好在现在 的MQ集群（比方Kafka或RocketMQ）,都拥有高可用性和高可靠性的保证机制, 只要能事先正确配置好，就可以满足数据的可靠性要求。
像订单服务这样，由于本来就有现成的数据变更消息可以订阅，因此像这样更新缓 存也是一个不错的选择，因为这种方式实现起来很简单，对系统的其他模块也完全 没有侵入。
02使用Binlog实时更新Redis缓存如果我们要缓存的数据，原本就没有一份数据更新的消息队列可以订阅，又该怎么 办呢？下面就来介绍很多大型互联网企业所采用的，也是更通用的解决方案。
数据更新服务只负责处理业务逻辑，更新MySQL,完全不用考虑如何更新缓存。负 责更新缓存的服务，把自己伪装成一个MySQL的从节点，从MySQL接收并解析Binlog 之后，就可以得到实时的数据变更信息，然后该服务就会根据这个变更信息去更新 Redis缓存。订阅Binlog更新缓存的结构如图2所示。
订单服务更新
Binlog
订单库
订单缓存
更新服务
更新
订单缓存
图2订阅Binlog更新缓存的结构
订阅Binlog更新缓存的方案，相较于上文中接收消息更新Redis缓存的方案，两 者的实现思路其实是一样的，都是异步实时订阅数据变更信息以更新Redis缓存。 只不过，直接读取Binlog这种方式，通用性更强。该方式不会要求订单服务再发 送订单消息，订单更新服务也不用额外考虑如何解决“消息发送失败了该怎么办？ ” 这种数据一致性问题。
除此之外，由于在整个缓存更新链路上，减少了一个收发消息队列的环节，从MySQL 更新到Redis更新的时延变得更短，出现故障的可能性也更低，因此很多大型互联 网企业更青睐于采用这种方案。
订阅Binlog更新缓存的方案唯一的缺点是，实现订单缓存更新服务比拟复杂，该 方案毕竟不像接收消息那样，收到的直接就是订单数据，解析Binlog还是挺麻烦 的。
很多开源的工程都提供了订阅和解析MySQL Binlog的功能，下面就以比拟常用的 开源工程Canal为例来演示，如何实时接收Binlog更新Redis缓存。
Canal通过模拟MySQL主从复制的交互协议，把自己伪装成一个MySQL的从节点， 向MySQL主节点发送dump请求。MySQL收到请求后，就会向Canal开始推送Binlog, Canal解析Binlog字节流之后，将其转换为便于读取的结构化数据，供下游程序订 阅使用。图3展示了如何使用Canal订阅Binlog更新Redis中的订单缓存。
Bi r log
Canal
j Canal Entry
订单缓存
更新服务
更新
订单缓存
图3使用Canal订阅Binlog更新缓存
在这个例如中，MySQL和Redis都在本地的默认端口上运行，MySQL的端口为3306, Redis的端口为6379。为了便于大家操作，下面还是以第5章中提到的账户余额表 account balance作为演示数据。
首先，下载并在本地解压Canal当前最新的1. 1.4版本，操作命令如下:
wget. com/alibaba/canal/re leases/down load/canal-1. 1. 4/canal. deployer-1 .1. 4. tar. gztar zvfx canal, deployer-1. 1. 4. tar. gz
然后，配置MySQL,我们需要在MySQL的配置文件中开启Binlog,并将Binlog的 格式设置为ROW，配置项如下：
[mysqld] log-bin=mysql-bin # 开启 Binlogo binlog-format=ROW # 将 Binlog 格式设置为ROWo server_id=l # 配置一个 ServerlD。
接下来，为Canal新建一个专门的MySQL用户并授权，以确保这个用户有复制Binlog 的权限，具体操作的SQL命令如下：
CREATE USER canal IDENTIFIED BY ' canal' ;GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *. * TO 'canal'®%' ; FLUSH PRIVILEGES;然后，重启MySQL,以确保所有的配置都能生效。重启后再检查一下当前的Binlog 文件和位置，SQL命令和输出结果具体如下:
mysql> show masterstatus; ++++++
| File| Position | Binlog Do DB | Binlog lgnore DB | Executed Gtid Set | ++
一++++1binlog. 0000091155|II
I ++++++记录下File和Position两列的值，然后再来配置Canal。编辑Canal的实例配置 文件 canal/conf/example/instance, properties ,以便让 Canal 连接到我们的 MySQL上，具体配置如下：
canal, instance. gtidon=false^ position7/?/?;canal. instance, master. address=127. 0. 0. 1:3306canal. instance, master, journal, n ame=binlog. 000009canal. instance, master. position=155canal. instance, master, timest amp=canal. instance, master, gtid二#
userncinie/pcisswoi^anal. instance. dbUsername=canalcanal, instance. dbPassword=canal canal, instance. connectionCharset = UTF-8canal・ instance, default Da tabas eName= test， table rogoxcanal. instance, filter. regex=. *\\・・ *这个配置文件需要配置MySQL的连接地址、库名、用户名和密码，除此之外，还要 配置 canal, instance, master, journal, name 和 canal, instance, master, position 这两个属性，取值就是刚刚记录的File和Position两列。然后就可以启动Canal 服务了，命令如下：
canal/bin/startup, sh启动之后再查看一下日志文件canal/logs/example/exaniple. log,如果日志中没有 报错信息，就说明Canal服务已启动成功并连接到我们的MySQL上了。
Canal服务启动之后，会开启一个端口（11111）等待客户端连接，客户端连接上 Canal服务之后，就可以从Canal服务拉取（PULL）数据了，每拉取一批数据，正 确写入Redis之后，需要向Canal服务返回处理成功的响应。如果发生客户端程序 宕机，或者处理失败等异常情况，Canal服务没有收到处理成功的响应，那么下次 客户端来拉取的就还是同一批数据，这样就可以保证读到的Binlog顺序不会乱， 并且不会丧失数据。
接下来，我们来开发一个账户余额缓存的更新程序，以下代码都是用Java语言编写的： while (true) { Message message = connector. getWithoutAck(batchSize);
指定数量的数据。long batchld = message. getld () ; try {int size =message. getEntries().size();if (batchld 二二一1 || size 二二 0)
{Thread, sleep(1000);} elseprocessEntries(message. getEntries(), jedis);