最基础的概念，什么是幂等性？幂等性：提交多次的情况下，结果都一样。比如数据库查询，可称为天然幂等性，即查询多次结果都一样，无需人为去做幂等性操作。但是update table set value=value+1 where id=1,每次执行的结构都会发生变化，不是幂等。inter into table(id,name)values(1,‘name’),如id不是主键或者没有唯一索引，重复操作上面的业务，会插入多条数据，不具备幂等性；所以我们在什么情景下需要确保幂等性呢？用户多次点击保存按钮用户保存成功后，返回上一页再次保存微服务相互调用，由于网络原因，导致请求失败解决方案一、token机制：1、根据业务场景，判断哪些业务存在幂等性问题，在执行业务之前先获取token，将token缓存止redis中2、调用业务接口时，将token携带过去，一般放在请求头，作为Auth认证3、服务器判断token是否存在于redis中，存在表示第一次请求，然后删除token，继续执行业务4、如果不存在，则表示反复操作，不执行业务逻辑，直接返回重复标志！结束风险性：业务执行前删除还是后删除token？如果是执行后删除，在业务执行中，未删除token，用户又点了请求进来，那么则无法保障幂等性。如果是执行前删除，在分布式下，用户快速请求2次，这时2个请求同时到redis去获取token，对比成功，同时删除，同时执行业务，那么也无法保障幂等性。so：使用执行前删除，在分布式情况下，获取，对比，删除必须确保原子性，所以要加分布式锁。二、加锁1、数据库锁select * from table where … for update2、业务层面加分布式锁将获取、对比、删除作为一个原子性的操作加锁，处理完成后释放锁，确保串行操作。三、约束数据库唯一约束：通过主键、唯一索引，确保无法重复新增同一笔数据，这就能确保幂等性

## 一、redis哨兵+主从的问题 假设我们在一台主从机器上配置了200G内存，但是业务需求是需要500G的时候，主从结构+哨兵可以实现高可用故障切换+冗余备份，但是并不能解决数据容量的问题，用哨兵，redis每个实例也是全量存储，每个redis存储的内容都是完整的数据，浪费内存且有木桶效应。 为了最大化利用内存，可以采用cluster集群，就是分布式存储。即每台redis存储不同的内容。 Redis 分布式方案一般有两种： ①、客户端分区方案：优点是分区逻辑可控，缺点是需要自己处理数据路由、高可用、故障转移等问题，比如在redis2.8之前通常的做法是获取某个key的hashcode，然后取余分布到不同节点，不过这种做法无法很好的支持动态伸缩性需求，一旦节点的增或者删操作，都会导致key无法在redis中命中。  ②、代理方案：优点是简化客户端分布式逻辑和升级维护便利，缺点是加重架构部署复杂度和性能损耗，比如twemproxy、Codis  ③、Redis官方为我们提供了专有的集群方案：Redis Cluster，它非常优雅地解决了 Redis 集群方面的问题，部署方便简单，因此理解应用好 Redis Cluster 将极大地解放我们使用分布式 Redis 的工作量。 ## 二、Redis Cluster ### 1、简介 Redis Cluster 是 Redis 的分布式解决方案，在3.0版本正式推出，有效地解决了 Redis 分布式方面的需求。当遇到单机内存、并发、流量等瓶颈时，可以采用 Cluster 架构方案达到负载均衡的目的。 架构图：  在上面这个图中，每一个蓝色的圈都代表着一个redis的服务器节点。它们任何两个节点之间都是相互连通的。客户端可以与任何一个节点相连接，然后就可以访问集群中的任何一个节点，对其进行存取和其他操作。 Redis 集群提供了以下两个好处： ①、将数据自动切分到多个节点的能力。 ②、当集群中的一部分节点失效或者无法进行通讯时， 仍然可以继续处理命令请求的能力，拥有自动故障转移的能力。 ### 2、redis cluster VS replication + sentinel如何选择？ 如果你的数据量很少，主要是承载高并发高性能的场景，比如你的缓存一般就几个G，单机足够了。 Replication：一个mater，多个slave，要几个slave跟你的要求的读吞吐量有关系，结合sentinel集群，去保证redis主从架构的高可用性，就可以了。 redis cluster：主要是针对海量数据+高并发+高可用的场景，海量数据，如果你的数据量很大，那么建议就用redis cluster。 ### 3、Redis Cluster集群中的数据分布是如何进行的？ 什么是数据分布？数据分布有两种方式，顺序分区和哈希分区。 分布式数据库首先要解决把整个数据集按照分区规则映射到多个节点的问题，即把数据集划分到多个节点上，每个节点负责整体数据的一个子集。 顺序分布就是把一整块数据分散到很多机器中，如下图所示。  顺序分布一般都是平均分配的。 哈希分区： 如下图所示，1~100这整块数字，通过 hash 的函数，取余产生的数。这样可以保证这串数字充分的打散，也保证了均匀的分配到各台机器上。  哈希分布和顺序分布只是场景上的适用。哈希分布不能顺序访问，比如你想访问1~100，哈希分布只能遍历全部数据，同时哈希分布因为做了 hash 后导致与业务数据无关了。  ### 4、数据倾斜与数据迁移跟节点伸缩 顺序分布是会导致数据倾斜的，主要是访问的倾斜。每次点击会重点访问某台机器，这就导致最后数据都到这台机器上了，这就是顺序分布最大的缺点。 但哈希分布其实是有个问题的，当我们要扩容机器的时候，专业上称之为“节点伸缩”，这个时候，因为是哈希算法，会导致数据迁移。 ### 5、哈希分区方式 因为redis-cluster使用的就是哈希分区规则所以分析下几种分区形式 **5.1、节点取余分区** 使用特定的数据（包括redis的键或用户ID），再根据节点数量N，使用公式：hash(key)%N计算出一个0~（N-1）值，用来决定数据映射到哪一个节点上。即哈希值对节点总数取余。 缺点：当节点数量N变化时（扩容或者收缩），数据和节点之间的映射关系需要重新计算，这样的话，按照新的规则映射，要么之前存储的数据找不到，要么之前数据被重新映射到新的节点（导致以前存储的数据发生数据迁移） 实践：常用于数据库的分库分表规则，一般采用预分区的方式，提前根据数据量规划好分区数，比如划分为512或1024张表，保证可支撑未来一段时间的数据量，再根据负载情况将表迁移到其他数据库中。 **5.2、一致性哈希** 一致性哈希分区（Distributed Hash Table）实现思路是为系统中每个节点分配一个 token，范围一般在0~232，这些 token 构成一个哈希环。数据读写执行节点查找操作时，先根据 key 计算 hash 值，然后顺时针找到第一个大于等于该哈希值的 token 节点  上图就是一个一致性哈希的原理解析。 假设我们有 n1~n4 这四台机器，我们对每一台机器分配一个唯一 token，每次有数据（图中黄色代表数据），一致性哈希算法规定每次都顺时针漂移数据，也就是图中黄色的数 据都指向 n3。 这个时候我们需要增加一个节点 n5，在 n2 和 n3 之间，数据还是会发生漂移（会偏移到大于等于的节点），但是这个时候你是否注意到，其实只有 n2~n3 这部分的数据被漂移，其他的数据都是不会变的，这种方式相比节点取余最大的好处在于加入和删除节点只影响哈希环中相邻的节点，对其他节点无影响 缺点：每个节点的负载不相同，因为每个节点的hash是根据key计算出来的,换句话说就是假设key足够多，被hash算法打散得非常均匀，但是节点过少，导致每个节点处理的key个数不太一样，甚至相差很大，这就会导致某些节点压力很大 实践：加减节点会造成哈希环中部分数据无法命中，需要手动处理或者忽略这部分数据，因此一致性哈希常用于缓存场景。 **5.3、虚拟槽分区\*（目前在redis集群中 数据存储和读取常用的方式就是这种 槽 的方式）\** 虚拟槽分区巧妙地使用了哈希空间，使用分散度良好的哈希函数把所有数据映射到一个固定范围的整数集合中，整数定义为槽（slot）。这个范围一般远远大于节点数，比如 Redis Cluster 槽范围是0~16383（也就是16384个槽。redis集群规定了16384个槽，这些槽将会平均分配给不同的redis节点）。槽是集群内数据管理和迁移的基本单位（也就是说 数据是存储在槽中，而槽被分配在了不同的redis节点中）。采用大范围槽的主要目的是为了方便数据拆分和集群扩展。每个节点会负责一定数量的槽，具体看下图所示。 当前集群有5个节点，每个节点平均大约负责3276个槽。由于采用高质量的哈希算法，每个槽所映射的数据通常比较均匀，将数据平均划分到5个节点进行数据分区。Redis Cluster 就是采用虚拟槽分区，下面就介绍 Redis 数据分区方法。  **上图步骤大概执行如下（数据写入）：** ①、redis根据所给定的key进行CRC16算法之后 算出一个结果，然后 再对该结果 进行16384取模（即 对该结果进行16384求余数），得到一个槽。 这样每个key都会对应一个编号在0---16383之间的槽号码，redis会根据节点数量大致均等的原则将哈希槽映射到不同的节点上。比如有3个redis节点，把16384分成3段，每个节点承担一段范围的哈希槽。 **注意：**这里是对16384进行取模，上图中那个CRC16(key)&16383写错了 正确公式为：CRC16(key)%16384 ②、再根据所得的槽位数，获得这个槽所在的redis节点（假设是Z节点）。 ③、连接Z节点，将该key保存到Z这台redis节点上。 **数据读取也是同理，都是根据key得到槽，根据槽得到该槽所在的redis节点，然后连接该redis节点进行数据读取操作。** 每当 key 访问过来，Redis Cluster 会计算哈希值是否在这个区间里。它们彼此都知道对应的槽在哪台机器上，这样就能做到平均分配了。 **redis-cluster集群方面有一些限制：** Redis集群相对单机来说，在功能上存在一些限制，需提前了解，在使用时做好规避。限制如下： key批量操作支持有限。如mset、mget等。 ## 总结redis主从、哨兵、集群的概念： 【redis主从】： 是备份关系， 我们操作主库，数据也会同步到从库。 如果主库机器坏了，从库可以上。就好比你 D盘的片丢了，但是你移动硬盘里边备份有。 【redis哨兵】： 哨兵保证的是HA（高可用），保证特殊情况故障自动切换，哨兵盯着你的“redis主从集群”，如果主库死了，它会告诉你新的老大是谁。 哨兵：主要针对redis主从中的某一个单节点故障后，无法自动恢复的解决方案。（哨兵 保证redis主从的高可用） 【redis集群】： 集群保证的是高并发，因为多了一些兄弟帮忙一起扛。同时集群会导致数据的分散，整个redis集群会分成一堆数据槽，即不同的key会放到不不同的槽中。 集群主要针对单节点容量、高并发问题、线性可扩展性的解决方案。 集群：是为了解决redis主从复制中 单机内存上限和并发问题，假如你现在的服务器内存为256GB，当达到这个内存时redis就没办法再提供服务，同时数据量能达到这个地步写数据量也会很大，容易造成缓冲区溢出，造成从节点无限的进行全量复制导致主从无法正常工作。

## 一、Sentinel 实现原理 接下来我们就来看下 Sentinel 的实现原理，主要分为以下三个步骤。 ①、检测问题，主要讲的是三个定时任务，这三个内部的执行任务可以保证出现问题马上让 Sentinel 知道。 ②、发现问题，主要讲的是主观下线和客观下线。当有一台 Sentinel 机器发现问题时，它就会主观对它主观下线，但是当多个 Sentinel 都发现有问题的时候，才会出现客观下线。 ③、找到解决问题的人，主要讲的是领导者选举，如何在 Sentinel 内部多台节点做领导者选举，选出一个领导者。 ④、解决问题，主要讲的是故障转移，即如何进行故障转移。 ### 三个定时任务： 首先要说的是内部 Sentinel 会执行以下三个定时任务。 每10秒每个 Sentinel 对 Master 和 Slave 执行一次 Info Replication。 每2秒每个 Sentinel 通过 Master 节点的 channel 交换信息（pub/sub）。 每1秒每个 Sentinel 对其他 Sentinel 和 Redis 执行 ping。 第1个定时任务，指的是 Redis Sentinel 可以对 Redis 节点做失败判断和故障转移，在 Redis 内部有三个定时任务作为基础，来 Info Replication 发现 Slave 节点，这个命令可以确定主从关系。 第2个定时任务，类似于发布订阅，Sentinel 会对主从关系进行判定，通过 _sentinel_:hello 频道交互。了解主从关系可以帮助更好的自动化操作 Redis。然后 Sentinel 会告知系统消息给其它 Sentinel 节点，最终达到共识，同时 Sentinel 节点能够互相感知到对方。 第3个定时任务，指的是对每个节点和其它 Sentinel 进行心跳检测，它是失败判定的依据。 ### 主观下线和客观下线： 回顾一下 Sentinel 的配置。 sentinel monitor mymaster 127.0.0.1 6379 2 sentinel down-after-milliseconds mymaster 30000 Sentinel 会 ping 每个节点，如果超过30秒，依然没有回复的话，做下线的判断。 **那么什么是主观下线呢？** 每个 Sentinel 节点对 Redis 节点失败的“偏见”。之所以是偏见，只是因为某一台机器30秒内没有得到回复。 **那么如何做到客观下线呢？** 这个时候需要所有 Sentinel 节点都发现它30秒内无回复，才会达到共识。 ### 领导者选举方式： ①、每个做主观下线的sentinel节点，会向其他的sentinel节点发送命令，要求将它设置成为领导者 ②、收到命令sentinel节点，如果没有同意通过其它节点发送的命令，那么就会同意请求，否则就会拒绝 ③、如果sentinel节点发现自己票数超过半数，同时也超过了sentinel monitor mymaster 127.0.0.1 6379 2 超过2个的时候，就会成为领导者 ④、进行故障转移操作 ### 如何选择“合适的”Slave 节点： Redis 内部其实是有一个优先级配置的，在配置文件中 slave-priority，这个参数是 Salve 节点的优先级配置，如果存在则返回，如果不存在则继续。 当上面这个优先级不满足的时候，Redis 还会选择复制偏移量最大的 Slave节点，如果存在则返回，如果不存在则继续。之所以选择偏移量最大，这是因为偏移量越小，和 Master 的数据越不接近，现在 Master挂掉了，说明这个偏移量小的机器数据也可能存在问题，这就是为什么要选偏移量最大的 Slave 的原因。 如果发现偏移量都一样，这个时候 Redis 会默认选择 runid 最小的节点。 ## 二、生产环境中部署技巧 1）Sentinel 节点不应该部署在一台物理“机器”上。 这里特意强调物理机是因为一台物理机做成了若干虚拟机或者现今比较流行的容器，它们虽然有不同的 IP 地址，但实际上它们都是同一台物理机，同一台物理机意味着如果这台机器有什么硬件故障，所有的虚拟机都会受到影响，为了实现 Sentinel 节点集合真正的高可用，请勿将 Sentinel 节点部署在同一台物理机器上。 2）部署至少三个且奇数个的 Sentinel 节点。 3个以上是通过增加 Sentinel 节点的个数提高对于故障判定的准确性，因为领导者选举需要至少一半加1个节点，奇数个节点可以在满足该条件的基础上节省一个节点。 ## 三、哨兵常见问题 哨兵集群在发现 master 节点挂掉后会进行故障转移，也就是启动其中一个slave 节点为master 节点。在这过程中，可能会导致数据丢失的情况。 ### 1、异步复制导致数据丢失 因为master->slave的复制是异步，所以可能有部分还没来得及复制到slave就宕机了，此时这些部分数据就丢失了。 在异步复制的过程当中，通过min-slaves-max-lag这个配置，就可以确保的说，一旦slave复制数据和ack延迟时间太长，就认为可能master宕机后损失的数据太多了，那么就拒绝写请求，这样就可以把master宕机时由于部分数据未同步到slave导致的数据丢失降低到可控范围内。 ### 2、集群脑裂导致数据丢失 脑裂：也就是说，某个master所在机器突然脱离了正常的网络，跟其它slave机器不能连接，但是实际上master还运行着。** **造成的问题：** 此时哨兵可能就会认为master宕机了，然后开始选举，将其它slave切换成master。这时候集群里就会有2个master，也就是所谓的脑裂。** 此时虽然某个slave被切换成了master，但是可能client还没来得及切换成新的master，还继续写向旧的master的数据可能就丢失了。 因此旧master再次恢复的时候，会被作为一个slave挂到新的master上去，自己的数据会被清空，重新从新的master复制数据。 **如何怎么解决？** min-slaves-to-write 1 min-slaves-max-lag 10 要求至少有1个slave，数据复制和同步的延迟不能超过10秒 如果说一旦所有的slave，数据复制和同步的延迟都超过了10秒钟，那么这个时候，master就不会再接收任何请求了 上面两个配置可以减少异步复制和脑裂导致的数据丢失 集群脑裂因为client还没来得及切换成新的master，还继续写向旧的master的数据可能就丢失了通过min-slaves-to-write 确保必须是有多少个从节点连接，并且延迟时间小于min-slaves-max-lag多少秒。 ### 当然对于客户端需要怎么做呢？ 对于client来讲，就需要做些处理，比如先将数据缓存到内存当中，然后过一段时间处理，或者连接失败，接收到错误切换新的master处理。

【功能模块】配置修改好各单节点的配置文件redis.conf【操作步骤&问题现象】1、启动节点时启动失败，一直停留在一个状态。【截图信息】【日志信息】（可选，上传日志内容或者附件）

【功能模块】【操作步骤&问题现象】1、顺利安装Docker与下载redis最新版本镜像后运行容器2、 docker run -d --name redis-chai -p 6379:6379 redis3、4、但是：5、查看工作路径时报错【截图信息】【日志信息】（可选，上传日志内容或者附件）

## 一、主从复制原理 主从复制过程大体可以分为3个阶段：连接建立阶段（即准备阶段）、数据同步阶段、命令传播阶段。 在从节点执行 slaveof 命令后，复制过程便开始运作，下面图示大概可以看到，从图中可以看出复制过程大致分为6个过程  主从配置之后的日志记录也可以看出这个流程（看主redis或从redis日志都可以）： ```shell vi` `/var/log/redis/redis``.log ```  **1）保存主节点（master）信息：** 执行 slaveof 后 Redis 会打印如下日志：  **2）从节点（slave）底层内部通过每秒运行的定时任务维护复制相关逻辑，当定时任务发现存在新的主节点后，会尝试与该节点建立网络连接** 从节点与主节点建立网络连接示例图：  从节点会建立一个 socket 套接字，从节点建立了一个端口为51234的套接字，专门用于接受主节点发送的复制命令。从节点连接成功后打印如下日志：  如果从节点无法建立连接，定时任务会无限重试直到连接成功或者手动执行 slaveof no one 取消复制（断开主从） 注：对一个从redis节点服务器执行命令 slaveof no one 将使得这个从redis节点服务器关闭复制功能，并从 从redis节点服务器转变回主redis节点服务器，原来同步所得的数据集不会被丢弃。 关于连接失败，可以在从节点执行 info replication 查看 master_link_down_since_seconds 指标，它会记录与主节点连接失败的系统时间。从节点连接主节点失败时也会每秒打印如下日志，方便发现问题： \# Error condition on socket for SYNC: {socket_error_reason} **3）发送 ping 命令** 连接建立成功后从节点发送 ping 请求进行首次通信，ping 请求主要目的如下：** ** ①、检测主从之间网络套接字是否可用。 ②、检测主节点当前是否可接受处理命令。 ③、如果发送 ping 命令后，从节点没有收到主节点的 pong 回复或者超时，比如网络超时或者主节点正在阻塞无法响应命令，从节点会断开复制连接，下次定时任务会发起重连。  从节点发送的 ping 命令成功返回，Redis 打印如下日志，并继续后续复制流程：  **4）权限验证。**如果主节点设置了 requirepass 参数，则需要密码验证，从节点必须配置 masterauth 参数保证与主节点相同的密码才能通过验证；如果验证失败复制将终止，从节点重新发起复制流程。 **5）同步数据集。**主从复制连接正常通信后，对于首次建立复制的场景，主节点会把持有的数据全部发送给从节点，这部分操作是耗时最长的步骤。 ** ** **6）命令持续复制。**当主节点把当前的数据同步给从节点后，便完成了复制的建立流程。接下来主节点会持续地把写命令发送给从节点，保证主从数据一致性。  ## 二、全量复制和部分复制 ### 2.1 相关概念 **全量复制**：用于初次复制或其它无法进行部分复制的情况，将主节点中的所有数据都发送给从节点，是一个非常重型的操作，当数据量较大时，会对主从节点和网络造成很大的开销 **部分复制**：用于处理在主从复制中因网络闪断等原因造成的数据丢失场景，当从节点再次连上主节点后，（如果条件允许，后面会说到具体那些条件需要被满足），主节点会补发丢失数据给从节点。因为补发的数据远远小于全量数据，可以有效避免全量复制的过高开销，需要注意的是，如果网络中断时间过长，造成主节点没有能够完整地保存中断期间执行的写命令，则无法进行部分复制，仍使用全量复制 **复制偏移量**：参与复制的主从节点都会维护自身复制偏移量。主节点（master）在处理完写入命令后，会把命令的字节长度做累加记录，统计信息在 info replication 中的 master_repl_offset 指标中： 提醒：记得进入redis客户端执行该命令 ```shell #在从节点的redis中执行以下命令``redis-cli``info replication ``` 截图如下：  从节点（slave）每秒钟上报自身的复制偏移量给主节点，因此主节点也会保存从节点的复制偏移量，统计指标如下： 再次提醒：记得进入redis客户端执行 info replication 命令 ```shell #在主节点的redis中执行以下命令``redis-cli``info replication ``` 截图如下：  从节点在接收到主节点发送的命令后，也会累加记录自身的偏移量。统计信息在 info replication 中的 slave_repl_offset 中 **复制积压缓冲区**：复制积压缓冲区是保存在主节点上的一个固定长度的队列，默认大小为1MB（可以调整大小，具体调整多大看各自需求业务来调整），当主节点有连接的从节点（slave）时被创建，这时主节点（master）响应写命令时，不但会把命令发送给从节点，还会写入复制积压缓冲区。 在命令传播阶段（也可以说是 同步数据阶段），主节点除了将写命令发送给从节点，还会发送一份给复制积压缓冲区，作为写命令的备份；除了存储写命令，复制积压缓冲区中还存储了其中的每个字节对应的复制偏移量(offset) 。由于复制积压缓冲区定长而且先进先出，所以它保存的是主节点最近执行的写命令；时间较早的写命令会被挤出缓冲区。  ### 2.2 Redis全量复制的过程如下图所示：  如上图所示： 1、Redis 内部会发出一个同步命令，刚开始是 Psync 命令，Psync ? -1表示要求 master 主机同步数据 2、主机会向从机发送 runid 和 offset，因为 slave 并没有对应的 offset，所以是全量复制 3、从机 slave 会保存 主机 master 的基本信息 save masterInfo 4、主节点收到全量复制的命令后，执行 bgsave 命令（异步执行），在后台生成RDB文件（快照文件），并使用一个缓冲区（称为复制缓冲区）记录从现在开始执行的所有写命令 5、主机发送 RDB 文件给从机 6、主机发送缓冲区数据到从机 7、刷新旧的数据，从节点在载入主节点的数据之前要先将老数据清除 8、加载 RDB 文件将数据库状态更新至主节点执行bgsave时的数据库状态和缓冲区数据的加载。 **全量复制开销，主要有以下几项：** 1、bgsave 时间 2、RDB 文件网络传输时间 3、从节点清空数据的时间 4、从节点加载 RDB 的时间 注：生成一个几GB 十几GB的RDB快照文件 大概是需要2-5分钟的样子 ### 2.3 部分复制（又称 增量复制） 部分复制是 Redis 2.8 以后出现的，之所以要加入部分复制，是因为全量复制会产生很多问题，比如像上面的时间开销大、无法隔离等问题， Redis 希望能够在 master 出现抖动（相当于断开连接）的时候，可以有一些机制将复制的损失降低到最低。 部分复制的过程如下图所示：  如上图所示： 1、如果网络抖动（连接断开 connection lost） 2、主机 master 还是会写 replbackbuffer（复制缓冲区） 3、从机 slave 会继续尝试连接主机 4、从机 slave 会把自己当前 runid 和偏移量传输给主机 master，并且执行 pysnc 命令同步 5、如果 master 发现从节点的偏移量是在缓冲区的范围内，就会返回 continue（在指的是继续复制的意思） 命令，如果从节点的偏移量不在缓冲区的范围之内，就不会进行部分复制操作了，这样就意味着它会执行全量复制，所以我们要尽量避免这种情况的发生。 6、同步了 offset 的部分数据，所以部分复制的基础就是偏移量 offset **正常情况下redis是如何决定是全量复制还是部分复制？** 从节点将offset（偏移量）发送给主节点后，主节点根据offset（偏移量）和缓冲区大小决定能否执行部分复制； 如果offset偏移量之后的数据，仍然都在复制积压缓冲区里，则执行部分复制； 如果offset偏移量之后的数据已不在复制积压缓冲区中（数据已被挤出），则执行全量复制； **缓冲区大小调节：** 由于缓冲区长度固定且有限，因此可以备份的写命令也有限，当主从节点offset的差距过大超过缓冲区长度时，将无法执行部分复制，只能执行全量复制。反过来说，为了提高网络中断时部分复制执行的概率，可以根据需要增大复制积压缓冲区的大小(通过配置repl-backlog-size)来设置；例如如果网络中断的平均时间是60s，而主节点平均每秒产生的写命令(特定协议格式)所占的字节数为100KB，则复制积压缓冲区的平均需求为6MB，保险起见，可以设置为12MB，来保证绝大多数断线情况都可以使用部分复制。 **服务器运行ID(runid)：** 每个Redis节点(无论主从)，在启动时都会自动生成一个随机ID(每次启动都不一样)，由40个随机的十六进制字符组成；runid用来唯一识别一个Redis节点。 通过info server命令，可以查看节点的runid：** **  主从节点初次复制时，主节点将自己的runid发送给从节点，从节点将这个runid保存起来；当断线重连时，从节点会将这个runid发送给主节点；主节点根据runid判断能否进行部分复制： 如果从节点保存的runid与主节点现在的runid相同，说明主从节点之前同步过，主节点会继续尝试使用部分复制（到底能不能部分复制还要看offset和复制积压缓冲区的情况） 如果从节点保存的runid与主节点现在的runid不同，说明从节点在断线前同步的Redis节点并不是当前的主节点，只能进行全量复制。 ## 三、主从复制的常用相关配置 **1、从redis节点配置：** ①、slaveof slave实例需要配置该项，指向master的（ip, port） ②、masterauth 如果master实例启用了密码保护，则该配置项需填master的启动密码，若master未启用密码，该配置项需要注释掉 ③、slave-serve-stale-data 指定slave与master连接中断时的动作。默认为yes，表明slave会继续应答来自client的请求，但这些数据可能已经过期（因为连接中断导致无法从master同步）。若配置为no，则slave除正常应答"INFO"和"SLAVEOF"命令外，其余来自客户端的请求命令均会得到"SYNC with master in progress"的应答，直到该slave与master的连接重建成功或该slave被提升为master ④、slave-read-only 指定slave是否只读，默认为yes。若配置为no，这表示slave是可写的，但写的内容在主从同步完成后会被删掉 ⑤、repl-disable-tcp-nodelay 指定向slave同步数据时，是否禁用socket的NO_DELAY选项。若配置为yes，则禁用NO_DELAY，则TCP协议栈会合并小包统一发送，这样可以减少主从节点间的包数量并节省带宽，但会增加数据同步到slave的时间。若配置为no，表明启用NO_DELAY，则TCP协议栈不会延迟小包的发送时机，这样数据同步的延时会减少，但需要更大的带宽。通常情况下，应该配置为no以降低同步延时，但在主从节点间网络负载已经很高的情况下，可以配置为yes 注：主从节点进行数据传输是基于tcp协议进行传输的 ⑥、slave-priority 指定slave的优先级。在不只1个slave存在的部署环境下，当master宕机时，Redis Sentinel会将priority值最小的slave提升为master。需要注意的是，若该配置项为0，则对应的slave永远不会被Redis Sentinel自动提升为master

## 一、开胃小菜（前期准备工作）： 1、准备一台鲲鹏Linux服务器并安装及启动docker 本文使用的是：CentOS 7.6 x64 2、准备一个Dockfile文件 用来构建redis镜像，该文件我会在下面提供 Dockerfile文件：[Dockerfile.zip](https://www.css3er.com/ueditor/php/upload/file/20190812/1565609627158640.zip) 自行下载解压后将文件上传到服务器。本文将该文件上传到服务器的路径：/root 3、启动系统的iptables，一般不需要更改该规则文件，如果下面步骤提示报相关网络错误，上网查一下即可解决 4、本系列博文使用的redis版本：3.2.12 ## 二、了解docker网络模式（为接下来的主从配置搭建做准备） Docker安装后，默认会创建下面三种网络类型 docker network ls 查看默认的网络  在启动容器时使用 --network bridge 指定网络类型 **bridge：桥接网络** 默认情况下启动的Docker容器，都是使用 bridge，Docker安装时创建的桥接网络，每次Docker容器重启时，会按照顺序获取对应的IP地址，这个就导致重启下，Docker的IP地址就变了（桥接网络模式也可以，就是通过端口映射访问到容器里面的redis，不过本文选择用下面的自定义网络模式） **none：无指定网络** 使用 --network=none ，docker 容器就不会分配局域网的IP **host： 主机网络** 使用 --network=host，此时，Docker 容器的网络会附属在主机上，两者是互通的。 例如，在容器中运行一个Web服务，监听8080端口，则主机的8080端口就会自动映射到容器中 ### 2.2、指定自定义网络，设置容器的固定IP（下面的Redis主从搭建选择用自定义网络） 因为默认的网络不能制定固定的地址，所以我们将创建自定义网络，并指定网段：172.10.0.0/16 并命名为mynetwork，这里选择了172.10.0.0/16网段，当然你也可以指定其他任意空闲的网段。将名字命名为mynetwork，你也可以换成其它的任意名字。具体创建指令如下： ```shell #创建一个mynetwork网络 并指定网段(子网)为：172.10.0.0/16(执行这一步，下面要用到该网络)``docker network create --subnet=172.10.0.0``/16` `mynetwork ```  /*******************伟大的分割线************************/ ## 三、开始搭建Redis主从模式 ### 概述： **主从复制说明，单一节点的redis(单台服务器的redis)容易面临的问题：** 比如：** ** 1、机器故障。我们部署到一台 Redis 服务器，当发生机器故障时，需要迁移到另外一台服务器并且要保证数据是同步的。而数据是最重要的，如果你不在乎，基本上也就不会使用 Redis 了。 2、容量瓶颈。当我们有需求需要扩容 Redis 内存时，从 16G 的内存升到 64G，单机肯定是满足不了。当然，你可以重新买个 128G 的新机器。 要实现分布式数据库的更大的存储容量和承受高并发访问量，我们会将原来集中式数据库的数据分别存储到其他多个网络节点上。 Redis 为了解决这个单一节点的问题，也会把数据复制多个副本部署到其他节点上进行复制，实现 Redis的高可用，实现对数据的冗余备份，从而保证数据和服务的高可用。 **什么是主从复制？** 主从复制，是指将一台Redis服务器的数据，复制到其他的Redis服务器。前者称为主节点(master)，后者称为从节点(slave),数据的复制是单向的，只能由主节点到从节点。** ** 默认情况下，每台Redis服务器都是主节点，且一个主节点可以有多个从节点(或没有从节点)，但一个从节点只能有一个主节点。也就是说一台redis从服务只能属于一台redis主服务。  **主从复制的作用：** 主从复制的作用主要包括： 1、数据冗余：主从复制实现了数据的热备份，是持久化之外的一种数据冗余方式。 2、故障恢复：当主节点出现问题时，可以由从节点提供服务，实现快速的故障恢复；实际上是一种服务的冗余。 3、负载均衡：在主从复制的基础上，配合读写分离，可以由主节点提供写服务，由从节点提供读服务（即写Redis数据时应用连接主节点，读Redis数据时应用连接从节点），分担服务器负载；尤其是在写少读多的场景下，通过多个从节点分担读负载，可以大大提高Redis服务器的并发量。 4、读写分离：可以用于实现读写分离，主库写、从库读，读写分离不仅可以提高服务器的负载能力，同时可根据需求的变化，改变从库的数量； 5、高可用基石（基础）：除了上述作用以外，主从复制还是哨兵和集群能够实施的基础，因此说主从复制是Redis高可用的基础。 ### 具体操作步骤如下： 3.1、进入到root目录下（具体换成你自己的路径），并使用Dockerfile文件构建redis镜像 ``` #进入到root目录下``cd` `/root/` `#使用Dockerfile文件构建一个redis镜像（注意后面有一个点）``docker build -t redis . ``` 可以使用docker images查看刚刚构建完成的redis镜像  3.2、使用此构建生成的redis镜像创建容器（创建2个redis容器 一主一从） ```shell #主redis，该容器命名为redis-master 使用mynetwork网络 端口映射6380对应容器内部端口的6379 指定容器固定IP:172.10.0.2 使用redis镜像来生成容器并在后台运行``docker run -itd --name redis-master --net mynetwork -p 6380:6379 --ip 172.10.0.2 redis` `#从redis，该容器命名为redis-slave 使用mynetwork网络 端口映射6381对应容器内部端口的6379 指定容器固定IP:172.10.0.3 使用redis镜像来生成容器并在后台运行``docker run -itd --name redis-slave --net mynetwork -p 6381:6379 --ip 172.10.0.3 redis ``` 可以使用docker ps 查看正在运行中的容器（查看我们上面创建的两个一主一从的redis容器）  可以使用docker network inspect mynetwork命令查看容器的ip地址等相关信息  3.3、分别进入主redis容器和从redis容器 修改redis.conf配置文件 ```shell #1、主redis容器redis.conf配置文件改动``使用docker ``exec``命令进入容器里面：docker ``exec` `-it 你生成的主redis容器ID 或 容器名字 ``/bin/bash` `本文中的对应命令为(容器``id``方式)：docker ``exec` `-it e4fbae61823b ``/bin/bash``本文中的对应命令为(容器名字方式)：docker ``exec` `-it redis-master ``/bin/bash` `修改redis.conf配置文件 ``vi` `/etc/redis``.conf``#将bind 127.0.0.1注释 或 改为bind 0.0.0.0 表示允许任何ip连接该redis服务，同时将protected-mode yes 改为 protected-mode no 表示关闭保护模式` `#2、从redis容器redis.conf配置文件改动(和主redis的改动一样，唯一区别就是进入容器的时候，要进入从redis容器也就是容器id或容器名字不一样了 不过还是演示下吧。。)``使用docker ``exec``命令进入容器里面：docker ``exec` `-it 你生成的从redis容器ID 或 容器名字 ``/bin/bash` `本文中的对应命令为(容器``id``方式)：docker ``exec` `-it a172d66c4e4d ``/bin/bash``本文中的对应命令为(容器名字方式)：docker ``exec` `-it redis-slave ``/bin/bash` `修改redis.conf配置文件 ``vi` `/etc/redis``.conf``#将bind 127.0.0.1注释 或 改为bind 0.0.0.0 表示允许任何ip连接该redis服务，同时将protected-mode yes 改为 protected-mode no 表示关闭保护模式 ``` redis.conf配置文件改动后的截图如下：   ### 3.4、启用主从模式（有三种方式）: （1）redis.conf配置文件（本文使用这种方式） 在从服务器的配置文件中加入：slaveof （2）启动命令 redis-server启动命令后加入 --slaveof （3）客户端命令 Redis服务器启动后，直接通过客户端执行命令：slaveof ，则该Redis实例成为从节点 通过 info replication 命令可以看到复制的一些参数信息 3.5、从redis容器中的redis.conf的redis配置文件加入配置信息，完成主从同步(别忘了 你要进入从redis容器里面) ```shell vi` `/etc/redis``.conf` `将 ``# slaveof 改为 slaveof 172.10.0.2 6379 也就是上面创建主redis容器是指定的主redis容器的固定IP地址。如果你和我的不一样，记得换成你自己设置的 ``` 从redis容器的里的redis.conf配置文件更改后的截图如下：  3.6、分别启动主从redis容器里面的redis服务(再次提醒。。别忘了进入主从redis容器里面启动) ```shell #主从redis容器里面启动redis服务都用以下这个命令即可``redis-server ``/etc/redis``.conf & ``#启动redis服务 &表示以后台守护进程方式启动(就是启动了后 在后台默默的服务) ``` 3.7、测试主从同步是否搭建成功(提醒啊。。别忘了进入主从redis容器里面进行操作。。真是操碎了心。。) 3.7.1、主redis容器操作： ``` redis-cli ``#进入redis客户端``set` `wzyl 123 ``#创建一个key ```  3.7.2、从redis容器操作： ```shell redis-cli ``#进入redis客户端` `使用get命令：get wzyl 或 使用keys命令：keys * ```  注：上面的keys * 命令不建议使用，除非你redis里面没有多少数据你可以使用keys * 命令，如果redis里面有很多数据 使用keys * 的话，会很慢。。会造成redis堵塞的。。 至此。。redis主从搭建成功。。往redis主节点写数据的时候，会自动同步到配置的redis从节点里（同步会有延时，因为需要用到网络，1秒钟之内会完成同步，大概也就几百毫秒的样子就可以了）实现了redis的读写分离。。这样写的操作可以写到主redis节点，读的操作可以读取从redis节点。。PS：一主一从搭建完了，一主多从应该也会了吧。。举一反三啊。。基于docker搭建redis主从都会了。。直接基于每一台服务器搭建redis主从就更简单了吧。。 **3.8、使用RedisDesktopManager工具来连接到docker里面的redis(我们就链接主redis这个容器里的redis吧)**  119.3.220.26是我服务器的公网IP，记得换成你自己服务器的公网IP，6380是我在上面创建主redis容器的时候指定的6380端口，实际上链接的时候会自动将6380映射成6379。。别直接上来就复制粘贴最后一点击测试发现各种error。。 **3.9、主从结构还有一种叫：树状主从结构** 树状主从结构：就是从节点它不但可以复制我们的主节点的数据，它同时也可以做为其它从节点的主节点，继续向下复制。说白了就是主节点有从节点，但是从节点还有从节点** ** 那这种树状主从结构是为了解决什么问题？ 主要就是避免主节点同步压力过大，造成性能干扰 树状主从结构示例图如下：  ## 四、后记、尾声 本文中没有设置redis密码，你可以在配置文件中进行设置，如果设置了验证密码，那么从redis节点中的配置文件中也要找到对应的密码位置进行修改才能主从同步成功，在生产环境中不建议将protected-mode yes 改为 no 并且会设置redis密码。。本文为了操作简单就没有设置密码这一步。 **总结redis主从、哨兵、集群的概念：** 【redis主从】： 是备份关系， 我们操作主库，数据也会同步到从库。 如果主库机器坏了，从库可以上。就好比你 D盘的片丢了，但是你移动硬盘里边备份有。 【redis哨兵】： 哨兵保证的是HA（高可用），保证特殊情况故障自动切换，哨兵盯着你的“redis主从集群”，如果主库死了，它会告诉你新的老大是谁。 哨兵：主要针对redis主从中的某一个单节点故障后，无法自动恢复的解决方案。（哨兵 保证redis主从的高可用） 【redis集群】： 集群保证的是高并发，因为多了一些兄弟帮忙一起扛。同时集群会导致数据的分散，整个redis集群会分成一堆数据槽，即不同的key会放到不不同的槽中。 集群主要针对单节点容量、高并发问题、线性可扩展性的解决方案。 集群：是为了解决redis主从复制中 单机内存上限和并发问题，假如你现在的服务器内存为256GB，当达到这个内存时redis就没办法再提供服务，同时数据量能达到这个地步写数据量也会很大，容易造成缓冲区溢出，造成从节点无限的进行全量复制导致主从无法正常工作。

最基础的概念，什么是幂等性？幂等性：提交多次的情况下，结果都一样。比如数据库查询，可称为天然幂等性，即查询多次结果都一样，无需人为去做幂等性操作。但是update table set value=value+1 where id=1,每次执行的结构都会发生变化，不是幂等。inter into table(id,name)values(1,‘name’),如id不是主键或者没有唯一索引，重复操作上面的业务，会插入多条数据，不具备幂等性；所以我们在什么情景下需要确保幂等性呢？用户多次点击保存按钮用户保存成功后，返回上一页再次保存微服务相互调用，由于网络原因，导致请求失败解决方案一、token机制：1、根据业务场景，判断哪些业务存在幂等性问题，在执行业务之前先获取token，将token缓存止redis中2、调用业务接口时，将token携带过去，一般放在请求头，作为Auth认证3、服务器判断token是否存在于redis中，存在表示第一次请求，然后删除token，继续执行业务4、如果不存在，则表示反复操作，不执行业务逻辑，直接返回重复标志！结束风险性：业务执行前删除还是后删除token？如果是执行后删除，在业务执行中，未删除token，用户又点了请求进来，那么则无法保障幂等性。如果是执行前删除，在分布式下，用户快速请求2次，这时2个请求同时到redis去获取token，对比成功，同时删除，同时执行业务，那么也无法保障幂等性。so：使用执行前删除，在分布式情况下，获取，对比，删除必须确保原子性，所以要加分布式锁。二、加锁1、数据库锁select * from table where … for update2、业务层面加分布式锁将获取、对比、删除作为一个原子性的操作加锁，处理完成后释放锁，确保串行操作。三、约束数据库唯一约束：通过主键、唯一索引，确保无法重复新增同一笔数据，这就能确保幂等性

最基础的概念，什么是幂等性？幂等性：提交多次的情况下，结果都一样。比如数据库查询，可称为天然幂等性，即查询多次结果都一样，无需人为去做幂等性操作。但是update table set value=value+1 where id=1,每次执行的结构都会发生变化，不是幂等。inter into table(id,name)values(1,‘name’),如id不是主键或者没有唯一索引，重复操作上面的业务，会插入多条数据，不具备幂等性；所以我们在什么情景下需要确保幂等性呢？用户多次点击保存按钮用户保存成功后，返回上一页再次保存微服务相互调用，由于网络原因，导致请求失败解决方案一、token机制：1、根据业务场景，判断哪些业务存在幂等性问题，在执行业务之前先获取token，将token缓存止redis中2、调用业务接口时，将token携带过去，一般放在请求头，作为Auth认证3、服务器判断token是否存在于redis中，存在表示第一次请求，然后删除token，继续执行业务4、如果不存在，则表示反复操作，不执行业务逻辑，直接返回重复标志！结束风险性：业务执行前删除还是后删除token？如果是执行后删除，在业务执行中，未删除token，用户又点了请求进来，那么则无法保障幂等性。如果是执行前删除，在分布式下，用户快速请求2次，这时2个请求同时到redis去获取token，对比成功，同时删除，同时执行业务，那么也无法保障幂等性。so：使用执行前删除，在分布式情况下，获取，对比，删除必须确保原子性，所以要加分布式锁。二、加锁1、数据库锁select * from table where … for update2、业务层面加分布式锁将获取、对比、删除作为一个原子性的操作加锁，处理完成后释放锁，确保串行操作。三、约束数据库唯一约束：通过主键、唯一索引，确保无法重复新增同一笔数据，这就能确保幂等性

最基础的概念，什么是幂等性？幂等性：提交多次的情况下，结果都一样。比如数据库查询，可称为天然幂等性，即查询多次结果都一样，无需人为去做幂等性操作。但是update table set value=value+1 where id=1,每次执行的结构都会发生变化，不是幂等。inter into table(id,name)values(1,‘name’),如id不是主键或者没有唯一索引，重复操作上面的业务，会插入多条数据，不具备幂等性；所以我们在什么情景下需要确保幂等性呢？用户多次点击保存按钮用户保存成功后，返回上一页再次保存微服务相互调用，由于网络原因，导致请求失败解决方案一、token机制：1、根据业务场景，判断哪些业务存在幂等性问题，在执行业务之前先获取token，将token缓存止redis中2、调用业务接口时，将token携带过去，一般放在请求头，作为Auth认证3、服务器判断token是否存在于redis中，存在表示第一次请求，然后删除token，继续执行业务4、如果不存在，则表示反复操作，不执行业务逻辑，直接返回重复标志！结束风险性：业务执行前删除还是后删除token？如果是执行后删除，在业务执行中，未删除token，用户又点了请求进来，那么则无法保障幂等性。如果是执行前删除，在分布式下，用户快速请求2次，这时2个请求同时到redis去获取token，对比成功，同时删除，同时执行业务，那么也无法保障幂等性。so：使用执行前删除，在分布式情况下，获取，对比，删除必须确保原子性，所以要加分布式锁。二、加锁1、数据库锁select * from table where … for update2、业务层面加分布式锁将获取、对比、删除作为一个原子性的操作加锁，处理完成后释放锁，确保串行操作。三、约束数据库唯一约束：通过主键、唯一索引，确保无法重复新增同一笔数据，这就能确保幂等性

本文分享自华为云社区《秒杀系统架构设计都有哪些关键点？》，作者：JavaEdge 。 # 0、目标 - 秒杀重复排队 重复排队：一个人抢购商品，若未支付，不准重复排队抢购 - 超卖问题 1个商品卖给多个人：1商品多订单 - 秒杀订单支付 秒杀支付：支付流程调整 - 超时未支付订单，库存回滚 1.RabbitMQ延时队列 2.利用延时队列实现支付订单的监听，根据订单支付状况进行订单数据库回滚 # 1、防止重复排队 用户每次抢单时，一旦排队，设置个自增值，让该值的初始值为1。 每次进入抢单时，对其递增，若值>1，则表明已排队，为禁止重复排队，直接对外抛异常信息xxx表示已在排队。 ## 1.1 后台排队记录 修改SeckillOrderServiceImpl#add方法，新增递增值判断是否已排队：  //递增，判断是否排队 Long userQueueCount = redisTemplate.boundHashOps("UserQueueCount").increment(username, 1); if(userQueueCount>1){ // 有重复抢单 throw new RuntimeException(String.valueOf(StatusCode.REPERROR)); } # 2、超卖问题 多人抢购同一商品时，多人同时判断是否有库存，若只剩一个，则都会判断有库存，此时会导致超卖，即一个商品被下了多个订单。 ## 2.1 思路分析  利用Redis list队列，给每件商品创建一个独立的商品个数队列，如：A商品有2个，A商品的ID为1001，则创建一个list，key=SeckillGoodsCountList_1001，往该队列中塞2次该商品ID。 每次给用户下单时，先从队列取数据： - 能取到数据 有库存 - 取不到 无库存 这就防止了超卖。 操作Redis大部分都是先查出数据查，在内存中修改，然后存入Redis。高并发下就有数据错乱问题，为控制数量准确，单独将商品数量整个自增键，自增键是线程安全的，无需担心并发问题。   ## 2.2 代码 每次将商品压入Redis时，创建一个商品队列。 修改SeckillGoodsPushTask，添加一个pushIds方法，用于将指定商品ID放入到指定数字： /*** * 将商品ID存入到数组中 * @param len:长度 * @param id :值 * @return */ public Long[] pushIds(int len,Long id){ Long[] ids = new Long[len]; for (int i = 0; i } return ids; } SeckillGoodsPushTask#loadGoodsPushRedis，添加队列操作：  ## 2.3 防止超卖 修改多线程下单方法，分别修改数量控制，以及售罄后用户抢单排队信息的清理：  # 3、订单支付  完成秒杀下订单后，进入支付页面，此时前端会每3s向后台发送一次请求，判断当前用户订单是否完成支付： 若完成支付，则清理排队信息，并修改订单状态。 ## 3.1 创建支付二维码 下单成功后，会跳转到支付选择页面，在支付选择页面要显示订单编号和订单金额，所以我们需要在下单的时候，将订单金额以及订单编号信息存储到用户查询对象。 选择微信支付后，会跳转到微信支付页面，微信支付页面会根据用户名查看用户秒杀订单，并根据用户秒杀订单的ID创建预支付信息并获取二维码信息，展示给用户看，此时页面每3秒查询一次支付状态，如果支付成功，需要修改订单状态信息。 ### 3.1.1 回显订单号、金额 下单后，进入支付选择页面，需显示订单号和订单金额，所以需要在用户下单后将该数据传入到pay.html页面，所以查询订单状态时，需要将订单号和金额封装到查询的信息中，修改查询订单装的方法加入他们。 修改SeckillOrderController#queryStatus：  测试：  ### 3.1.2 创建二维码 用户创建二维码，可以先查询用户的秒杀订单抢单信息，然后再发送请求到支付微服务中创建二维码，将订单编号以及订单对应的金额传递到支付微服务:/weixin/pay/create/native。 使用Postman测试效果如下：  ## 3.2 支付流程分析  1. 用户抢单，经过秒杀系统实现抢单，下单后会将向MQ发送一个延时消息，包含抢单信息 2. 秒杀系统同时启用延时消息监听，一旦监听到订单抢单信息，判断Redis缓存中是否存在订单信息，若存在，则回滚 3. 秒杀系统启动监听支付回调信息。若支付完成，则将订单持久化到MySQL，如果没完成，清理排队信息，回滚库存 4. 每次秒杀下单后调用支付系统，创建二维码。若用户支付成功，微信系统会将支付信息发给支付系统指定的回调地址，支付系统收到信息后，将信息发给MQ，第3个步骤就能监听到消息。 ## 3.3 支付回调更新 支付回调这一块代码已经实现了，但之前实现的是订单信息的回调数据发送给MQ，指定了对应队列，不过现在需要实现的是秒杀信息发送给指定队列，所以之前的代码那块需要动态指定队列。 ### 3.3.1 支付回调队列指定 创建支付二维码需要指定队列 回调地址回调时，获取支付二维码指定的队列，将支付信息发到指定队列 在微信支付统一下单API中，有个附加参数： `attach:附加数据,String(127)，在查询API和支付通知中原样返回，可作为自定义参数使用。` 可在创建二维码时，设置该参数以指定回调支付信息的对应队列，每次回调时，会获取该参数，然后将回调信息发到该参数对应的队列。 #### 3.3.1.1 改造支付方法 修改支付微服务的WeixinPayController#createNative：  修改支付微服务的WeixinPayService#createNative：  修改支付微服务的WeixinPayServiceImpl#createNative：  创建二维码时，传递如下参数 - username，用户名 可根据用户名，查询用户排队信息 - outtradeno商户订单号 下单必需 - money，支付金额 支付必需 - queue，队列名字 回调时，可知将支付信息发送到哪个队列 修改WeixinPayApplication，添加对应队列以及对应交换机绑定，代码如下： @SpringBootApplication @EnableEurekaClient @EnableFeignClients public class WeixinPayApplication { public static void main(String[] args) { SpringApplication.run(WeixinPayApplication.class,args); } @Autowired private Environment env; // 创建DirectExchange交换机 @Bean public DirectExchange basicExchange(){ return new DirectExchange(env.getProperty("mq.pay.exchange.order"), true,false); } // 创建队列 @Bean(name = "queueOrder") public Queue queueOrder(){ return new Queue(env.getProperty("mq.pay.queue.order"), true); } // 创建秒杀队列 @Bean(name = "queueSeckillOrder") public Queue queueSeckillOrder(){ return new Queue(env.getProperty("mq.pay.queue.seckillorder"), true); } // 队列绑定到交换机上 @Bean public Binding basicBindingOrder(){ return BindingBuilder .bind(queueOrder()) .to(basicExchange()) .with(env.getProperty("mq.pay.routing.orderkey")); } // 队列绑定到交换机 @Bean public Binding basicBindingSeckillOrder(){ return BindingBuilder .bind(queueSeckillOrder()) .to(basicExchange()) .with(env.getProperty("mq.pay.routing.seckillorderkey")); } } 修改application.yml，添加如下配置 #位置支付交换机和队列 mq: pay: exchange: order: exchange.order seckillorder: exchange.seckillorder queue: order: queue.order seckillorder: queue.seckillorder routing: key: queue.order seckillkey: queue.seckillorder #### 3.3.1.2 测试 创建二维码测试  以后每次支付，都需要带上对应的参数，包括前面的订单支付。 #### 3.3.1.3 改造支付回调方法 修改WeixinPayController#notifyUrl，获取自定义参数，并转成Map，获取queue地址，并将支付信息发送到绑定的queue：  ### 3.3.2 支付状态监听 支付状态通过回调地址发给MQ后，需要在秒杀系统中监听支付信息： - 支付成功，修改用户订单状态 - 支付失败，删除订单，回滚库存。 SeckillOrderPayMessageListener监听消息： @Component @RabbitListener(queues = "${mq.pay.queue.seckillorder}") public class SeckillOrderPayMessageListener { // 监听消费消息 @RabbitHandler public void consumeMessage(@Payload String message) { System.out.println(message); // 将消息转换成Map对象 Map resultMap = JSON.parseObject(message,Map.class); System.out.println("监听到的消息:"+resultMap); } } 修改SeckillApplication创建对应的队列以及绑定对应交换机。 @SpringBootApplication @EnableEurekaClient @EnableFeignClients @MapperScan(basePackages = {"com.changgou.seckill.dao"}) @EnableScheduling @EnableAsync public class SeckillApplication { public static void main(String[] args) { SpringApplication.run(SeckillApplication.class,args); } @Bean public IdWorker idWorker(){ return new IdWorker(1,1); } @Autowired private Environment env; /*** * 创建DirectExchange交换机 * @return */ @Bean public DirectExchange basicExchange(){ return new DirectExchange(env.getProperty("mq.pay.exchange.order"), true,false); } /*** * 创建队列 * @return */ @Bean(name = "queueOrder") public Queue queueOrder(){ return new Queue(env.getProperty("mq.pay.queue.order"), true); } /*** * 创建秒杀队列 * @return */ @Bean(name = "queueSeckillOrder") public Queue queueSeckillOrder(){ return new Queue(env.getProperty("mq.pay.queue.seckillorder"), true); } /**** * 队列绑定到交换机上 * @return */ @Bean public Binding basicBindingOrder(){ return BindingBuilder .bind(queueOrder()) .to(basicExchange()) .with(env.getProperty("mq.pay.routing.orderkey")); } /**** * 队列绑定到交换机上 * @return */ @Bean public Binding basicBindingSeckillOrder(){ return BindingBuilder .bind(queueSeckillOrder()) .to(basicExchange()) .with(env.getProperty("mq.pay.routing.seckillorderkey")); } } 添加配置： #位置支付交换机和队列 mq: pay: exchange: order: exchange.order seckillorder: exchange.seckillorder queue: order: queue.order seckillorder: queue.seckillorder routing: key: queue.order seckillkey: queue.seckillorder ### 3.3.3 修改订单状态 监听到支付信息后，根据支付信息判断，如果用户支付成功，则修改订单信息，并将订单入库，删除用户排队信息，如果用户支付失败，则删除订单信息，回滚库存，删除用户排队信息。 #### 3.3.3.1 业务层 修改SeckillOrderService，添加修改订单方法： /*** * 更新订单状态 */ @Override public void updatePayStatus(String out_trade_no, String transaction_id,String username) { //订单数据从Redis数据库查询出来 SeckillOrder seckillOrder = (SeckillOrder) redisTemplate.boundHashOps("SeckillOrder").get(username); //修改状态 seckillOrder.setStatus("1"); //支付时间 seckillOrder.setPayTime(new Date()); //同步到MySQL中 seckillOrderMapper.insertSelective(seckillOrder); //清空Redis缓存 redisTemplate.boundHashOps("SeckillOrder").delete(username); //清空用户排队数据 redisTemplate.boundHashOps("UserQueueCount").delete(username); //删除抢购状态信息 redisTemplate.boundHashOps("UserQueueStatus").delete(username); } #### 3.3.3.2 修改订单对接 修改微信支付状态监听的代码，当用户支付成功后，修改订单状态：  ### 3.3.4 删除订单回滚库存 如果用户支付失败，我们需要删除用户订单数据，并回滚库存。关闭订单： /*** * 关闭订单，回滚库存 */ @Override public void closeOrder(String username) { //将消息转换成SeckillStatus SeckillStatus seckillStatus = (SeckillStatus) redisTemplate.boundHashOps("UserQueueStatus").get(username); //获取Redis中订单信息 SeckillOrder seckillOrder = (SeckillOrder) redisTemplate.boundHashOps("SeckillOrder").get(username); //如果Redis中有订单信息，说明用户未支付 if(seckillStatus!=null && seckillOrder!=null){ //删除订单 redisTemplate.boundHashOps("SeckillOrder").delete(username); //回滚库存 //1)从Redis中获取该商品 SeckillGoods seckillGoods = (SeckillGoods) redisTemplate.boundHashOps("SeckillGoods_"+seckillStatus.getTime()).get(seckillStatus.getGoodsId()); //2)如果Redis中没有，则从数据库中加载 if(seckillGoods==null){ seckillGoods = seckillGoodsMapper.selectByPrimaryKey(seckillStatus.getGoodsId()); } //3)数量+1 (递增数量+1，队列数量+1) Long surplusCount = redisTemplate.boundHashOps("SeckillGoodsCount").increment(seckillStatus.getGoodsId(), 1); seckillGoods.setStockCount(surplusCount.intValue()); redisTemplate.boundListOps("SeckillGoodsCountList_" + seckillStatus.getGoodsId()).leftPush(seckillStatus.getGoodsId()); //4)数据同步到Redis中 redisTemplate.boundHashOps("SeckillGoods_"+seckillStatus.getTime()).put(seckillStatus.getGoodsId(),seckillGoods); //清理排队标示 redisTemplate.boundHashOps("UserQueueCount").delete(seckillStatus.getUsername()); //清理抢单标示 redisTemplate.boundHashOps("UserQueueStatus").delete(seckillStatus.getUsername()); } } #### 3.3.4.1 调用删除订单 SeckillOrderPayMessageListener，在用户支付失败后调用关闭订单： //支付失败,删除订单 seckillOrderService.closeOrder(attachMap.get("username")); 支付微服务 server: port: 9022 spring: application: name: pay main: allow-bean-definition-overriding: true rabbitmq: host: 127.0.0.1 #mq的服务器地址 username: guest #账号 password: guest #密码 eureka: client: service-url: defaultZone: http://127.0.0.1:6868/eureka instance: prefer-ip-address: true feign: hystrix: enabled: true #hystrix 配置 hystrix: command: default: execution: timeout: #如果enabled设置为false，则请求超时交给ribbon控制 enabled: true isolation: strategy: SEMAPHORE #微信支付信息配置 weixin: appid: wx8397f8696b538317 partner: 1473426802 partnerkey: T6m9iK73b0kn9g5v426MKfHQH7X8rKwb notifyurl: http://2cw4969042.wicp.vip:36446/weixin/pay/notify/url #位置支付交换机和队列 mq: pay: exchange: order: exchange.order queue: order: queue.order seckillorder: queue.seckillorder routing: orderkey: queue.order seckillorderkey: queue.seckillorder 秒杀微服务配置 server: port: 18084 spring: application: name: seckill datasource: driver-class-name: com.mysql.jdbc.Driver url: jdbc:mysql://127.0.0.1:3306/changgou_seckill?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC username: root password: itcast rabbitmq: host: 127.0.0.1 #mq的服务器地址 username: guest #账号 password: guest #密码 main: allow-bean-definition-overriding: true eureka: client: service-url: defaultZone: http://127.0.0.1:6868/eureka instance: prefer-ip-address: true feign: hystrix: enabled: true mybatis: configuration: map-underscore-to-camel-case: true mapper-locations: classpath:mapper/*Mapper.xml type-aliases-package: com.changgou.seckill.pojo #hystrix 配置 hystrix: command: default: execution: timeout: #如果enabled设置为false，则请求超时交给ribbon控制 enabled: true isolation: thread: timeoutInMilliseconds: 10000 strategy: SEMAPHORE #位置支付交换机和队列 mq: pay: exchange: order: exchange.order queue: order: queue.order seckillorder: queue.seckillorder routing: orderkey: queue.order seckillorderkey: queue.seckillorder # 4、库存回滚 ## 4.1 秒杀流程回顾  1.用户抢单，经过秒杀系统实现抢单，下单后会将向MQ发送一个延时队列消息，包含抢单信息，延时半小时后才能监听到 2.秒杀系统同时启用延时消息监听，一旦监听到订单抢单信息，判断Redis缓存中是否存在订单信息，如果存在，则回滚 3.秒杀系统还启动支付回调信息监听，如果支付完成，则将订单吃句话到MySQL，如果没完成，清理排队信息回滚库存 4.每次秒杀下单后调用支付系统，创建二维码，如果用户支付成功了，微信系统会将支付信息发送给支付系统指定的回调地址，支付系统收到信息后，将信息发送给MQ，第3个步骤就可以监听到消息了。 延时队列实现订单关闭回滚库存： 1.创建一个过期队列 Queue1 2.接收消息的队列 Queue2 3.中转交换机 4.监听Queue2 1)SeckillStatus->检查Redis中是否有订单信息 2)如果有订单信息，调用删除订单回滚库存->[需要先关闭微信支付] 3)如果关闭订单时，用于已支付，修改订单状态即可 4)如果关闭订单时，发生了别的错误，记录日志，人工处理 ## 4.2 关闭支付 用户如果半个小时没有支付，我们会关闭支付订单，但在关闭之前，需要先关闭微信支付，防止中途用户支付。 修改支付微服务的WeixinPayService，添加关闭支付方法，代码如下： /*** * 关闭支付 * @param orderId * @return */ Map closePay(Long orderId) throws Exception; 修改WeixinPayServiceImpl，实现关闭微信支付方法，代码如下： /*** * 关闭微信支付 * @param orderId * @return * @throws Exception */ @Override public Map closePay(Long orderId) throws Exception { //参数设置 Map paramMap = new HashMap(); paramMap.put("appid",appid); //应用ID paramMap.put("mch_id",partner); //商户编号 paramMap.put("nonce_str",WXPayUtil.generateNonceStr());//随机字符 paramMap.put("out_trade_no",String.valueOf(orderId)); //商家的唯一编号 //将Map数据转成XML字符 String xmlParam = WXPayUtil.generateSignedXml(paramMap,partnerkey); //确定url String url = "https://api.mch.weixin.qq.com/pay/closeorder"; //发送请求 HttpClient httpClient = new HttpClient(url); //https httpClient.setHttps(true); //提交参数 httpClient.setXmlParam(xmlParam); //提交 httpClient.post(); //获取返回数据 String content = httpClient.getContent(); //将返回数据解析成Map return WXPayUtil.xmlToMap(content); } ## 4.3 关闭订单回滚库存 ### 4.3.1 配置延时队列 在application.yml文件中引入队列信息配置，如下： #位置支付交换机和队列 mq: pay: exchange: order: exchange.order queue: order: queue.order seckillorder: queue.seckillorder seckillordertimer: queue.seckillordertimer seckillordertimerdelay: queue.seckillordertimerdelay routing: orderkey: queue.order seckillorderkey: queue.seckillorder 配置队列与交换机,在SeckillApplication中添加如下方法 /** * 到期数据队列 * @return */ @Bean public Queue seckillOrderTimerQueue() { return new Queue(env.getProperty("mq.pay.queue.seckillordertimer"), true); } /** * 超时数据队列 * @return */ @Bean public Queue delaySeckillOrderTimerQueue() { return QueueBuilder.durable(env.getProperty("mq.pay.queue.seckillordertimerdelay")) .withArgument("x-dead-letter-exchange", env.getProperty("mq.pay.exchange.order")) // 消息超时进入死信队列，绑定死信队列交换机 .withArgument("x-dead-letter-routing-key", env.getProperty("mq.pay.queue.seckillordertimer")) // 绑定指定的routing-key .build(); } /*** * 交换机与队列绑定 * @return */ @Bean public Binding basicBinding() { return BindingBuilder.bind(seckillOrderTimerQueue()) .to(basicExchange()) .with(env.getProperty("mq.pay.queue.seckillordertimer")); } ### 4.3.2 发送延时消息 修改MultiThreadingCreateOrder，添加如下方法： /*** * 发送延时消息到RabbitMQ中 * @param seckillStatus */ public void sendTimerMessage(SeckillStatus seckillStatus){ rabbitTemplate.convertAndSend(env.getProperty("mq.pay.queue.seckillordertimerdelay"), (Object) JSON.toJSONString(seckillStatus), new MessagePostProcessor() { @Override public Message postProcessMessage(Message message) throws AmqpException { message.getMessageProperties().setExpiration("10000"); return message; } }); } 在createOrder方法中调用上面方法，如下代码： //发送延时消息到MQ中 sendTimerMessage(seckillStatus);

1.String 字符串类型是redis中最基本的数据类型，一个key对应一个value。String类型是二进制安全的，意思是 redis 的 string 可以包含任何数据。如数字，字符串，jpg图片或者序列化的对象。使用：get 、 set 、 del 、 incr、 decr 等实战场景：1.缓存： 经典使用场景，把常用信息，字符串，图片或者视频等信息放到redis中，redis作为缓存层，mysql做持久化层，降低mysql的读写压力。2.计数器：redis是单线程模型，一个命令执行完才会执行下一个，同时数据可以一步落地到其他的数据源。3.session：常见方案spring session + redis实现session共享2.Hash （哈希）是一个Mapmap，指值本身又是一种键值对结构，如 value={{field1,value1},......fieldN,valueN}}实战场景：1.缓存： 能直观，相比string更节省空间，的维护缓存信息，如用户信息，视频信息等。3.链表 List 说白了就是链表（redis 使用双端链表实现的 List），是有序的，value可以重复，可以通过下标取出对应的value值，左右两边都能进行插入和删除数据。实战场景：1.timeline：例如微博的时间轴，有人发布微博，用lpush加入时间轴，展示新的列表信息。4.Set   集合集合类型也是用来保存多个字符串的元素，但和列表不同的是集合中  1. 不允许有重复的元素，2.集合中的元素是无序的，不能通过索引下标获取元素，3.支持集合间的操作，可以取多个集合取交集、并集、差集。实战场景;1.标签（tag）,给用户添加标签，或者用户给消息添加标签，这样有同一标签或者类似标签的可以给推荐关注的事或者关注的人。2.点赞，或点踩，收藏等，可以放到set中实现5.zset  有序集合有序集合和集合有着必然的联系，保留了集合不能有重复成员的特性，区别是，有序集合中的元素是可以排序的，它给每个元素设置一个分数，作为排序的依据。（有序集合中的元素不可以重复，但是score 分数 可以重复，就和一个班里的同学学号不能重复，但考试成绩可以相同）。实战场景：1.排行榜：有序集合经典使用场景。例如小说视频等网站需要对用户上传的小说视频做排行榜，榜单可以按照用户关注数，更新时间，字数等打分，做排行。

最基础的概念，什么是幂等性？幂等性：提交多次的情况下，结果都一样。比如数据库查询，可称为天然幂等性，即查询多次结果都一样，无需人为去做幂等性操作。但是update table set value=value+1 where id=1,每次执行的结构都会发生变化，不是幂等。inter into table(id,name)values(1,‘name’),如id不是主键或者没有唯一索引，重复操作上面的业务，会插入多条数据，不具备幂等性；所以我们在什么情景下需要确保幂等性呢？用户多次点击保存按钮用户保存成功后，返回上一页再次保存微服务相互调用，由于网络原因，导致请求失败解决方案一、token机制：1、根据业务场景，判断哪些业务存在幂等性问题，在执行业务之前先获取token，将token缓存止redis中2、调用业务接口时，将token携带过去，一般放在请求头，作为Auth认证3、服务器判断token是否存在于redis中，存在表示第一次请求，然后删除token，继续执行业务4、如果不存在，则表示反复操作，不执行业务逻辑，直接返回重复标志！结束风险性：业务执行前删除还是后删除token？如果是执行后删除，在业务执行中，未删除token，用户又点了请求进来，那么则无法保障幂等性。如果是执行前删除，在分布式下，用户快速请求2次，这时2个请求同时到redis去获取token，对比成功，同时删除，同时执行业务，那么也无法保障幂等性。so：使用执行前删除，在分布式情况下，获取，对比，删除必须确保原子性，所以要加分布式锁。二、加锁1、数据库锁select * from table where … for update2、业务层面加分布式锁将获取、对比、删除作为一个原子性的操作加锁，处理完成后释放锁，确保串行操作。三、约束数据库唯一约束：通过主键、唯一索引，确保无法重复新增同一笔数据，这就能确保幂等性

漏洞信息：漏洞名称：Redis 沙盒逃逸漏洞 漏洞编号：CVE-2022-0543 漏洞等级：高 披漏时间：2022年03月10日漏洞描述：Redis存在沙盒逃逸漏洞。攻击者利用该漏洞可能打破Redis中的Lua沙盒并且执行任意代码。漏洞状态：脆弱组件覆盖面利用门槛POC工具EXP工具中等高已公开已公开影响版本：   运行在 Debian、Ubuntu 或其他基于 Debian 的 Linux 发行版系统上的 Redis 服务。漏洞排查：用户需排查Redis Server版本是否为以下版本: Debian Redis：   5:5.0.14-1+deb10u2  5:6.0.16-1+deb11u2  5:6.0.16-2 Ubuntu Redis：   5:6.0.15-1ubuntu0.1  5:5.0.7-2ubuntu0.1漏洞加固：目前厂商已发布升级补丁以修复漏洞，补丁获取链接：   https://bugs.debian.org/cgi-bin/bugreport.cgi?bug=1005787

本文分享自华为云社区《[【高并发】什么是缓存穿透？击穿？雪崩？如何解决？](https://bbs.huaweicloud.com/blogs/336298?utm_source=csdn&utm_medium=bbs-ex&utm_campaign=other&utm_content=content)》，作者：冰 河。 说到Redis，往往更多的场景是被用作系统的缓存，说到缓存，尤其是分布式缓存系统，在实际高并发场景下，稍有不慎，就会造成缓存穿透、缓存击穿和缓存雪崩的问题。那什么是缓存穿透？什么是缓存击穿，又什么是缓存雪崩呢？它们是如何造成的？又该如何解决呢？今天，我们就一起来探讨这些问题。 # 缓存穿透 首先，我们来说说缓存穿透。什么是缓存穿透呢？缓存穿透问题在一定程度上与缓存命中率有关。如果我们的缓存设计的不合理，缓存的命中率非常低，那么，数据访问的绝大部分压力都会集中在后端数据库层面。 ## 什么是缓存穿透？ 如果在请求数据时，在缓存层和数据库层都没有找到符合条件的数据，也就是说，在缓存层和数据库层都没有命中数据，那么，这种情况就叫作缓存穿透。 我们可以使用下图来表示缓存穿透的现象。  造成缓存穿透的主要原因就是：查询某个Key对应的数据，Redis缓存中没有相应的数据，则直接到数据库中查询。数据库中也不存在要查询的数据，则数据库会返回空，而Redis也不会缓存这个空结果。这就造成每次通过这样的Key去查询数据都会直接到数据库中查询，Redis不会缓存空结果。这就造成了缓存穿透的问题。 ## 如何解决缓存穿透问题？ 既然我们知道了造成缓存穿透的主要原因就是缓存中不存在相应的数据，直接到数据库查询，数据库返回空结果，缓存中不存储空结果。 那我们就自然而然的想到了第一种解决方案：就是把空对象缓存起来。当第一次从数据库中查询出来的结果为空时，我们就将这个空对象加载到缓存，并设置合理的过期时间，这样，就能够在一定程度上保障后端数据库的安全。 第二种解决缓存穿透问题的解决方案：就是使用布隆过滤器，布隆过滤器可以针对大数据量的、有规律的键值进行处理。一条记录是不是存在，本质上是一个Bool值，只需要使用 1bit 就可以存储。我们可以使用布隆过滤器将这种表示是、否等操作，压缩到一个数据结构中。比如，我们最熟悉的用户性别这种数据，就非常适合使用布隆过滤器来处理。 # 缓存击穿 如果我们为缓存中的大部分数据设置了相同的过期时间，则到了某一时刻，缓存中的数据就会批量过期。 ## 什么是缓存击穿？ 如果缓存中的数据在某个时刻批量过期，导致大部分用户的请求都会直接落在数据库上，这种现象就叫作缓存击穿。 我们可以使用下图来表示缓存击穿的线程。  造成缓存击穿的主要原因就是：我们为缓存中的数据设置了过期时间。如果在某个时刻从数据库获取了大量的数据，并设置了相同的过期时间，这些缓存的数据就会在同一时刻失效，造成缓存击穿问题。 ## 如何解决缓存击穿问题？ 对于比较热点的数据，我们可以在缓存中设置这些数据永不过期；也可以在访问数据的时候，在缓存中更新这些数据的过期时间；如果是批量入库的缓存项，我们可以为这些缓存项分配比较合理的过期时间，避免同一时刻失效。 还有一种解决方案就是：使用分布式锁，保证对于每个Key同时只有一个线程去查询后端的服务，某个线程在查询后端服务的同时，其他线程没有获得分布式锁的权限，需要进行等待。不过在高并发场景下，这种解决方案对于分布式锁的访问压力比较大。 # 缓存雪崩 如果缓存系统出现故障，所有的并发流量就会直接到达数据库。 ## 什么是缓存雪崩？ 如果在某一时刻缓存集中失效，或者缓存系统出现故障，所有的并发流量就会直接到达数据库。数据存储层的调用量就会暴增，用不了多长时间，数据库就会被大流量压垮，这种级联式的服务故障，就叫作缓存雪崩。 我们可以用下图来表示缓存雪崩的现象。  造成缓存雪崩的主要原因就是缓存集中失效，或者缓存服务发生故障，瞬间的大并发流量压垮了数据库。 ## 如何解决缓存雪崩问题？ 解决缓存雪崩问题最常用的一种方案就是保证Redis的高可用，将Redis缓存部署成高可用集群（必要时候做成异地多活），可以有效的防止缓存雪崩问题的发生。 为了缓解大并发流量，我们也可以使用限流降级的方式防止缓存雪崩。例如，在缓存失效后，通过加锁或者使用队列来控制读数据库写缓存的线程数量。具体点就是设置某些Key只允许一个线程查询数据和写缓存，其他线程等待。则能够有效的缓解大并发流量对数据库打来的巨大冲击。 另外，我们也可以通过数据预热的方式将可能大量访问的数据加载到缓存，在即将发生大并发访问的时候，提前手动触发加载不同的数据到缓存中，并为数据设置不同的过期时间，让缓存失效的时间点尽量均匀，不至于在同一时刻全部失效。