AI 摘要 文章介绍将Netty RPC客户端接收服务端返回结果的方式，从阻塞的AttributeMap改为CompletableFuture，实现异步非阻塞调用，代码更清晰，使用更简单，并建议限制未处理请求Map容量防OOM。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结12 （优化）使用CompletableFuture优化接受服务提供端返回结果使用 AttributeMap 接受服务端返回结果最开始的时候是通过 AttributeMap 绑定到Channel上实现的，相关代码如下：NettyClientTransport.java（用来发送 RpcRequest 请求）NettyClientHandler.java （自定义客户端 ChannelHandler 来处理服务端发过来的数据）这种是实现的缺点是不清晰，而且你每次都要调用 channel.closeFuture().sync(); 阻塞来手动等待请求返回。使用 CompletableFuture 进行优化我使用 CompletableFuture 包装返回结果，对代码进行了重构，重要部分的代码如下：NettyClientTransport.java（用来发送 RpcRequest 请求）NettyClientHandler.java （自定义客户端 ChannelHandler 来处理服务端发过来的数据）UnprocessedRequests.java 存放了未处理的请求（建议限制 map 容器大小，避免未处理请求过多 OOM现在，在你只需要通过下面的方式就能成功接收到客户端返回的结果：CompletableFuture<RpcResponse> completableFuture = (CompletableFuture<RpcResponse>) clientTransport.sendRpcRequest(rpcRequest); rpcResponse = completableFuture.get();
AI 摘要 文章系统梳理 Redis 高频面试考点：内存高速读写、丰富数据类型及底层实现、单线程+IO多路复用、6.0多线程IO、AOF/RDB/混合持久化、主从+哨兵+Cluster高可用、惰性+定期过期删除、8种内存淘汰、缓存雪崩/击穿/穿透及一致性策略、分布式锁与Redlock、大Key与异步删除、管道与事务、延迟队列等，并给出场景与代码示例。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结Redis 常见面试题大家好，我是樱舞。认识 Redis什么是 Redis？我们直接看 Redis 官方是怎么介绍自己的。Redis 官方的介绍原版是英文的，我翻译成了中文后截图的，所以有些文字读起来会比较拗口，没关系，我会把里面比较重要的特性抽出来讲一下。Redis 是一种基于内存的数据库，对数据的读写操作都是在内存中完成，因此读写速度非常快，常用于缓存，消息队列、分布式锁等场景。Redis 提供了多种数据类型来支持不同的业务场景，比如 String(字符串)、Hash(哈希)、List (列表)、Set(集合)、Zset(有序集合)、Bitmaps（位图）、HyperLogLog（基数统计）、GEO（地理信息）、Stream（流），并且对数据类型的操作都是原子性的，因为执行命令由单线程负责的，不存在并发竞争的问题。除此之外，Redis 还支持事务、持久化、Lua 脚本、多种集群方案（主从复制模式、哨兵模式、切片集群模式）、发布/订阅模式，内存淘汰机制、过期删除机制等等。Redis 和 Memcached 有什么区别？很多人都说用 Redis 作为缓存，但是 Memcached 也是基于内存的数据库，为什么不选择它作为缓存呢？要解答这个问题，我们就要弄清楚 Redis 和 Memcached 的区别。Redis 与 Memcached 共同点：都是基于内存的数据库，一般都用来当做缓存使用。都有过期策略。两者的性能都非常高。Redis 与 Memcached 区别：Redis 支持的数据类型更丰富（String、Hash、List、Set、ZSet），而 Memcached 只支持最简单的 key-value 数据类型；Redis 支持数据的持久化，可以将内存中的数据保持在磁盘中，重启的时候可以再次加载进行使用，而 Memcached 没有持久化功能，数据全部存在内存之中，Memcached 重启或者挂掉后，数据就没了；Redis 原生支持集群模式，Memcached 没有原生的集群模式，需要依靠客户端来实现往集群中分片写入数据；Redis 支持发布订阅模型、Lua 脚本、事务等功能，而 Memcached 不支持；为什么用 Redis 作为 MySQL 的缓存？主要是因为 Redis 具备「高性能」和「高并发」两种特性。1、Redis 具备高性能假如用户第一次访问 MySQL 中的某些数据。这个过程会比较慢，因为是从硬盘上读取的。将该用户访问的数据缓存在 Redis 中，这样下一次再访问这些数据的时候就可以直接从缓存中获取了，操作 Redis 缓存就是直接操作内存，所以速度相当快。如果 MySQL 中的对应数据改变了之后，同步改变 Redis 缓存中相应的数据即可，不过这里会有 Redis 和 MySQL 双写一致性的问题，后面我们会提到。2、Redis 具备高并发单台设备的 Redis 的 QPS（Query Per Second，每秒钟处理完请求的次数）是 MySQL 的 10 倍，Redis 单机的 QPS 能轻松破 10w，而 MySQL 单机的 QPS 很难破 1w。所以，直接访问 Redis 能够承受的请求是远远大于直接访问 MySQL 的，所以我们可以考虑把数据库中的部分数据转移到缓存中去，这样用户的一部分请求会直接到缓存这里而不用经过数据库。Redis 数据结构Redis 数据类型以及使用场景分别是什么？Redis 提供了丰富的数据类型，常见的有五种数据类型：String（字符串），Hash（哈希），List（列表），Set（集合）、Zset（有序集合）。随着 Redis 版本的更新，后面又支持了四种数据类型： BitMap（2.2 版新增）、HyperLogLog（2.8 版新增）、GEO（3.2 版新增）、Stream（5.0 版新增）。Redis 五种数据类型的应用场景：String 类型的应用场景：缓存对象、常规计数、分布式锁、共享 session 信息等。List 类型的应用场景：消息队列（但是有两个问题：1. 生产者需要自行实现全局唯一 ID；2. 不能以消费组形式消费数据）等。Hash 类型：缓存对象、购物车等。Set 类型：聚合计算（并集、交集、差集）场景，比如点赞、共同关注、抽奖活动等。Zset 类型：排序场景，比如排行榜、电话和姓名排序等。Redis 后续版本又支持四种数据类型，它们的应用场景如下：BitMap（2.2 版新增）：二值状态统计的场景，比如签到、判断用户登陆状态、连续签到用户总数等；HyperLogLog（2.8 版新增）：海量数据基数统计的场景，比如百万级网页 UV 计数等；GEO（3.2 版新增）：存储地理位置信息的场景，比如滴滴叫车；Stream（5.0 版新增）：消息队列，相比于基于 List 类型实现的消息队列，有这两个特有的特性：自动生成全局唯一消息 ID，支持以消费组形式消费数据。::: tip想深入了解这 9 种数据类型，可以看这篇：2 万字 + 20 张图 ｜ 细说 Redis 常见数据类型和应用场景:::五种常见的 Redis 数据类型是怎么实现？我画了一张 Redis 数据类型和底层数据结构的对应关图，左边是 Redis 3.0 版本的，也就是《Redis 设计与实现》这本书讲解的版本，现在看还是有点过时了，右边是现在 Redis 7.0 版本的。String 类型内部实现String 类型的底层的数据结构实现主要是 SDS（简单动态字符串）。SDS 和我们认识的 C 字符串不太一样，之所以没有使用 C 语言的字符串表示，因为 SDS 相比于 C 的原生字符串：SDS 不仅可以保存文本数据，还可以保存二进制数据。因为 SDS 使用 len 属性的值而不是空字符来判断字符串是否结束，并且 SDS 的所有 API 都会以处理二进制的方式来处理 SDS 存放在 buf[] 数组里的数据。所以 SDS 不光能存放文本数据，而且能保存图片、音频、视频、压缩文件这样的二进制数据。SDS 获取字符串长度的时间复杂度是 O(1)。因为 C 语言的字符串并不记录自身长度，所以获取长度的复杂度为 O(n)；而 SDS 结构里用 len 属性记录了字符串长度，所以复杂度为 O(1)。Redis 的 SDS API 是安全的，拼接字符串不会造成缓冲区溢出。因为 SDS 在拼接字符串之前会检查 SDS 空间是否满足要求，如果空间不够会自动扩容，所以不会导致缓冲区溢出的问题。List 类型内部实现List 类型的底层数据结构是由双向链表或压缩列表实现的：如果列表的元素个数小于 512 个（默认值，可由 list-max-ziplist-entries 配置），列表每个元素的值都小于 64 字节（默认值，可由 list-max-ziplist-value 配置），Redis 会使用压缩列表作为 List 类型的底层数据结构；如果列表的元素不满足上面的条件，Redis 会使用双向链表作为 List 类型的底层数据结构；但是在 Redis 3.2 版本之后，List 数据类型底层数据结构就只由 quicklist 实现了，替代了双向链表和压缩列表。Hash 类型内部实现Hash 类型的底层数据结构是由压缩列表或哈希表实现的：如果哈希类型元素个数小于 512 个（默认值，可由 hash-max-ziplist-entries 配置），所有值小于 64 字节（默认值，可由 hash-max-ziplist-value 配置）的话，Redis 会使用压缩列表作为 Hash 类型的底层数据结构；如果哈希类型元素不满足上面条件，Redis 会使用哈希表作为 Hash 类型的底层数据结构。在 Redis 7.0 中，压缩列表数据结构已经废弃了，交由 listpack 数据结构来实现了。Set 类型内部实现Set 类型的底层数据结构是由哈希表或整数集合实现的：如果集合中的元素都是整数且元素个数小于 512（默认值，set-maxintset-entries 配置）个，Redis 会使用整数集合作为 Set 类型的底层数据结构；如果集合中的元素不满足上面条件，则 Redis 使用哈希表作为 Set 类型的底层数据结构。ZSet 类型内部实现Zset 类型的底层数据结构是由压缩列表或跳表实现的：如果有序集合的元素个数小于 128 个，并且每个元素的值小于 64 字节时，Redis 会使用压缩列表作为 Zset 类型的底层数据结构；如果有序集合的元素不满足上面的条件，Redis 会使用跳表作为 Zset 类型的底层数据结构；在 Redis 7.0 中，压缩列表数据结构已经废弃了，交由 listpack 数据结构来实现了。::: tip想深入了解这 9 种数据结构，可以看这篇：2 万字 + 40 张图 ｜ 细说 Redis 数据结构:::Redis 线程模型Redis 是单线程吗？Redis 单线程指的是「接收客户端请求->解析请求 ->进行数据读写等操作->发送数据给客户端」这个过程是由一个线程（主线程）来完成的，这也是我们常说 Redis 是单线程的原因。但是，Redis 程序并不是单线程的，Redis 在启动的时候，是会启动后台线程（BIO）的：Redis 在 2.6 版本，会启动 2 个后台线程，分别处理关闭文件、AOF 刷盘这两个任务；Redis 在 4.0 版本之后，新增了一个新的后台线程，用来异步释放 Redis 内存，也就是 lazyfree 线程。例如执行 unlink key / flushdb async / flushall async 等命令，会把这些删除操作交给后台线程来执行，好处是不会导致 Redis 主线程卡顿。因此，当我们要删除一个大 key 的时候，不要使用 del 命令删除，因为 del 是在主线程处理的，这样会导致 Redis 主线程卡顿，因此我们应该使用 unlink 命令来异步删除大 key。之所以 Redis 为「关闭文件、AOF 刷盘、释放内存」这些任务创建单独的线程来处理，是因为这些任务的操作都是很耗时的，如果把这些任务都放在主线程来处理，那么 Redis 主线程就很容易发生阻塞，这样就无法处理后续的请求了。后台线程相当于一个消费者，生产者把耗时任务丢到任务队列中，消费者（BIO）不停轮询这个队列，拿出任务就去执行对应的方法即可。关闭文件、AOF 刷盘、释放内存这三个任务都有各自的任务队列：BIO_CLOSE_FILE，关闭文件任务队列：当队列有任务后，后台线程会调用 close(fd) ，将文件关闭；BIO_AOF_FSYNC，AOF 刷盘任务队列：当 AOF 日志配置成 everysec 选项后，主线程会把 AOF 写日志操作封装成一个任务，也放到队列中。当发现队列有任务后，后台线程会调用 fsync(fd)，将 AOF 文件刷盘，BIO_LAZY_FREE，lazy free 任务队列：当队列有任务后，后台线程会 free(obj) 释放对象 / free(dict) 删除数据库所有对象 / free(skiplist) 释放跳表对象；Redis 单线程模式是怎样的？Redis 6.0 版本之前的单线模式如下图：图中的蓝色部分是一个事件循环，是由主线程负责的，可以看到网络 I/O 和命令处理都是单线程。Redis 初始化的时候，会做下面这几件事情：首先，调用 epoll_create() 创建一个 epoll 对象和调用 socket() 一个服务端 socket然后，调用 bind() 绑定端口和调用 listen() 监听该 socket；然后，将调用 epoll_ctl() 将 listen socket 加入到 epoll，同时注册「连接事件」处理函数。初始化完后，主线程就进入到一个事件循环函数，主要会做以下事情：首先，先调用处理发送队列函数，看是发送队列里是否有任务，如果有发送任务，则通过 write 函数将客户端发送缓存区里的数据发送出去，如果这一轮数据没有发送完，就会注册写事件处理函数，等待 epoll_wait 发现可写后再处理。接着，调用 epoll_wait 函数等待事件的到来：如果是连接事件到来，则会调用连接事件处理函数，该函数会做这些事情：调用 accpet 获取已连接的 socket -> 调用 epoll_ctl 将已连接的 socket 加入到 epoll -> 注册「读事件」处理函数；如果是读事件到来，则会调用读事件处理函数，该函数会做这些事情：调用 read 获取客户端发送的数据 -> 解析命令 -> 处理命令 -> 将客户端对象添加到发送队列 -> 将执行结果写到发送缓存区等待发送；如果是写事件到来，则会调用写事件处理函数，该函数会做这些事情：通过 write 函数将客户端发送缓存区里的数据发送出去，如果这一轮数据没有发送完，就会继续注册写事件处理函数，等待 epoll_wait 发现可写后再处理。以上就是 Redis 单线模式的工作方式，如果你想看源码解析，可以参考这一篇：为什么单线程的 Redis 如何做到每秒数万 QPS？Redis 采用单线程为什么还这么快？官方使用基准测试的结果是，单线程的 Redis 吞吐量可以达到 10W/每秒，如下图所示：之所以 Redis 采用单线程（网络 I/O 和执行命令）那么快，有如下几个原因：Redis 的大部分操作都在内存中完成，并且采用了高效的数据结构，因此 Redis 瓶颈可能是机器的内存或者网络带宽，而并非 CPU，既然 CPU 不是瓶颈，那么自然就采用单线程的解决方案了；Redis 采用单线程模型可以避免了多线程之间的竞争，省去了多线程切换带来的时间和性能上的开销，而且也不会导致死锁问题。Redis 采用了 I/O 多路复用机制处理大量的客户端 Socket 请求，IO 多路复用机制是指一个线程处理多个 IO 流，就是我们经常听到的 select/epoll 机制。简单来说，在 Redis 只运行单线程的情况下，该机制允许内核中，同时存在多个监听 Socket 和已连接 Socket。内核会一直监听这些 Socket 上的连接请求或数据请求。一旦有请求到达，就会交给 Redis 线程处理，这就实现了一个 Redis 线程处理多个 IO 流的效果。Redis 6.0 之前为什么使用单线程？我们都知道单线程的程序是无法利用服务器的多核 CPU 的，那么早期 Redis 版本的主要工作（网络 I/O 和执行命令）为什么还要使用单线程呢？我们不妨先看一下 Redis 官方给出的FAQ。核心意思是：CPU 并不是制约 Redis 性能表现的瓶颈所在，更多情况下是受到内存大小和网络 I/O 的限制，所以 Redis 核心网络模型使用单线程并没有什么问题，如果你想要使用服务的多核 CPU，可以在一台服务器上启动多个节点或者采用分片集群的方式。除了上面的官方回答，选择单线程的原因也有下面的考虑。使用了单线程后，可维护性高，多线程模型虽然在某些方面表现优异，但是它却引入了程序执行顺序的不确定性，带来了并发读写的一系列问题，增加了系统复杂度、同时可能存在线程切换、甚至加锁解锁、死锁造成的性能损耗。Redis 6.0 之后为什么引入了多线程？虽然 Redis 的主要工作（网络 I/O 和执行命令）一直是单线程模型，但是在 Redis 6.0 版本之后，也采用了多个 I/O 线程来处理网络请求，这是因为随着网络硬件的性能提升，Redis 的性能瓶颈有时会出现在网络 I/O 的处理上。所以为了提高网络 I/O 的并行度，Redis 6.0 对于网络 I/O 采用多线程来处理。但是对于命令的执行，Redis 仍然使用单线程来处理，所以大家不要误解 Redis 有多线程同时执行命令。Redis 官方表示，Redis 6.0 版本引入的多线程 I/O 特性对性能提升至少是一倍以上。Redis 6.0 版本支持的 I/O 多线程特性，默认情况下 I/O 多线程只针对发送响应数据（write client socket），并不会以多线程的方式处理读请求（read client socket）。要想开启多线程处理客户端读请求，就需要把 Redis.conf 配置文件中的 io-threads-do-reads 配置项设为 yes。//读请求也使用 io 多线程 io-threads-do-reads yes 同时，Redis.conf 配置文件中提供了 IO 多线程个数的配置项。// io-threads N，表示启用 N-1 个 I/O 多线程（主线程也算一个 I/O 线程） io-threads 4 关于线程数的设置，官方的建议是如果为 4 核的 CPU，建议线程数设置为 2 或 3，如果为 8 核 CPU 建议线程数设置为 6，线程数一定要小于机器核数，线程数并不是越大越好。因此，Redis 6.0 版本之后，Redis 在启动的时候，默认情况下会创建 6 个线程：Redis-server：Redis 的主线程，主要负责执行命令；bio_close_file、bio_aof_fsync、bio_lazy_free：三个后台线程，分别异步处理关闭文件任务、AOF 刷盘任务、释放内存任务；io_thd_1、io_thd_2、io_thd_3：三个 I/O 线程，io-threads 默认是 4，所以会启动 3（4-1）个 I/O 多线程，用来分担 Redis 网络 I/O 的压力。Redis 持久化Redis 如何实现数据不丢失？Redis 的读写操作都是在内存中，所以 Redis 性能才会高，但是当 Redis 重启后，内存中的数据就会丢失，那为了保证内存中的数据不会丢失，Redis 实现了数据持久化的机制，这个机制会把数据存储到磁盘，这样在 Redis 重启就能够从磁盘中恢复原有的数据。Redis 共有三种数据持久化的方式：AOF 日志：每执行一条写操作命令，就把该命令以追加的方式写入到一个文件里；RDB 快照：将某一时刻的内存数据，以二进制的方式写入磁盘；混合持久化方式：Redis 4.0 新增的方式，集成了 AOF 和 RBD 的优点；AOF 日志是如何实现的？Redis 在执行完一条写操作命令后，就会把该命令以追加的方式写入到一个文件里，然后 Redis 重启时，会读取该文件记录的命令，然后逐一执行命令的方式来进行数据恢复。我这里以「_set name xiaolin_」命令作为例子，Redis 执行了这条命令后，记录在 AOF 日志里的内容如下图：我这里给大家解释下。「*3」表示当前命令有三个部分，每部分都是以「$+数字」开头，后面紧跟着具体的命令、键或值。然后，这里的「数字」表示这部分中的命令、键或值一共有多少字节。例如，「$3 set」表示这部分有 3 个字节，也就是「set」命令这个字符串的长度。为什么先执行命令，再把数据写入日志呢？Reids 是先执行写操作命令后，才将该命令记录到 AOF 日志里的，这么做其实有两个好处。避免额外的检查开销：因为如果先将写操作命令记录到 AOF 日志里，再执行该命令的话，如果当前的命令语法有问题，那么如果不进行命令语法检查，该错误的命令记录到 AOF 日志里后，Redis 在使用日志恢复数据时，就可能会出错。不会阻塞当前写操作命令的执行：因为当写操作命令执行成功后，才会将命令记录到 AOF 日志。当然，这样做也会带来风险：数据可能会丢失： 执行写操作命令和记录日志是两个过程，那当 Redis 在还没来得及将命令写入到硬盘时，服务器发生宕机了，这个数据就会有丢失的风险。可能阻塞其他操作： 由于写操作命令执行成功后才记录到 AOF 日志，所以不会阻塞当前命令的执行，但因为 AOF 日志也是在主线程中执行，所以当 Redis 把日志文件写入磁盘的时候，还是会阻塞后续的操作无法执行。AOF 写回策略有几种？先来看看，Redis 写入 AOF 日志的过程，如下图：具体说说：Redis 执行完写操作命令后，会将命令追加到 server.aof_buf 缓冲区；然后通过 write() 系统调用，将 aof_buf 缓冲区的数据写入到 AOF 文件，此时数据并没有写入到硬盘，而是拷贝到了内核缓冲区 page cache，等待内核将数据写入硬盘；具体内核缓冲区的数据什么时候写入到硬盘，由内核决定。Redis 提供了 3 种写回硬盘的策略，控制的就是上面说的第三步的过程。在 Redis.conf 配置文件中的 appendfsync 配置项可以有以下 3 种参数可填：Always，这个单词的意思是「总是」，所以它的意思是每次写操作命令执行完后，同步将 AOF 日志数据写回硬盘；Everysec，这个单词的意思是「每秒」，所以它的意思是每次写操作命令执行完后，先将命令写入到 AOF 文件的内核缓冲区，然后每隔一秒将缓冲区里的内容写回到硬盘；No，意味着不由 Redis 控制写回硬盘的时机，转交给操作系统控制写回的时机，也就是每次写操作命令执行完后，先将命令写入到 AOF 文件的内核缓冲区，再由操作系统决定何时将缓冲区内容写回硬盘。我也把这 3 个写回策略的优缺点总结成了一张表格：AOF 日志过大，会触发什么机制？AOF 日志是一个文件，随着执行的写操作命令越来越多，文件的大小会越来越大。如果当 AOF 日志文件过大就会带来性能问题，比如重启 Redis 后，需要读 AOF 文件的内容以恢复数据，如果文件过大，整个恢复的过程就会很慢。所以，Redis 为了避免 AOF 文件越写越大，提供了 AOF 重写机制，当 AOF 文件的大小超过所设定的阈值后，Redis 就会启用 AOF 重写机制，来压缩 AOF 文件。AOF 重写机制是在重写时，读取当前数据库中的所有键值对，然后将每一个键值对用一条命令记录到「新的 AOF 文件」，等到全部记录完后，就将新的 AOF 文件替换掉现有的 AOF 文件。举个例子，在没有使用重写机制前，假设前后执行了「_set name xiaolin_」和「_set name xiaolincoding_」这两个命令的话，就会将这两个命令记录到 AOF 文件。但是在使用重写机制后，就会读取 name 最新的 value（键值对） ，然后用一条「set name xiaolincoding」命令记录到新的 AOF 文件，之前的第一个命令就没有必要记录了，因为它属于「历史」命令，没有作用了。这样一来，一个键值对在重写日志中只用一条命令就行了。重写工作完成后，就会将新的 AOF 文件覆盖现有的 AOF 文件，这就相当于压缩了 AOF 文件，使得 AOF 文件体积变小了。重写 AOF 日志的过程是怎样的？Redis 的重写 AOF 过程是由后台子进程 bgrewriteaof 来完成的，这么做可以达到两个好处：子进程进行 AOF 重写期间，主进程可以继续处理命令请求，从而避免阻塞主进程；子进程带有主进程的数据副本，这里使用子进程而不是线程，因为如果是使用线程，多线程之间会共享内存，那么在修改共享内存数据的时候，需要通过加锁来保证数据的安全，而这样就会降低性能。而使用子进程，创建子进程时，父子进程是共享内存数据的，不过这个共享的内存只能以只读的方式，而当父子进程任意一方修改了该共享内存，就会发生「写时复制」，于是父子进程就有了独立的数据副本，就不用加锁来保证数据安全。触发重写机制后，主进程就会创建重写 AOF 的子进程，此时父子进程共享物理内存，重写子进程只会对这个内存进行只读，重写 AOF 子进程会读取数据库里的所有数据，并逐一把内存数据的键值对转换成一条命令，再将命令记录到重写日志（新的 AOF 文件）。但是重写过程中，主进程依然可以正常处理命令，那问题来了，重写 AOF 日志过程中，如果主进程修改了已经存在 key-value，那么会发生写时复制，此时这个 key-value 数据在子进程的内存数据就跟主进程的内存数据不一致了，这时要怎么办呢？为了解决这种数据不一致问题，Redis 设置了一个 AOF 重写缓冲区，这个缓冲区在创建 bgrewriteaof 子进程之后开始使用。在重写 AOF 期间，当 Redis 执行完一个写命令之后，它会同时将这个写命令写入到「AOF 缓冲区」和「AOF 重写缓冲区」。也就是说，在 bgrewriteaof 子进程执行 AOF 重写期间，主进程需要执行以下三个工作：执行客户端发来的命令；将执行后的写命令追加到「AOF 缓冲区」；将执行后的写命令追加到「AOF 重写缓冲区」；当子进程完成 AOF 重写工作（_扫描数据库中所有数据，逐一把内存数据的键值对转换成一条命令，再将命令记录到重写日志_）后，会向主进程发送一条信号，信号是进程间通讯的一种方式，且是异步的。主进程收到该信号后，会调用一个信号处理函数，该函数主要做以下工作：将 AOF 重写缓冲区中的所有内容追加到新的 AOF 的文件中，使得新旧两个 AOF 文件所保存的数据库状态一致；新的 AOF 的文件进行改名，覆盖现有的 AOF 文件。信号函数执行完后，主进程就可以继续像往常一样处理命令了。::: tipAOF 日志的内容就暂时提这些，想更详细了解 AOF 日志的工作原理，可以详细看这篇：AOF 持久化是怎么实现的:::RDB 快照是如何实现的呢？因为 AOF 日志记录的是操作命令，不是实际的数据，所以用 AOF 方法做故障恢复时，需要全量把日志都执行一遍，一旦 AOF 日志非常多，势必会造成 Redis 的恢复操作缓慢。为了解决这个问题，Redis 增加了 RDB 快照。所谓的快照，就是记录某一个瞬间东西，比如当我们给风景拍照时，那一个瞬间的画面和信息就记录到了一张照片。所以，RDB 快照就是记录某一个瞬间的内存数据，记录的是实际数据，而 AOF 文件记录的是命令操作的日志，而不是实际的数据。因此在 Redis 恢复数据时，RDB 恢复数据的效率会比 AOF 高些，因为直接将 RDB 文件读入内存就可以，不需要像 AOF 那样还需要额外执行操作命令的步骤才能恢复数据。RDB 做快照时会阻塞线程吗？Redis 提供了两个命令来生成 RDB 文件，分别是 save 和 bgsave，他们的区别就在于是否在「主线程」里执行：执行了 save 命令，就会在主线程生成 RDB 文件，由于和执行操作命令在同一个线程，所以如果写入 RDB 文件的时间太长，会阻塞主线程；执行了 bgsave 命令，会创建一个子进程来生成 RDB 文件，这样可以避免主线程的阻塞；Redis 还可以通过配置文件的选项来实现每隔一段时间自动执行一次 bgsave 命令，默认会提供以下配置：save 900 1 save 300 10 save 60 10000别看选项名叫 save，实际上执行的是 bgsave 命令，也就是会创建子进程来生成 RDB 快照文件。只要满足上面条件的任意一个，就会执行 bgsave，它们的意思分别是：900 秒之内，对数据库进行了至少 1 次修改；300 秒之内，对数据库进行了至少 10 次修改；60 秒之内，对数据库进行了至少 10000 次修改。这里提一点，Redis 的快照是全量快照，也就是说每次执行快照，都是把内存中的「所有数据」都记录到磁盘中。所以执行快照是一个比较重的操作，如果频率太频繁，可能会对 Redis 性能产生影响。如果频率太低，服务器故障时，丢失的数据会更多。RDB 在执行快照的时候，数据能修改吗？可以的，执行 bgsave 过程中，Redis 依然可以继续处理操作命令的，也就是数据是能被修改的，关键的技术就在于写时复制技术（Copy-On-Write, COW）。执行 bgsave 命令的时候，会通过 fork() 创建子进程，此时子进程和父进程是共享同一片内存数据的，因为创建子进程的时候，会复制父进程的页表，但是页表指向的物理内存还是一个，此时如果主线程执行读操作，则主线程和 bgsave 子进程互相不影响。如果主线程执行写操作，则被修改的数据会复制一份副本，然后 bgsave 子进程会把该副本数据写入 RDB 文件，在这个过程中，主线程仍然可以直接修改原来的数据。::: tipRDB 快照的内容就暂时提这些，想更详细了解 RDB 快照的工作原理，可以详细看这篇：RDB 快照是怎么实现的？:::为什么会有混合持久化？RDB 优点是数据恢复速度快，但是快照的频率不好把握。频率太低，丢失的数据就会比较多，频率太高，就会影响性能。AOF 优点是丢失数据少，但是数据恢复不快。为了集成了两者的优点，Redis 4.0 提出了混合使用 AOF 日志和内存快照，也叫混合持久化，既保证了 Redis 重启速度，又降低数据丢失风险。混合持久化工作在 AOF 日志重写过程，当开启了混合持久化时，在 AOF 重写日志时，fork 出来的重写子进程会先将与主线程共享的内存数据以 RDB 方式写入到 AOF 文件，然后主线程处理的操作命令会被记录在重写缓冲区里，重写缓冲区里的增量命令会以 AOF 方式写入到 AOF 文件，写入完成后通知主进程将新的含有 RDB 格式和 AOF 格式的 AOF 文件替换旧的的 AOF 文件。也就是说，使用了混合持久化，AOF 文件的前半部分是 RDB 格式的全量数据，后半部分是 AOF 格式的增量数据。这样的好处在于，重启 Redis 加载数据的时候，由于前半部分是 RDB 内容，这样加载的时候速度会很快。加载完 RDB 的内容后，才会加载后半部分的 AOF 内容，这里的内容是 Redis 后台子进程重写 AOF 期间，主线程处理的操作命令，可以使得数据更少的丢失。混合持久化优点：混合持久化结合了 RDB 和 AOF 持久化的优点，开头为 RDB 的格式，使得 Redis 可以更快的启动，同时结合 AOF 的优点，有减低了大量数据丢失的风险。混合持久化缺点：AOF 文件中添加了 RDB 格式的内容，使得 AOF 文件的可读性变得很差；兼容性差，如果开启混合持久化，那么此混合持久化 AOF 文件，就不能用在 Redis 4.0 之前版本了。Redis 集群Redis 如何实现服务高可用？要想设计一个高可用的 Redis 服务，一定要从 Redis 的多服务节点来考虑，比如 Redis 的主从复制、哨兵模式、切片集群。主从复制主从复制是 Redis 高可用服务的最基础的保证，实现方案就是将从前的一台 Redis 服务器，同步数据到多台从 Redis 服务器上，即一主多从的模式，且主从服务器之间采用的是「读写分离」的方式。主服务器可以进行读写操作，当发生写操作时自动将写操作同步给从服务器，而从服务器一般是只读，并接受主服务器同步过来写操作命令，然后执行这条命令。也就是说，所有的数据修改只在主服务器上进行，然后将最新的数据同步给从服务器，这样就使得主从服务器的数据是一致的。注意，主从服务器之间的命令复制是异步进行的。具体来说，在主从服务器命令传播阶段，主服务器收到新的写命令后，会发送给从服务器。但是，主服务器并不会等到从服务器实际执行完命令后，再把结果返回给客户端，而是主服务器自己在本地执行完命令后，就会向客户端返回结果了。如果从服务器还没有执行主服务器同步过来的命令，主从服务器间的数据就不一致了。所以，无法实现强一致性保证（主从数据时时刻刻保持一致），数据不一致是难以避免的。::: tip想更详细了解 Redis 主从复制的工作原理，可以详细看这篇：主从复制是怎么实现的？:::哨兵模式在使用 Redis 主从服务的时候，会有一个问题，就是当 Redis 的主从服务器出现故障宕机时，需要手动进行恢复。为了解决这个问题，Redis 增加了哨兵模式（Redis Sentinel），因为哨兵模式做到了可以监控主从服务器，并且提供主从节点故障转移的功能。::: tip想更详细了解 Redis 哨兵的工作原理，可以详细看这篇：哨兵是怎么实现的？:::切片集群模式当 Redis 缓存数据量大到一台服务器无法缓存时，就需要使用 Redis 切片集群（Redis Cluster）方案，它将数据分布在不同的服务器上，以此来降低系统对单主节点的依赖，从而提高 Redis 服务的读写性能。Redis Cluster 方案采用哈希槽（Hash Slot），来处理数据和节点之间的映射关系。在 Redis Cluster 方案中，一个切片集群共有 16384 个哈希槽，这些哈希槽类似于数据分区，每个键值对都会根据它的 key，被映射到一个哈希槽中，具体执行过程分为两大步：根据键值对的 key，按照 CRC16 算法计算一个 16 bit 的值。再用 16bit 值对 16384 取模，得到 0~16383 范围内的模数，每个模数代表一个相应编号的哈希槽。接下来的问题就是，这些哈希槽怎么被映射到具体的 Redis 节点上的呢？有两种方案：平均分配： 在使用 cluster create 命令创建 Redis 集群时，Redis 会自动把所有哈希槽平均分布到集群节点上。比如集群中有 9 个节点，则每个节点上槽的个数为 16384/9 个。手动分配： 可以使用 cluster meet 命令手动建立节点间的连接，组成集群，再使用 cluster addslots 命令，指定每个节点上的哈希槽个数。为了方便你的理解，我通过一张图来解释数据、哈希槽，以及节点三者的映射分布关系。上图中的切片集群一共有 3 个节点，假设有 4 个哈希槽（Slot 0～Slot 3）时，我们就可以通过命令手动分配哈希槽，比如节点 1 保存哈希槽 0 和 1，节点 2 保存哈希槽 2 和 3。redis-cli -h 192.168.1.10 –p 6379 cluster addslots 0,1 redis-cli -h 192.168.1.11 –p 6379 cluster addslots 2,3然后在集群运行的过程中，key1 和 key2 计算完 CRC16 值后，对哈希槽总个数 5 进行取模，再根据各自的模数结果，就可以被映射到对应的节点 1 和节点 3 上了。需要注意的是，在手动分配哈希槽时，需要把 16384 个槽都分配完，否则 Redis 集群无法正常工作。集群脑裂导致数据丢失怎么办？什么是脑裂？先来理解集群的脑裂现象，这就好比一个人有两个大脑，那么到底受谁控制呢？那么在 Redis 中，集群脑裂产生数据丢失的现象是怎样的呢？在 Redis 主从架构中，部署方式一般是「一主多从」，主节点提供写操作，从节点提供读操作。如果主节点的网络突然发生了问题，它与所有的从节点都失联了，但是此时的主节点和客户端的网络是正常的，这个客户端并不知道 Redis 内部已经出现了问题，还在照样的向这个失联的主节点写数据（过程 A），此时这些数据被旧主节点缓存到了缓冲区里，因为主从节点之间的网络问题，这些数据都是无法同步给从节点的。这时，哨兵也发现主节点失联了，它就认为主节点挂了（但实际上主节点正常运行，只是网络出问题了），于是哨兵就会在「从节点」中选举出一个 leeder 作为主节点，这时集群就有两个主节点了 —— 脑裂出现了。然后，网络突然好了，哨兵因为之前已经选举出一个新主节点了，它就会把旧主节点降级为从节点（A），然后从节点（A）会向新主节点请求数据同步，因为第一次同步是全量同步的方式，此时的从节点（A）会清空掉自己本地的数据，然后再做全量同步。所以，之前客户端在过程 A 写入的数据就会丢失了，也就是集群产生脑裂数据丢失的问题。总结一句话就是：由于网络问题，集群节点之间失去联系。主从数据不同步；重新平衡选举，产生两个主服务。等网络恢复，旧主节点会降级为从节点，再与新主节点进行同步复制的时候，由于会从节点会清空自己的缓冲区，所以导致之前客户端写入的数据丢失了。解决方案当主节点发现从节点下线或者通信超时的总数量小于阈值时，那么禁止主节点进行写数据，直接把错误返回给客户端。在 Redis 的配置文件中有两个参数我们可以设置：min-slaves-to-write x，主节点必须要有至少 x 个从节点连接，如果小于这个数，主节点会禁止写数据。min-slaves-max-lag x，主从数据复制和同步的延迟不能超过 x 秒，如果超过，主节点会禁止写数据。我们可以把 min-slaves-to-write 和 min-slaves-max-lag 这两个配置项搭配起来使用，分别给它们设置一定的阈值，假设为 N 和 T。这两个配置项组合后的要求是，主库连接的从库中至少有 N 个从库，和主库进行数据复制时的 ACK 消息延迟不能超过 T 秒，否则，主库就不会再接收客户端的写请求了。即使原主库是假故障，它在假故障期间也无法响应哨兵心跳，也不能和从库进行同步，自然也就无法和从库进行 ACK 确认了。这样一来，min-slaves-to-write 和 min-slaves-max-lag 的组合要求就无法得到满足，原主库就会被限制接收客户端写请求，客户端也就不能在原主库中写入新数据了。等到新主库上线时，就只有新主库能接收和处理客户端请求，此时，新写的数据会被直接写到新主库中。而原主库会被哨兵降为从库，即使它的数据被清空了，也不会有新数据丢失。再来举个例子。假设我们将 min-slaves-to-write 设置为 1，把 min-slaves-max-lag 设置为 12s，把哨兵的 down-after-milliseconds 设置为 10s，主库因为某些原因卡住了 15s，导致哨兵判断主库客观下线，开始进行主从切换。同时，因为原主库卡住了 15s，没有一个从库能和原主库在 12s 内进行数据复制，原主库也无法接收客户端请求了。这样一来，主从切换完成后，也只有新主库能接收请求，不会发生脑裂，也就不会发生数据丢失的问题了。Redis 过期删除与内存淘汰Redis 使用的过期删除策略是什么？Redis 是可以对 key 设置过期时间的，因此需要有相应的机制将已过期的键值对删除，而做这个工作的就是过期键值删除策略。每当我们对一个 key 设置了过期时间时，Redis 会把该 key 带上过期时间存储到一个过期字典（expires dict）中，也就是说「过期字典」保存了数据库中所有 key 的过期时间。当我们查询一个 key 时，Redis 首先检查该 key 是否存在于过期字典中：如果不在，则正常读取键值；如果存在，则会获取该 key 的过期时间，然后与当前系统时间进行比对，如果比系统时间大，那就没有过期，否则判定该 key 已过期。Redis 使用的过期删除策略是「惰性删除 + 定期删除」这两种策略配和使用。什么是惰性删除策略？惰性删除策略的做法是，不主动删除过期键，每次从数据库访问 key 时，都检测 key 是否过期，如果过期则删除该 key。惰性删除的流程图如下：惰性删除策略的优点：因为每次访问时，才会检查 key 是否过期，所以此策略只会使用很少的系统资源，因此，惰性删除策略对 CPU 时间最友好。惰性删除策略的缺点：如果一个 key 已经过期，而这个 key 又仍然保留在数据库中，那么只要这个过期 key 一直没有被访问，它所占用的内存就不会释放，造成了一定的内存空间浪费。所以，惰性删除策略对内存不友好。什么是定期删除策略？定期删除策略的做法是，每隔一段时间「随机」从数据库中取出一定数量的 key 进行检查，并删除其中的过期 key。Redis 的定期删除的流程：从过期字典中随机抽取 20 个 key；检查这 20 个 key 是否过期，并删除已过期的 key；如果本轮检查的已过期 key 的数量，超过 5 个（20/4），也就是「已过期 key 的数量」占比「随机抽取 key 的数量」大于 25%，则继续重复步骤 1；如果已过期的 key 比例小于 25%，则停止继续删除过期 key，然后等待下一轮再检查。可以看到，定期删除是一个循环的流程。那 Redis 为了保证定期删除不会出现循环过度，导致线程卡死现象，为此增加了定期删除循环流程的时间上限，默认不会超过 25ms。定期删除的流程如下：定期删除策略的优点：通过限制删除操作执行的时长和频率，来减少删除操作对 CPU 的影响，同时也能删除一部分过期的数据减少了过期键对空间的无效占用。定期删除策略的缺点：难以确定删除操作执行的时长和频率。如果执行的太频繁，就会对 CPU 不友好；如果执行的太少，那又和惰性删除一样了，过期 key 占用的内存不会及时得到释放。可以看到，惰性删除策略和定期删除策略都有各自的优点，所以 Redis 选择「惰性删除 + 定期删除」这两种策略配和使用，以求在合理使用 CPU 时间和避免内存浪费之间取得平衡。::: tipRedis 的过期删除的内容就暂时提这些，想更详细了解的，可以详细看这篇：Redis 过期删除策略和内存淘汰策略有什么区别？:::Redis 持久化时，对过期键会如何处理的？Redis 持久化文件有两种格式：RDB（Redis Database）和 AOF（Append Only File），下面我们分别来看过期键在这两种格式中的呈现状态。RDB 文件分为两个阶段，RDB 文件生成阶段和加载阶段。RDB 文件生成阶段：从内存状态持久化成 RDB（文件）的时候，会对 key 进行过期检查，过期的键「不会」被保存到新的 RDB 文件中，因此 Redis 中的过期键不会对生成新 RDB 文件产生任何影响。RDB 加载阶段：RDB 加载阶段时，要看服务器是主服务器还是从服务器，分别对应以下两种情况：如果 Redis 是「主服务器」运行模式的话，在载入 RDB 文件时，程序会对文件中保存的键进行检查，过期键「不会」被载入到数据库中。所以过期键不会对载入 RDB 文件的主服务器造成影响；如果 Redis 是「从服务器」运行模式的话，在载入 RDB 文件时，不论键是否过期都会被载入到数据库中。但由于主从服务器在进行数据同步时，从服务器的数据会被清空。所以一般来说，过期键对载入 RDB 文件的从服务器也不会造成影响。AOF 文件分为两个阶段，AOF 文件写入阶段和 AOF 重写阶段。AOF 文件写入阶段：当 Redis 以 AOF 模式持久化时，如果数据库某个过期键还没被删除，那么 AOF 文件会保留此过期键，当此过期键被删除后，Redis 会向 AOF 文件追加一条 DEL 命令来显式地删除该键值。AOF 重写阶段：执行 AOF 重写时，会对 Redis 中的键值对进行检查，已过期的键不会被保存到重写后的 AOF 文件中，因此不会对 AOF 重写造成任何影响。Redis 主从模式中，对过期键会如何处理？当 Redis 运行在主从模式下时，从库不会进行过期扫描，从库对过期的处理是被动的。也就是即使从库中的 key 过期了，如果有客户端访问从库时，依然可以得到 key 对应的值，像未过期的键值对一样返回。从库的过期键处理依靠主服务器控制，主库在 key 到期时，会在 AOF 文件里增加一条 del 指令，同步到所有的从库，从库通过执行这条 del 指令来删除过期的 key。Redis 内存满了，会发生什么？在 Redis 的运行内存达到了某个阀值，就会触发内存淘汰机制，这个阀值就是我们设置的最大运行内存，此值在 Redis 的配置文件中可以找到，配置项为 maxmemory。Redis 内存淘汰策略有哪些？Redis 内存淘汰策略共有八种，这八种策略大体分为「不进行数据淘汰」和「进行数据淘汰」两类策略。1、不进行数据淘汰的策略noeviction（Redis3.0 之后，默认的内存淘汰策略） ：它表示当运行内存超过最大设置内存时，不淘汰任何数据，而是不再提供服务，直接返回错误。2、进行数据淘汰的策略针对「进行数据淘汰」这一类策略，又可以细分为「在设置了过期时间的数据中进行淘汰」和「在所有数据范围内进行淘汰」这两类策略。在设置了过期时间的数据中进行淘汰：volatile-random：随机淘汰设置了过期时间的任意键值；volatile-ttl：优先淘汰更早过期的键值。volatile-lru（Redis3.0 之前，默认的内存淘汰策略）：淘汰所有设置了过期时间的键值中，最久未使用的键值；volatile-lfu（Redis 4.0 后新增的内存淘汰策略）：淘汰所有设置了过期时间的键值中，最少使用的键值；在所有数据范围内进行淘汰：allkeys-random：随机淘汰任意键值;allkeys-lru：淘汰整个键值中最久未使用的键值；allkeys-lfu（Redis 4.0 后新增的内存淘汰策略）：淘汰整个键值中最少使用的键值。LRU 算法和 LFU 算法有什么区别？什么是 LRU 算法？LRU 全称是 Least Recently Used 翻译为最近最少使用，会选择淘汰最近最少使用的数据。传统 LRU 算法的实现是基于「链表」结构，链表中的元素按照操作顺序从前往后排列，最新操作的键会被移动到表头，当需要内存淘汰时，只需要删除链表尾部的元素即可，因为链表尾部的元素就代表最久未被使用的元素。Redis 并没有使用这样的方式实现 LRU 算法，因为传统的 LRU 算法存在两个问题：需要用链表管理所有的缓存数据，这会带来额外的空间开销；当有数据被访问时，需要在链表上把该数据移动到头端，如果有大量数据被访问，就会带来很多链表移动操作，会很耗时，进而会降低 Redis 缓存性能。Redis 是如何实现 LRU 算法的？Redis 实现的是一种近似 LRU 算法，目的是为了更好的节约内存，它的实现方式是在 Redis 的对象结构体中添加一个额外的字段，用于记录此数据的最后一次访问时间。当 Redis 进行内存淘汰时，会使用随机采样的方式来淘汰数据，它是随机取 5 个值（此值可配置），然后淘汰最久没有使用的那个。Redis 实现的 LRU 算法的优点：不用为所有的数据维护一个大链表，节省了空间占用；不用在每次数据访问时都移动链表项，提升了缓存的性能；但是 LRU 算法有一个问题，无法解决缓存污染问题，比如应用一次读取了大量的数据，而这些数据只会被读取这一次，那么这些数据会留存在 Redis 缓存中很长一段时间，造成缓存污染。因此，在 Redis 4.0 之后引入了 LFU 算法来解决这个问题。什么是 LFU 算法？LFU 全称是 Least Frequently Used 翻译为最近最不常用的，LFU 算法是根据数据访问次数来淘汰数据的，它的核心思想是“如果数据过去被访问多次，那么将来被访问的频率也更高”。所以，LFU 算法会记录每个数据的访问次数。当一个数据被再次访问时，就会增加该数据的访问次数。这样就解决了偶尔被访问一次之后，数据留存在缓存中很长一段时间的问题，相比于 LRU 算法也更合理一些。Redis 是如何实现 LFU 算法的？LFU 算法相比于 LRU 算法的实现，多记录了「数据的访问频次」的信息。Redis 对象的结构如下：typedef struct redisObject { ... // 24 bits，用于记录对象的访问信息 unsigned lru:24; ... } robj;Redis 对象头中的 lru 字段，在 LRU 算法下和 LFU 算法下使用方式并不相同。在 LRU 算法中，Redis 对象头的 24 bits 的 lru 字段是用来记录 key 的访问时间戳，因此在 LRU 模式下，Redis 可以根据对象头中的 lru 字段记录的值，来比较最后一次 key 的访问时间长，从而淘汰最久未被使用的 key。在 LFU 算法中，Redis 对象头的 24 bits 的 lru 字段被分成两段来存储，高 16bit 存储 ldt(Last Decrement Time)，用来记录 key 的访问时间戳；低 8bit 存储 logc(Logistic Counter)，用来记录 key 的访问频次。::: tipRedis 的内存淘汰的内容就暂时提这些，想更详细了解的，可以详细看这篇：Redis 过期删除策略和内存淘汰策略有什么区别？:::Redis 缓存设计如何避免缓存雪崩、缓存击穿、缓存穿透？如何避免缓存雪崩？通常我们为了保证缓存中的数据与数据库中的数据一致性，会给 Redis 里的数据设置过期时间，当缓存数据过期后，用户访问的数据如果不在缓存里，业务系统需要重新生成缓存，因此就会访问数据库，并将数据更新到 Redis 里，这样后续请求都可以直接命中缓存。那么，当大量缓存数据在同一时间过期（失效）时，如果此时有大量的用户请求，都无法在 Redis 中处理，于是全部请求都直接访问数据库，从而导致数据库的压力骤增，严重的会造成数据库宕机，从而形成一系列连锁反应，造成整个系统崩溃，这就是缓存雪崩的问题。对于缓存雪崩问题，我们可以采用两种方案解决。将缓存失效时间随机打散： 我们可以在原有的失效时间基础上增加一个随机值（比如 1 到 10 分钟）这样每个缓存的过期时间都不重复了，也就降低了缓存集体失效的概率。设置缓存不过期： 我们可以通过后台服务来更新缓存数据，从而避免因为缓存失效造成的缓存雪崩，也可以在一定程度上避免缓存并发问题。如何避免缓存击穿？我们的业务通常会有几个数据会被频繁地访问，比如秒杀活动，这类被频繁访问的数据被称为热点数据。如果缓存中的某个热点数据过期了，此时大量的请求访问了该热点数据，就无法从缓存中读取，直接访问数据库，数据库很容易就被高并发的请求冲垮，这就是缓存击穿的问题。可以发现缓存击穿跟缓存雪崩很相似，你可以认为缓存击穿是缓存雪崩的一个子集。应对缓存击穿可以采取前面说到两种方案：互斥锁方案（Redis 中使用 setNX 方法设置一个状态位，表示这是一种锁定状态），保证同一时间只有一个业务线程请求缓存，未能获取互斥锁的请求，要么等待锁释放后重新读取缓存，要么就返回空值或者默认值。不给热点数据设置过期时间，由后台异步更新缓存，或者在热点数据准备要过期前，提前通知后台线程更新缓存以及重新设置过期时间；如何避免缓存穿透？当发生缓存雪崩或击穿时，数据库中还是保存了应用要访问的数据，一旦缓存恢复相对应的数据，就可以减轻数据库的压力，而缓存穿透就不一样了。当用户访问的数据，既不在缓存中，也不在数据库中，导致请求在访问缓存时，发现缓存缺失，再去访问数据库时，发现数据库中也没有要访问的数据，没办法构建缓存数据，来服务后续的请求。那么当有大量这样的请求到来时，数据库的压力骤增，这就是缓存穿透的问题。缓存穿透的发生一般有这两种情况：业务误操作，缓存中的数据和数据库中的数据都被误删除了，所以导致缓存和数据库中都没有数据；黑客恶意攻击，故意大量访问某些读取不存在数据的业务；应对缓存穿透的方案，常见的方案有三种。非法请求的限制：当有大量恶意请求访问不存在的数据的时候，也会发生缓存穿透，因此在 API 入口处我们要判断求请求参数是否合理，请求参数是否含有非法值、请求字段是否存在，如果判断出是恶意请求就直接返回错误，避免进一步访问缓存和数据库。设置空值或者默认值：当我们线上业务发现缓存穿透的现象时，可以针对查询的数据，在缓存中设置一个空值或者默认值，这样后续请求就可以从缓存中读取到空值或者默认值，返回给应用，而不会继续查询数据库。使用布隆过滤器快速判断数据是否存在，避免通过查询数据库来判断数据是否存在：我们可以在写入数据库数据时，使用布隆过滤器做个标记，然后在用户请求到来时，业务线程确认缓存失效后，可以通过查询布隆过滤器快速判断数据是否存在，如果不存在，就不用通过查询数据库来判断数据是否存在，即使发生了缓存穿透，大量请求只会查询 Redis 和布隆过滤器，而不会查询数据库，保证了数据库能正常运行，Redis 自身也是支持布隆过滤器的。::: tip推荐阅读：什么是缓存雪崩、击穿、穿透？:::如何设计一个缓存策略，可以动态缓存热点数据呢？由于数据存储受限，系统并不是将所有数据都需要存放到缓存中的，而只是将其中一部分热点数据缓存起来，所以我们要设计一个热点数据动态缓存的策略。热点数据动态缓存的策略总体思路：通过数据最新访问时间来做排名，并过滤掉不常访问的数据，只留下经常访问的数据。以电商平台场景中的例子，现在要求只缓存用户经常访问的 Top 1000 的商品。具体细节如下：先通过缓存系统做一个排序队列（比如存放 1000 个商品），系统会根据商品的访问时间，更新队列信息，越是最近访问的商品排名越靠前；同时系统会定期过滤掉队列中排名最后的 200 个商品，然后再从数据库中随机读取出 200 个商品加入队列中；这样当请求每次到达的时候，会先从队列中获取商品 ID，如果命中，就根据 ID 再从另一个缓存数据结构中读取实际的商品信息，并返回。在 Redis 中可以用 zadd 方法和 zrange 方法来完成排序队列和获取 200 个商品的操作。说说常见的缓存更新策略？常见的缓存更新策略共有 3 种：Cache Aside（旁路缓存）策略；Read/Write Through（读穿 / 写穿）策略；Write Back（写回）策略；实际开发中，Redis 和 MySQL 的更新策略用的是 Cache Aside，另外两种策略应用不了。Cache Aside（旁路缓存）策略Cache Aside（旁路缓存）策略是最常用的，应用程序直接与「数据库、缓存」交互，并负责对缓存的维护，该策略又可以细分为「读策略」和「写策略」。写策略的步骤：先更新数据库中的数据，再删除缓存中的数据。读策略的步骤：如果读取的数据命中了缓存，则直接返回数据；如果读取的数据没有命中缓存，则从数据库中读取数据，然后将数据写入到缓存，并且返回给用户。注意，写策略的步骤的顺序不能倒过来，即不能先删除缓存再更新数据库，原因是在「读 + 写」并发的时候，会出现缓存和数据库的数据不一致性的问题。举个例子，假设某个用户的年龄是 20，请求 A 要更新用户年龄为 21，所以它会删除缓存中的内容。这时，另一个请求 B 要读取这个用户的年龄，它查询缓存发现未命中后，会从数据库中读取到年龄为 20，并且写入到缓存中，然后请求 A 继续更改数据库，将用户的年龄更新为 21。最终，该用户年龄在缓存中是 20（旧值），在数据库中是 21（新值），缓存和数据库的数据不一致。为什么「先更新数据库再删除缓存」不会有数据不一致的问题？继续用「读 + 写」请求的并发的场景来分析。假如某个用户数据在缓存中不存在，请求 A 读取数据时从数据库中查询到年龄为 20，在未写入缓存中时另一个请求 B 更新数据。它更新数据库中的年龄为 21，并且清空缓存。这时请求 A 把从数据库中读到的年龄为 20 的数据写入到缓存中。最终，该用户年龄在缓存中是 20（旧值），在数据库中是 21（新值），缓存和数据库数据不一致。从上面的理论上分析，先更新数据库，再删除缓存也是会出现数据不一致性的问题，但是在实际中，这个问题出现的概率并不高。因为缓存的写入通常要远远快于数据库的写入，所以在实际中很难出现请求 B 已经更新了数据库并且删除了缓存，请求 A 才更新完缓存的情况。而一旦请求 A 早于请求 B 删除缓存之前更新了缓存，那么接下来的请求就会因为缓存不命中而从数据库中重新读取数据，所以不会出现这种不一致的情况。Cache Aside 策略适合读多写少的场景，不适合写多的场景，因为当写入比较频繁时，缓存中的数据会被频繁地清理，这样会对缓存的命中率有一些影响。如果业务对缓存命中率有严格的要求，那么可以考虑两种解决方案：一种做法是在更新数据时也更新缓存，只是在更新缓存前先加一个分布式锁，因为这样在同一时间只允许一个线程更新缓存，就不会产生并发问题了。当然这么做对于写入的性能会有一些影响；另一种做法同样也是在更新数据时更新缓存，只是给缓存加一个较短的过期时间，这样即使出现缓存不一致的情况，缓存的数据也会很快过期，对业务的影响也是可以接受。Read/Write Through（读穿 / 写穿）策略Read/Write Through（读穿 / 写穿）策略原则是应用程序只和缓存交互，不再和数据库交互，而是由缓存和数据库交互，相当于更新数据库的操作由缓存自己代理了。1、Read Through 策略先查询缓存中数据是否存在，如果存在则直接返回，如果不存在，则由缓存组件负责从数据库查询数据，并将结果写入到缓存组件，最后缓存组件将数据返回给应用。2、Write Through 策略当有数据更新的时候，先查询要写入的数据在缓存中是否已经存在：如果缓存中数据已经存在，则更新缓存中的数据，并且由缓存组件同步更新到数据库中，然后缓存组件告知应用程序更新完成。如果缓存中数据不存在，直接更新数据库，然后返回；下面是 Read Through/Write Through 策略的示意图：Read Through/Write Through 策略的特点是由缓存节点而非应用程序来和数据库打交道，在我们开发过程中相比 Cache Aside 策略要少见一些，原因是我们经常使用的分布式缓存组件，无论是 Memcached 还是 Redis 都不提供写入数据库和自动加载数据库中的数据的功能。而我们在使用本地缓存的时候可以考虑使用这种策略。Write Back（写回）策略Write Back（写回）策略在更新数据的时候，只更新缓存，同时将缓存数据设置为脏的，然后立马返回，并不会更新数据库。对于数据库的更新，会通过批量异步更新的方式进行。实际上，Write Back（写回）策略也不能应用到我们常用的数据库和缓存的场景中，因为 Redis 并没有异步更新数据库的功能。Write Back 是计算机体系结构中的设计，比如 CPU 的缓存、操作系统中文件系统的缓存都采用了 Write Back（写回）策略。Write Back 策略特别适合写多的场景，因为发生写操作的时候，只需要更新缓存，就立马返回了。比如，写文件的时候，实际上是写入到文件系统的缓存就返回了，并不会写磁盘。但是带来的问题是，数据不是强一致性的，而且会有数据丢失的风险，因为缓存一般使用内存，而内存是非持久化的，所以一旦缓存机器掉电，就会造成原本缓存中的脏数据丢失。所以你会发现系统在掉电之后，之前写入的文件会有部分丢失，就是因为 Page Cache 还没有来得及刷盘造成的。这里贴一张 CPU 缓存与内存使用 Write Back 策略的流程图：有没有觉得这个流程很熟悉？因为我在写 CPU 缓存文章的时候提到过。如何保证缓存和数据库数据的一致性？::: tip推荐阅读：数据库和缓存如何保证一致性？:::Redis 实战Redis 如何实现延迟队列？延迟队列是指把当前要做的事情，往后推迟一段时间再做。延迟队列的常见使用场景有以下几种：在淘宝、京东等购物平台上下单，超过一定时间未付款，订单会自动取消；打车的时候，在规定时间没有车主接单，平台会取消你的单并提醒你暂时没有车主接单；点外卖的时候，如果商家在 10 分钟还没接单，就会自动取消订单；在 Redis 可以使用有序集合（ZSet）的方式来实现延迟消息队列的，ZSet 有一个 Score 属性可以用来存储延迟执行的时间。 使用 zadd score1 value1 命令就可以一直往内存中生产消息。再利用 zrangebysocre 查询符合条件的所有待处理的任务，通过循环执行队列任务即可。 Redis 的大 key 如何处理？什么是 Redis 大 key？大 key 并不是指 key 的值很大，而是 key 对应的 value 很大。一般而言，下面这两种情况被称为大 key：String 类型的值大于 10 KB；Hash、List、Set、ZSet 类型的元素的个数超过 5000 个；大 key 会造成什么问题？大 key 会带来以下四种影响：客户端超时阻塞。由于 Redis 执行命令是单线程处理，然后在操作大 key 时会比较耗时，那么就会阻塞 Redis，从客户端这一视角看，就是很久很久都没有响应。引发网络阻塞。每次获取大 key 产生的网络流量较大，如果一个 key 的大小是 1 MB，每秒访问量为 1000，那么每秒会产生 1000MB 的流量，这对于普通千兆网卡的服务器来说是灾难性的。阻塞工作线程。如果使用 del 删除大 key 时，会阻塞工作线程，这样就没办法处理后续的命令。内存分布不均。集群模型在 slot 分片均匀情况下，会出现数据和查询倾斜情况，部分有大 key 的 Redis 节点占用内存多，QPS 也会比较大。如何找到大 key？1、redis-cli --bigkeys 查找大 key可以通过 redis-cli --bigkeys 命令查找大 key：redis-cli -h 127.0.0.1 -p6379 -a "password" --bigkeys使用的时候注意事项：最好选择在从节点上执行该命令。因为主节点上执行时，会阻塞主节点；如果没有从节点，那么可以选择在 Redis 实例业务压力的低峰阶段进行扫描查询，以免影响到实例的正常运行；或者可以使用 -i 参数控制扫描间隔，避免长时间扫描降低 Redis 实例的性能。该方式的不足之处：这个方法只能返回每种类型中最大的那个 bigkey，无法得到大小排在前 N 位的 bigkey；对于集合类型来说，这个方法只统计集合元素个数的多少，而不是实际占用的内存量。但是，一个集合中的元素个数多，并不一定占用的内存就多。因为，有可能每个元素占用的内存很小，这样的话，即使元素个数有很多，总内存开销也不大；2、使用 SCAN 命令查找大 key使用 SCAN 命令对数据库扫描，然后用 TYPE 命令获取返回的每一个 key 的类型。对于 String 类型，可以直接使用 STRLEN 命令获取字符串的长度，也就是占用的内存空间字节数。对于集合类型来说，有两种方法可以获得它占用的内存大小：如果能够预先从业务层知道集合元素的平均大小，那么，可以使用下面的命令获取集合元素的个数，然后乘以集合元素的平均大小，这样就能获得集合占用的内存大小了。List 类型：LLEN 命令；Hash 类型：HLEN 命令；Set 类型：SCARD 命令；Sorted Set 类型：ZCARD 命令；如果不能提前知道写入集合的元素大小，可以使用 MEMORY USAGE 命令（需要 Redis 4.0 及以上版本），查询一个键值对占用的内存空间。3、使用 RdbTools 工具查找大 key使用 RdbTools 第三方开源工具，可以用来解析 Redis 快照（RDB）文件，找到其中的大 key。比如，下面这条命令，将大于 10 kb 的 key 输出到一个表格文件。rdb dump.rdb -c memory --bytes 10240 -f redis.csv如何删除大 key？删除操作的本质是要释放键值对占用的内存空间，不要小瞧内存的释放过程。释放内存只是第一步，为了更加高效地管理内存空间，在应用程序释放内存时，操作系统需要把释放掉的内存块插入一个空闲内存块的链表，以便后续进行管理和再分配。这个过程本身需要一定时间，而且会阻塞当前释放内存的应用程序。所以，如果一下子释放了大量内存，空闲内存块链表操作时间就会增加，相应地就会造成 Redis 主线程的阻塞，如果主线程发生了阻塞，其他所有请求可能都会超时，超时越来越多，会造成 Redis 连接耗尽，产生各种异常。因此，删除大 key 这一个动作，我们要小心。具体要怎么做呢？这里给出两种方法：分批次删除异步删除（Redis 4.0 版本以上）1、分批次删除对于删除大 Hash，使用 hscan 命令，每次获取 100 个字段，再用 hdel 命令，每次删除 1 个字段。Python 代码：def del_large_hash(): r = redis.StrictRedis(host='redis-host1', port=6379) large_hash_key ="xxx" #要删除的大 hash 键名 cursor = '0' while cursor != 0: # 使用 hscan 命令，每次获取 100 个字段 cursor, data = r.hscan(large_hash_key, cursor=cursor, count=100) for item in data.items(): # 再用 hdel 命令，每次删除 1 个字段 r.hdel(large_hash_key, item[0])对于删除大 List，通过 ltrim 命令，每次删除少量元素。Python 代码：def del_large_list(): r = redis.StrictRedis(host='redis-host1', port=6379) large_list_key = 'xxx' #要删除的大 list 的键名 while r.llen(large_list_key)>0: #每次只删除最右 100 个元素 r.ltrim(large_list_key, 0, -101) 对于删除大 Set，使用 sscan 命令，每次扫描集合中 100 个元素，再用 srem 命令每次删除一个键。Python 代码：def del_large_set(): r = redis.StrictRedis(host='redis-host1', port=6379) large_set_key = 'xxx' # 要删除的大 set 的键名 cursor = '0' while cursor != 0: # 使用 sscan 命令，每次扫描集合中 100 个元素 cursor, data = r.sscan(large_set_key, cursor=cursor, count=100) for item in data: # 再用 srem 命令每次删除一个键 r.srem(large_size_key, item)对于删除大 ZSet，使用 zremrangebyrank 命令，每次删除 top 100 个元素。Python 代码：def del_large_sortedset(): r = redis.StrictRedis(host='large_sortedset_key', port=6379) large_sortedset_key='xxx' while r.zcard(large_sortedset_key)>0: # 使用 zremrangebyrank 命令，每次删除 top 100 个元素 r.zremrangebyrank(large_sortedset_key,0,99) 2、异步删除从 Redis 4.0 版本开始，可以采用异步删除法，用 unlink 命令代替 del 来删除。这样 Redis 会将这个 key 放入到一个异步线程中进行删除，这样不会阻塞主线程。除了主动调用 unlink 命令实现异步删除之外，我们还可以通过配置参数，达到某些条件的时候自动进行异步删除。主要有 4 种场景，默认都是关闭的：lazyfree-lazy-eviction no lazyfree-lazy-expire no lazyfree-lazy-server-del noslave-lazy-flush no它们代表的含义如下：lazyfree-lazy-eviction：表示当 Redis 运行内存超过 maxmeory 时，是否开启 lazy free 机制删除；lazyfree-lazy-expire：表示设置了过期时间的键值，当过期之后是否开启 lazy free 机制删除；lazyfree-lazy-server-del：有些指令在处理已存在的键时，会带有一个隐式的 del 键的操作，比如 rename 命令，当目标键已存在，Redis 会先删除目标键，如果这些目标键是一个 big key，就会造成阻塞删除的问题，此配置表示在这种场景中是否开启 lazy free 机制删除；slave-lazy-flush：针对 slave (从节点) 进行全量数据同步，slave 在加载 master 的 RDB 文件前，会运行 flushall 来清理自己的数据，它表示此时是否开启 lazy free 机制删除。建议开启其中的 lazyfree-lazy-eviction、lazyfree-lazy-expire、lazyfree-lazy-server-del 等配置，这样就可以有效的提高主线程的执行效率。Redis 管道有什么用？管道技术（Pipeline）是客户端提供的一种批处理技术，用于一次处理多个 Redis 命令，从而提高整个交互的性能。普通命令模式，如下图所示：管道模式，如下图所示：使用管道技术可以解决多个命令执行时的网络等待，它是把多个命令整合到一起发送给服务器端处理之后统一返回给客户端，这样就免去了每条命令执行后都要等待的情况，从而有效地提高了程序的执行效率。但使用管道技术也要注意避免发送的命令过大，或管道内的数据太多而导致的网络阻塞。要注意的是，管道技术本质上是客户端提供的功能，而非 Redis 服务器端的功能。Redis 事务支持回滚吗？MySQL 在执行事务时，会提供回滚机制，当事务执行发生错误时，事务中的所有操作都会撤销，已经修改的数据也会被恢复到事务执行前的状态。Redis 中并没有提供回滚机制，虽然 Redis 提供了 DISCARD 命令，但是这个命令只能用来主动放弃事务执行，把暂存的命令队列清空，起不到回滚的效果。下面是 DISCARD 命令用法：#读取 count 的值4 127.0.0.1:6379> GET count "1" #开启事务 127.0.0.1:6379> MULTI OK #发送事务的第一个操作，对count减1 127.0.0.1:6379> DECR count QUEUED #执行DISCARD命令，主动放弃事务 127.0.0.1:6379> DISCARD OK #再次读取a:stock的值，值没有被修改 127.0.0.1:6379> GET count "1"事务执行过程中，如果命令入队时没报错，而事务提交后，实际执行时报错了，正确的命令依然可以正常执行，所以这可以看出 Redis 并不一定保证原子性（原子性：事务中的命令要不全部成功，要不全部失败）。比如下面这个例子：#获取name原本的值 127.0.0.1:6379> GET name "xiaolin" #开启事务 127.0.0.1:6379> MULTI OK #设置新值 127.0.0.1:6379(TX)> SET name xialincoding QUEUED #注意，这条命令是错误的 # expire 过期时间正确来说是数字，并不是‘10s’字符串，但是还是入队成功了 127.0.0.1:6379(TX)> EXPIRE name 10s QUEUED #提交事务，执行报错 #可以看到 set 执行成功，而 expire 执行错误。 127.0.0.1:6379(TX)> EXEC 1) OK 2) (error) ERR value is not an integer or out of range #可以看到，name 还是被设置为新值了 127.0.0.1:6379> GET name "xialincoding"为什么 Redis 不支持事务回滚？Redis 官方文档的解释如下：大概的意思是，作者不支持事务回滚的原因有以下两个：他认为 Redis 事务的执行时，错误通常都是编程错误造成的，这种错误通常只会出现在开发环境中，而很少会在实际的生产环境中出现，所以他认为没有必要为 Redis 开发事务回滚功能；不支持事务回滚是因为这种复杂的功能和 Redis 追求的简单高效的设计主旨不符合。这里不支持事务回滚，指的是不支持事务运行时错误的事务回滚。如何用 Redis 实现分布式锁的？分布式锁是用于分布式环境下并发控制的一种机制，用于控制某个资源在同一时刻只能被一个应用所使用。如下图所示：Redis 本身可以被多个客户端共享访问，正好就是一个共享存储系统，可以用来保存分布式锁，而且 Redis 的读写性能高，可以应对高并发的锁操作场景。Redis 的 SET 命令有个 NX 参数可以实现「key 不存在才插入」，所以可以用它来实现分布式锁：如果 key 不存在，则显示插入成功，可以用来表示加锁成功；如果 key 存在，则会显示插入失败，可以用来表示加锁失败。基于 Redis 节点实现分布式锁时，对于加锁操作，我们需要满足三个条件。加锁包括了读取锁变量、检查锁变量值和设置锁变量值三个操作，但需要以原子操作的方式完成，所以，我们使用 SET 命令带上 NX 选项来实现加锁；锁变量需要设置过期时间，以免客户端拿到锁后发生异常，导致锁一直无法释放，所以，我们在 SET 命令执行时加上 EX/PX 选项，设置其过期时间；锁变量的值需要能区分来自不同客户端的加锁操作，以免在释放锁时，出现误释放操作，所以，我们使用 SET 命令设置锁变量值时，每个客户端设置的值是一个唯一值，用于标识客户端；满足这三个条件的分布式命令如下：SET lock_key unique_value NX PX 10000 lock_key 就是 key 键；unique_value 是客户端生成的唯一的标识，区分来自不同客户端的锁操作；NX 代表只在 lock_key 不存在时，才对 lock_key 进行设置操作；PX 10000 表示设置 lock_key 的过期时间为 10s，这是为了避免客户端发生异常而无法释放锁。而解锁的过程就是将 lock_key 键删除（del lock_key），但不能乱删，要保证执行操作的客户端就是加锁的客户端。所以，解锁的时候，我们要先判断锁的 unique_value 是否为加锁客户端，是的话，才将 lock_key 键删除。可以看到，解锁是有两个操作，这时就需要 Lua 脚本来保证解锁的原子性，因为 Redis 在执行 Lua 脚本时，可以以原子性的方式执行，保证了锁释放操作的原子性。// 释放锁时，先比较 unique_value 是否相等，避免锁的误释放 if redis.call("get",KEYS[1]) == ARGV[1] then return redis.call("del",KEYS[1]) else return 0 end这样一来，就通过使用 SET 命令和 Lua 脚本在 Redis 单节点上完成了分布式锁的加锁和解锁。基于 Redis 实现分布式锁有什么优缺点？基于 Redis 实现分布式锁的优点：性能高效（这是选择缓存实现分布式锁最核心的出发点）。实现方便。很多研发工程师选择使用 Redis 来实现分布式锁，很大成分上是因为 Redis 提供了 setnx 方法，实现分布式锁很方便。避免单点故障（因为 Redis 是跨集群部署的，自然就避免了单点故障）。基于 Redis 实现分布式锁的缺点：超时时间不好设置。如果锁的超时时间设置过长，会影响性能，如果设置的超时时间过短会保护不到共享资源。比如在有些场景中，一个线程 A 获取到了锁之后，由于业务代码执行时间可能比较长，导致超过了锁的超时时间，自动失效，注意 A 线程没执行完，后续线程 B 又意外的持有了锁，意味着可以操作共享资源，那么两个线程之间的共享资源就没办法进行保护了。那么如何合理设置超时时间呢？ 我们可以基于续约的方式设置超时时间：先给锁设置一个超时时间，然后启动一个守护线程，让守护线程在一段时间后，重新设置这个锁的超时时间。实现方式就是：写一个守护线程，然后去判断锁的情况，当锁快失效的时候，再次进行续约加锁，当主线程执行完成后，销毁续约锁即可，不过这种方式实现起来相对复杂。Redis 主从复制模式中的数据是异步复制的，这样导致分布式锁的不可靠性。如果在 Redis 主节点获取到锁后，在没有同步到其他节点时，Redis 主节点宕机了，此时新的 Redis 主节点依然可以获取锁，所以多个应用服务就可以同时获取到锁。Redis 如何解决集群情况下分布式锁的可靠性？为了保证集群环境下分布式锁的可靠性，Redis 官方已经设计了一个分布式锁算法 Redlock（红锁）。它是基于多个 Redis 节点的分布式锁，即使有节点发生了故障，锁变量仍然是存在的，客户端还是可以完成锁操作。官方推荐是至少部署 5 个 Redis 节点，而且都是主节点，它们之间没有任何关系，都是一个个孤立的节点。Redlock 算法的基本思路，是让客户端和多个独立的 Redis 节点依次请求申请加锁，如果客户端能够和半数以上的节点成功地完成加锁操作，那么我们就认为，客户端成功地获得分布式锁，否则加锁失败。这样一来，即使有某个 Redis 节点发生故障，因为锁的数据在其他节点上也有保存，所以客户端仍然可以正常地进行锁操作，锁的数据也不会丢失。Redlock 算法加锁三个过程：第一步是，客户端获取当前时间（t1）。第二步是，客户端按顺序依次向 N 个 Redis 节点执行加锁操作：加锁操作使用 SET 命令，带上 NX，EX/PX 选项，以及带上客户端的唯一标识。如果某个 Redis 节点发生故障了，为了保证在这种情况下，Redlock 算法能够继续运行，我们需要给「加锁操作」设置一个超时时间（不是对「锁」设置超时时间，而是对「加锁操作」设置超时时间），加锁操作的超时时间需要远远地小于锁的过期时间，一般也就是设置为几十毫秒。第三步是，一旦客户端从超过半数（大于等于 N/2+1）的 Redis 节点上成功获取到了锁，就再次获取当前时间（t2），然后计算计算整个加锁过程的总耗时（t2-t1）。如果 t2-t1 < 锁的过期时间，此时，认为客户端加锁成功，否则认为加锁失败。可以看到，加锁成功要同时满足两个条件（简述：如果有超过半数的 Redis 节点成功的获取到了锁，并且总耗时没有超过锁的有效时间，那么就是加锁成功）：条件一：客户端从超过半数（大于等于 N/2+1）的 Redis 节点上成功获取到了锁；条件二：客户端从大多数节点获取锁的总耗时（t2-t1）小于锁设置的过期时间。加锁成功后，客户端需要重新计算这把锁的有效时间，计算的结果是「锁最初设置的过期时间」减去「客户端从大多数节点获取锁的总耗时（t2-t1）」。如果计算的结果已经来不及完成共享数据的操作了，我们可以释放锁，以免出现还没完成数据操作，锁就过期了的情况。加锁失败后，客户端向所有 Redis 节点发起释放锁的操作，释放锁的操作和在单节点上释放锁的操作一样，只要执行释放锁的 Lua 脚本就可以了。
AI 摘要 文章剖析RPC框架的“其他模块”：用动态代理RpcClientProxy屏蔽网络细节，统一封装请求；借Spring BeanPostProcessor，通过@RpcService发布服务、@RpcReference注入代理，实现注解式注册与消费。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结11 RPC 框架代码分析之其他模块动态代理屏蔽网络传输细节我们在前面的章节讲到过我们需要用到动态代理来屏蔽复杂的网络传输细节。对应的代码: RpcClientProxy.java@Slf4j public class RpcClientProxy implements InvocationHandler { public <T> T getProxy(Class<T> clazz) { return (T) Proxy.newProxyInstance(clazz.getClassLoader(), new Class<?>[]{clazz}, this); } @Override public Object invoke(Object proxy, Method method, Object[] args) { } }当我们去调用一个远程的方法的时候，实际上是通过代理对象调用的。获取代理对象的方法如下： public <T> T getProxy(Class<T> clazz) { return (T) Proxy.newProxyInstance(clazz.getClassLoader(), new Class<?>[]{clazz}, this); }网络传输细节都被封装在了 invoke() 方法中。 public Object invoke(Object proxy, Method method, Object[] args) { log.info("invoked method: [{}]", method.getName()); RpcRequest rpcRequest = RpcRequest.builder().methodName(method.getName()) .parameters(args) .interfaceName(method.getDeclaringClass().getName()) .paramTypes(method.getParameterTypes()) .requestId(UUID.randomUUID().toString()) .group(rpcServiceProperties.getGroup()) .version(rpcServiceProperties.getVersion()) .build(); RpcResponse<Object> rpcResponse = null; if (rpcRequestTransport instanceof NettyRpcClient) { CompletableFuture<RpcResponse<Object>> completableFuture = (CompletableFuture<RpcResponse<Object>>) rpcRequestTransport.sendRpcRequest(rpcRequest); rpcResponse = completableFuture.get(); } if (rpcRequestTransport instanceof SocketRpcClient) { rpcResponse = (RpcResponse<Object>) rpcRequestTransport.sendRpcRequest(rpcRequest); } this.check(rpcResponse, rpcRequest); return rpcResponse.getData(); }通过注解注册/消费服务我们这里借用了 Spring 容器相关的功能。核心代码都放在了 ： src/main/java/github/javaguide/spring 包下面。我们定义两个注解：RcpService ：注册服务RpcReference ：消费服务**RcpService.java**@Documented @Retention(RetentionPolicy.RUNTIME) @Target({ElementType.TYPE}) @Inherited public @interface RpcService { /** * Service version, default value is empty string */ String version() default ""; /** * Service group, default value is empty string */ String group() default ""; }**RpcReference.java**@Documented @Retention(RetentionPolicy.RUNTIME) @Target({ElementType.FIELD}) @Inherited public @interface RpcReference { /** * Service version, default value is empty string */ String version() default ""; /** * Service group, default value is empty string */ String group() default ""; }简单说一下原理。我们实现需要 BeanPostProcessor 接口并重写 postProcessBeforeInitialization()方法和 postProcessAfterInitialization() 方法。Spring bean 在实例化之前会调用 postProcessBeforeInitialization()方法，在 Spring bean 实例化之后会调用 postProcessAfterInitialization() 方法。@Slf4j @Component public class SpringBeanPostProcessor implements BeanPostProcessor { @Override public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { } @Override public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { } }被我们使用 RpcService和RpcReference 注解的类都算是 Spring Bean。我们可以在postProcessBeforeInitialization()方法中去判断类上是否有RpcService 注解。如果有的话，就取出 group 和 version 的值。然后，再调用 ServiceProvider 的 publishService() 方法发布服务即可！我们可以在 postProcessAfterInitialization() 方法中遍历类的属性上是否有 RpcReference 注解。如果有的话，我们就通过反射将这个属性赋值即可！
AI 摘要 文章剖析自研RPC框架的注册中心模块：定义ServiceRegistry与ServiceDiscovery接口，基于ZooKeeper实现服务注册与发现，支持多版本、多分组、多实例，配合负载均衡策略完成地址选取。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结10 RPC 框架代码分析之注册中心模块我们之前在“如何自己实现一个 RPC 框架？”这篇文章中介绍到说：注册中心负责服务地址的注册与查找，相当于目录服务。 服务端启动的时候将服务名称及其对应的地址(ip+port)注册到注册中心，服务消费端根据服务名称找到对应的服务地址。有了服务地址之后，服务消费端就可以通过网络请求服务端了。简单来说注册中心就像是一个中转站，提供的作用就是根据调用的服务名称找到远程服务的地址（数据保存服务）。注册中心模块整体结构如下：我们定义了两个接口 ServiceDiscovery.java 和 ServiceRegistry.java，这两个接口分别定义了服务发现和服务注册行为。**ServiceRegistry.java**/** * 服务注册 * * @author shuang.kou * @createTime 2020年05月13日 08:39:00 */ public interface ServiceRegistry { /** * 注册服务到注册中心 * * @param rpcServiceName 完整的服务名称（class name+group+version） * @param inetSocketAddress 远程服务地址 */ void registerService(String rpcServiceName, InetSocketAddress inetSocketAddress); }**ServiceDiscovery.java**/** * 服务发现 * * @author shuang.kou * @createTime 2020年06月01日 15:16:00 */ public interface ServiceDiscovery { /** * 根据 rpcServiceName 获取远程服务地址 * * @param rpcServiceName 完整的服务名称（class name+group+version） * @return 远程服务地址 */ InetSocketAddress lookupService(String rpcServiceName); }接下来，我们使用 zookeeper 作为注册中心的实现方式，并实现了这两个接口。**ZkServiceRegistry.java**/** * 服务注册（基于zookeeper实现） */ @Slf4j public class ZkServiceRegistry implements ServiceRegistry { @Override public void registerService(String rpcServiceName, InetSocketAddress inetSocketAddress) { String servicePath = CuratorUtils.ZK_REGISTER_ROOT_PATH + "/" + rpcServiceName + inetSocketAddress.toString(); CuratorFramework zkClient = CuratorUtils.getZkClient(); CuratorUtils.createPersistentNode(zkClient, servicePath); } }当我们的服务被注册进 zookeeper 的时候，我们将完整的服务名称 rpcServiceName （class name+group+version）作为根节点 ，子节点是对应的服务地址（ip+端口号）。class name : 服务接口名也就是类名比如：github.javaguide.HelloService。version : 服务版本。主要是为后续不兼容升级提供可能group :服务所在的组。主要用于处理一个接口有多个类实现的情况。一个根节点（rpcServiceName）可能会对应多个服务地址（相同服务被部署多份的情况）。如果我们要获得某个服务对应的地址的话，就直接根据完整的服务名称来获取到其下的所有子节点，然后通过具体的负载均衡策略取出一个就可以了。相关代码如下在 ZkServiceDiscovery.java中已经给出。**ZkServiceDiscovery.java**/** * 服务发现（基于zookeeper实现） */ @Slf4j public class ZkServiceDiscovery implements ServiceDiscovery { private final LoadBalance loadBalance; public ZkServiceDiscovery() { this.loadBalance = new RandomLoadBalance(); } @Override public InetSocketAddress lookupService(String rpcServiceName) { CuratorFramework zkClient = CuratorUtils.getZkClient(); List<String> serviceUrlList = CuratorUtils.getChildrenNodes(zkClient, rpcServiceName); if (serviceUrlList.size() == 0) { throw new RpcException(RpcErrorMessage.SERVICE_CAN_NOT_BE_FOUND, rpcServiceName); } // load balancing String targetServiceUrl = loadBalance.selectServiceAddress(serviceUrlList); log.info("Successfully found the service address:[{}]", targetServiceUrl); String[] socketAddressArray = targetServiceUrl.split(":"); String host = socketAddressArray[0]; int port = Integer.parseInt(socketAddressArray[1]); return new InetSocketAddress(host, port); } }我们根据完整的服务名称便可以将对应的服务地址查出来， 查出来的服务地址可能并不止一个。所以，我们可以通过对应的负载均衡策略来选择出一个服务地址。**CuratorUtils.java**另外，我们还自定义了一个 ZooKeeper Java 客户端 Curtor 的工具类 CuratorUtils.java 。关于这个工具类，这里就不再提了。在《08 Zookeeper 常用命令+ Curtor 使用详解》中已经介绍的非常详细了。
AI 摘要 文章剖析了自研 RPC 框架的网络传输模块：先定义 RpcRequest/Response 实体，再抽象 RpcRequestTransport 接口，分别用 Socket 与 Netty 实现客户端与服务端收发；Netty 版引入编解码器、自定义协议及心跳保活，并借助 CompletableFuture 实现异步回调，完成高性能远程调用。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结09 RPC 框架代码分析之网络传输模块以下提到的 服务端 指的是提供服务/方法的一端，客户端 指的是调用远程(服务端)服务/方法的一端。我们之前在“如何自己实现一个 RPC 框架？”这篇文章中介绍到说：既然我们要调用远程的方法，就要发送网络请求来传递目标类和方法的信息以及方法的参数等数据到服务端。 这就涉及到了网络传输！网络传输具体实现你可以使用 Socket （ Java 中最原始、最基础的网络通信方式。但是，Socket 是阻塞 IO、性能低并且功能单一）。你也可以使用同步非阻塞的 I/O 模型 NIO ，但是用它来进行网络编程真的太麻烦了。不过没关系，你可以使用基于 NIO 的网络编程框架 Netty ，它将是你最好的选择！guide-rpc-framework 使用了一种基于 Socket，一种基于 Netty 的方式（循序渐进）。网络传输模块整体结构如下：一共被分为了 4 个包constants : 存放一些网络传输模块共用的常量dto : 用于网络传输的类。handler : 里面只有一个用于处理 rpc 请求的类RpcRequestHandler（根据 rpc 请求调用目标类的目标方法）。transport : 用户网络传输相关类（真正传输网络请求的地方。提供了 Socket 和 Netty 两种网络传输方式）。网络传输实体类网络传输实体类在 dto 包下，主要有两个类。**RpcRequest.java**rpc 请求实体类。当你要调用远程方法的时候，你需要先传输一个 RpcRequest 给对方，RpcRequest 里面包含了要调用的目标方法和类的名称、参数等数据。另外，version 字段（服务版本）主要是为后续不兼容升级提供可能。group 字段主要用于处理一个接口有多个类实现的情况。@AllArgsConstructor @NoArgsConstructor @Getter @Builder @ToString public class RpcRequest implements Serializable { private static final long serialVersionUID = 1905122041950251207L; private String requestId; private String interfaceName; private String methodName; private Object[] parameters; private Class<?>[] paramTypes; private RpcMessageType rpcMessageType; private String version; private String group; public RpcServiceProperties toRpcProperties() { return RpcServiceProperties.builder().serviceName(this.getInterfaceName()) .version(this.getVersion()) .group(this.getGroup()).build(); } }**RpcResponse.java**既然有了 rpc 请求实体类，那肯定就要有 rpc 响应实体类了。当服务端通过 RpcRequest 中的相关数据调用到目标服务的目标方法之后，调用结果就通过 RpcResponse 返回给客户端。@AllArgsConstructor @NoArgsConstructor @Getter @Setter @Builder @ToString public class RpcResponse<T> implements Serializable { private static final long serialVersionUID = 715745410605631233L; private String requestId; /** * response code */ private Integer code; /** * response message */ private String message; /** * response body */ private T data; public static <T> RpcResponse<T> success(T data, String requestId) { RpcResponse<T> response = new RpcResponse<>(); response.setCode(RpcResponseCode.SUCCESS.getCode()); response.setMessage(RpcResponseCode.SUCCESS.getMessage()); response.setRequestId(requestId); if (null != data) { response.setData(data); } return response; } public static <T> RpcResponse<T> fail(RpcResponseCode rpcResponseCode) { RpcResponse<T> response = new RpcResponse<>(); response.setCode(rpcResponseCode.getCode()); response.setMessage(rpcResponseCode.getMessage()); return response; } }网络传输由于，这部分我提供了一种基于 Socket，一种基于 Netty 的网络传输方式（循序渐进）。因此，我先定义了一个发送 RPC 请求的顶层接口，然后我们分别使用 Socket 和 Netty 两种方式对这个接口进行实现即可！RpcRequestTransport.java 传输请求的接口/** * send RpcRequest。 * * @author shuang.kou * @createTime 2020年05月29日 13:26:00 */ @SPI public interface RpcRequestTransport { /** * send rpc request to server and get result * * @param rpcRequest message body * @return data from server */ Object sendRpcRequest(RpcRequest rpcRequest); }下面，我们先来看一下比较简单点的使用 Socket 进行网络传输的方式。Socket客户端这里的客户端实际就是发送 RPC 请求的一端，可以对照我们之间画的 RPC 调用的原理图来理解。客户端主要用于发送网络请求到服务端（目标方法所在的服务器）。当我们知道了服务端的地址之后，我们就可以通过 SocketRpcClient 发送 rpc 请求(RpcRequest) 到服务端了(如果我们要找到服务端的地址，涉及到了注册中心相关的知识。下一节会提到。)。我们直接实现上面定义的 RpcRequestTransport.java 即可。这样的话，通过 Socket 来传输消息的模块就写好了！/** * 基于 Socket 传输 RpcRequest * * @author shuang.kou * @createTime 2020年05月10日 18:40:00 */ @AllArgsConstructor @Slf4j public class SocketRpcClient implements RpcRequestTransport { private final ServiceDiscovery serviceDiscovery; public SocketRpcClient() { this.serviceDiscovery = ExtensionLoader.getExtensionLoader(ServiceDiscovery.class).getExtension("zk"); } @Override public Object sendRpcRequest(RpcRequest rpcRequest) { // build rpc service name by rpcRequest String rpcServiceName = RpcServiceProperties.builder().serviceName(rpcRequest.getInterfaceName()) .group(rpcRequest.getGroup()).version(rpcRequest.getVersion()).build().toRpcServiceName(); InetSocketAddress inetSocketAddress = serviceDiscovery.lookupService(rpcServiceName); try (Socket socket = new Socket()) { socket.connect(inetSocketAddress); ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream()); // Send data to the server through the output stream objectOutputStream.writeObject(rpcRequest); ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream()); // Read RpcResponse from the input stream return objectInputStream.readObject(); } catch (IOException | ClassNotFoundException e) { throw new RpcException("调用服务失败:", e); } } }上面的逻辑很简单，就是对 Socket 发送网络请求这个基础知识的运用。我这里就不再对上面的代码进行解析了，看不懂的小伙伴自行翻看之前关于 Socket 讲解的章节。服务端**SocketRpcServer.java**Socket 服务端。用于等待客户端连接。当客户端成功连接之后，就可以发送 rpc 请求(RpcRequest) 到服务端了。然后，服务端拿到 RpcRequest就会去执行对应的方法。执行完对应的方法之后，就把执行得到的结果放在 RpcResponse 中返回给客户端。/** * @author shuang.kou * @createTime 2020年05月10日 08:01:00 */ @Slf4j public class SocketRpcServer { private final ExecutorService threadPool; private final ServiceProvider serviceProvider; public SocketRpcServer() { threadPool = ThreadPoolFactoryUtils.createCustomThreadPoolIfAbsent("socket-server-rpc-pool"); serviceProvider = SingletonFactory.getInstance(ServiceProviderImpl.class); } public void registerService(Object service) { serviceProvider.publishService(service); } public void registerService(Object service, RpcServiceProperties rpcServiceProperties) { serviceProvider.publishService(service, rpcServiceProperties); } public void start() { try (ServerSocket server = new ServerSocket()) { String host = InetAddress.getLocalHost().getHostAddress(); server.bind(new InetSocketAddress(host, PORT)); CustomShutdownHook.getCustomShutdownHook().clearAll(); Socket socket; while ((socket = server.accept()) != null) { log.info("client connected [{}]", socket.getInetAddress()); threadPool.execute(new SocketRpcRequestHandlerRunnable(socket)); } threadPool.shutdown(); } catch (IOException e) { log.error("occur IOException:", e); } } }NettyNetty 这部分的原理也差不多，不过实现代码差别很大。客户端**NettyClient.java**Netty 客户端主要提供了:**doConnect()** :用于连接服务端（目标方法所在的服务器）并返回对应的 Channel。当我们知道了服务端的地址之后，我们就可以通过 NettyClient 成功连接服务端了。（有了 Channel 之后就能发送数据到服务端了）**sendRpcRequest()** : 用于传输 rpc 请求(RpcRequest) 到服务端。@Slf4j public final class NettyRpcClient implements RpcRequestTransport { private final ServiceDiscovery serviceDiscovery; private final UnprocessedRequests unprocessedRequests; private final ChannelProvider channelProvider; private final Bootstrap bootstrap; private final EventLoopGroup eventLoopGroup; @SneakyThrows public Channel doConnect(InetSocketAddress inetSocketAddress) { CompletableFuture<Channel> completableFuture = new CompletableFuture<>(); bootstrap.connect(inetSocketAddress).addListener((ChannelFutureListener) future -> { if (future.isSuccess()) { log.info("The client has connected [{}] successful!", inetSocketAddress.toString()); completableFuture.complete(future.channel()); } else { throw new IllegalStateException(); } }); return completableFuture.get(); } @Override public Object sendRpcRequest(RpcRequest rpcRequest) { // build return value CompletableFuture<RpcResponse<Object>> resultFuture = new CompletableFuture<>(); // build rpc service name by rpcRequest String rpcServiceName = rpcRequest.toRpcProperties().toRpcServiceName(); // get server address InetSocketAddress inetSocketAddress = serviceDiscovery.lookupService(rpcServiceName); // get server address related channel Channel channel = getChannel(inetSocketAddress); if (channel.isActive()) { // put unprocessed request unprocessedRequests.put(rpcRequest.getRequestId(), resultFuture); RpcMessage rpcMessage = new RpcMessage(); rpcMessage.setData(rpcRequest); rpcMessage.setCodec(SerializationTypeEnum.PROTOSTUFF.getCode()); rpcMessage.setCompress(CompressTypeEnum.GZIP.getCode()); rpcMessage.setMessageType(RpcConstants.REQUEST_TYPE); channel.writeAndFlush(rpcMessage).addListener((ChannelFutureListener) future -> { if (future.isSuccess()) { log.info("client send message: [{}]", rpcMessage); } else { future.channel().close(); resultFuture.completeExceptionally(future.cause()); log.error("Send failed:", future.cause()); } }); } else { throw new IllegalStateException(); } return resultFuture; } }**UnprocessedRequests.java**用于存放未被服务端处理的请求（建议限制 map 容器大小，避免未处理请求过多 OOM)。public class UnprocessedRequests { private static final Map<String, CompletableFuture<RpcResponse<Object>>> UNPROCESSED_RESPONSE_FUTURES = new ConcurrentHashMap<>(); public void put(String requestId, CompletableFuture<RpcResponse<Object>> future) { UNPROCESSED_RESPONSE_FUTURES.put(requestId, future); } public void complete(RpcResponse<Object> rpcResponse) { CompletableFuture<RpcResponse<Object>> future = UNPROCESSED_RESPONSE_FUTURES.remove(rpcResponse.getRequestId()); if (null != future) { future.complete(rpcResponse); } else { throw new IllegalStateException(); } } }**NettyClientHandler**自定义客户端 ChannelHandler 用于处理服务器发送的数据。@Slf4j public class NettyClientHandler extends ChannelInboundHandlerAdapter { private final UnprocessedRequests unprocessedRequests; private final ChannelProvider channelProvider; public NettyClientHandler() { this.unprocessedRequests = SingletonFactory.getInstance(UnprocessedRequests.class); this.channelProvider = SingletonFactory.getInstance(ChannelProvider.class); } /** * 读取从服务端返回的消息 */ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { try { log.info("client receive msg: [{}]", msg); if (msg instanceof RpcResponse) { RpcResponse<Object> rpcResponse = (RpcResponse<Object>) msg; unprocessedRequests.complete(rpcResponse); } } finally { ReferenceCountUtil.release(msg); } } // Netty 心跳机制相关。保证客户端和服务端的连接不被断掉，避免重连。 @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { //省略部分代码 } }从代码中，可以看出当 rpc 请求被成功处理（客户端收到服务端的执行结果）之后，我们调用了 unprocessedRequests.complete(rpcResponse) 方法，这样的话，你只需要通过下面的方式就能成功接收到服务端返回的结果。CompletableFuture<RpcResponse> completableFuture = (CompletableFuture<RpcResponse>) clientTransport.sendRpcRequest(rpcRequest); rpcResponse = completableFuture.get();**ChannelProvider.java**用于存放 Channel（Channel用于在服务端和客户端之间传输数据）。@Slf4j public class ChannelProvider { private final Map<String, Channel> channelMap; public ChannelProvider() { channelMap = new ConcurrentHashMap<>(); } public Channel get(InetSocketAddress inetSocketAddress) { String key = inetSocketAddress.toString(); // determine if there is a connection for the corresponding address if (channelMap.containsKey(key)) { Channel channel = channelMap.get(key); // if so, determine if the connection is available, and if so, get it directly if (channel != null && channel.isActive()) { return channel; } else { channelMap.remove(key); } } return null; } public void set(InetSocketAddress inetSocketAddress, Channel channel) { String key = inetSocketAddress.toString(); channelMap.put(key, channel); } public void remove(InetSocketAddress inetSocketAddress) { String key = inetSocketAddress.toString(); channelMap.remove(key); log.info("Channel map size :[{}]", channelMap.size()); } }服务端相关**NettyRpcServer.java**Netty 服务端。并监听客户端的连接。另外，还提供了两个用户手动注册服务的方法（_还可以通过注解__RpcService__注册服务，这个后面也会介绍到_）。@Slf4j @Component public class NettyRpcServer { public static final int PORT = 9998; private final ServiceProvider serviceProvider = SingletonFactory.getInstance(ServiceProviderImpl.class); public void registerService(Object service, RpcServiceProperties rpcServiceProperties) { serviceProvider.publishService(service, rpcServiceProperties); } @SneakyThrows public void start() { CustomShutdownHook.getCustomShutdownHook().clearAll(); String host = InetAddress.getLocalHost().getHostAddress(); EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workerGroup = new NioEventLoopGroup(); DefaultEventExecutorGroup serviceHandlerGroup = new DefaultEventExecutorGroup( RuntimeUtil.cpus() * 2, ThreadPoolFactoryUtils.createThreadFactory("service-handler-group", false) ); try { ServerBootstrap b = new ServerBootstrap(); b.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) // TCP默认开启了 Nagle 算法，该算法的作用是尽可能的发送大数据快，减少网络传输。TCP_NODELAY 参数的作用就是控制是否启用 Nagle 算法。 .childOption(ChannelOption.TCP_NODELAY, true) // 是否开启 TCP 底层心跳机制 .childOption(ChannelOption.SO_KEEPALIVE, true) //表示系统用于临时存放已完成三次握手的请求的队列的最大长度,如果连接建立频繁，服务器处理创建新连接较慢，可以适当调大这个参数 .option(ChannelOption.SO_BACKLOG, 128) .handler(new LoggingHandler(LogLevel.INFO)) // 当客户端第一次进行请求的时候才会进行初始化 .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) { // 30 秒之内没有收到客户端请求的话就关闭连接 ChannelPipeline p = ch.pipeline(); p.addLast(new IdleStateHandler(30, 0, 0, TimeUnit.SECONDS)); p.addLast(new RpcMessageEncoder()); p.addLast(new RpcMessageDecoder()); p.addLast(serviceHandlerGroup, new NettyRpcServerHandler()); } }); // 绑定端口，同步等待绑定成功 ChannelFuture f = b.bind(host, PORT).sync(); // 等待服务端监听端口关闭 f.channel().closeFuture().sync(); } catch (InterruptedException e) { log.error("occur exception when start server:", e); } finally { log.error("shutdown bossGroup and workerGroup"); bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); serviceHandlerGroup.shutdownGracefully(); } } }**NettyServerHandler.java**自定义服务端 ChannelHandler 用于处理客户端发送的数据。当客户端发的 rpc 请求(RpcRequest) 来了之后，服务端就会处理 rpc 请求(RpcRequest) ，处理完之后就把得到 rpc 相应(RpcResponse)传输给客户端。@Slf4j public class NettyServerHandler extends ChannelInboundHandlerAdapter { private final RpcRequestHandler rpcRequestHandler; public NettyServerHandler() { this.rpcRequestHandler = SingletonFactory.getInstance(RpcRequestHandler.class); } /** * 读取从客户端消息，然后调用目标服务的目标方法并返回给客户端。 */ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { // 省略部分代码 } // Netty 心跳机制相关。保证客户端和服务端的连接不被断掉，避免重连。 @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { // 省略部分代码 } }传输协议在《如何自己实现一个 RPC 框架》这一节，我们就提到了传输协议的作用。简单来说：通过设计协议，我们定义需要传输哪些类型的数据， 并且还会规定每一种类型的数据应该占多少字节。这样我们在接收到二级制数据之后，就可以正确的解析出我们需要的数据。这有一点像密文传输的感觉。以下便是我们设计的传输协议（编解码器这里会用到！！！）： * 0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 * +-----+-----+-----+-----+--------+----+----+----+------+-----------+-------+----- --+-----+-----+-------+ * | magic code |version | full length | messageType| codec|compress| RequestId | * +-----------------------+--------+---------------------+-----------+-----------+-----------+------------+ * | | * | body | * | | * | ... ... | * +-------------------------------------------------------------------------------------------------------+ * 4B magic code（魔法数） 1B version（版本） 4B full length（消息长度） 1B messageType（消息类型） * 1B compress（压缩类型） 1B codec（序列化类型） 4B requestId（请求的Id）魔法数 ： 通常是 4 个字节。这个魔数主要是为了筛选来到服务端的数据包，有了这个魔数之后，服务端首先取出前面四个字节进行比对，能够在第一时间识别出这个数据包并非是遵循自定义协议的，也就是无效数据包，为了安全考虑可以直接关闭连接以节省资源。序列化器类型 ：标识序列化的方式，比如是使用 Java 自带的序列化，还是 json，kyro 等序列化方式。消息长度 ： 运行时计算出来。......编解码器编解码器这里主要用到了 Kryo 序列化和反序列化以及 Netty 网络传输字节容器 ByteBuf 相关的知识。编解码器的作用主要是让我们在 Netty 进行网络传输所用的对象类型 ByteBuf 与 我们代码层面需要的业务对象之间转换。这部分的代码还是比较多的，小伙伴们可以自己阅读以下，整体逻辑还是比较简单的。一定要先搞懂传输协议之后再去看这部分代码。**RpcMessageDecoder.java**自定义解码器。负责处理"入站"消息，将 ByteBuf 消息格式的对象转换为我们需要的业务对象。网络传输需要通过字节流来实现，ByteBuf 可以看作是 Netty 提供的字节数据的容器，使用它会让我们更加方便地处理字节数据。**RpcMessageEncoder.java**自定义编码器。负责处理"出站"消息，将消息格式转换字节数组然后写入到字节数据的容器 ByteBuf 对象中。
AI 摘要 文章解析缓存雪崩、击穿、穿透：雪崩是批量失效或Redis宕机致数据库崩溃；击穿是热点key过期引发高并发查库；穿透是查询既不在缓存也不在数据库的无效数据，使数据库压力激增，并给出随机过期、互斥锁、布隆过滤器等应对方案。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结什么是缓存雪崩、击穿、穿透？用户的数据一般都是存储于数据库，数据库的数据是落在磁盘上的，磁盘的读写速度可以说是计算机里最慢的硬件了。当用户的请求，都访问数据库的话，请求数量一上来，数据库很容易就崩溃的了，所以为了避免用户直接访问数据库，会用 Redis 作为缓存层。因为 Redis 是内存数据库，我们可以将数据库的数据缓存在 Redis 里，相当于数据缓存在内存，内存的读写速度比硬盘快好几个数量级，这样大大提高了系统性能。引入了缓存层，就会有缓存异常的三个问题，分别是缓存雪崩、缓存击穿、缓存穿透。这三个问题也是面试中很常考察的问题，我们不光要清楚地知道它们是怎么发生，还需要知道如何解决它们。话不多说，发车！缓存雪崩通常我们为了保证缓存中的数据与数据库中的数据一致性，会给 Redis 里的数据设置过期时间，当缓存数据过期后，用户访问的数据如果不在缓存里，业务系统需要重新生成缓存，因此就会访问数据库，并将数据更新到 Redis 里，这样后续请求都可以直接命中缓存。那么，当大量缓存数据在同一时间过期（失效）或者 Redis 故障宕机时，如果此时有大量的用户请求，都无法在 Redis 中处理，于是全部请求都直接访问数据库，从而导致数据库的压力骤增，严重的会造成数据库宕机，从而形成一系列连锁反应，造成整个系统崩溃，这就是缓存雪崩的问题。可以看到，发生缓存雪崩有两个原因：大量数据同时过期；Redis 故障宕机；不同的诱因，应对的策略也会不同。大量数据同时过期针对大量数据同时过期而引发的缓存雪崩问题，常见的应对方法有下面这几种：均匀设置过期时间；互斥锁；双 key 策略；后台更新缓存；1. 均匀设置过期时间如果要给缓存数据设置过期时间，应该避免将大量的数据设置成同一个过期时间。我们可以在对缓存数据设置过期时间时，给这些数据的过期时间加上一个随机数，这样就保证数据不会在同一时间过期。2. 互斥锁当业务线程在处理用户请求时，如果发现访问的数据不在 Redis 里，就加个互斥锁，保证同一时间内只有一个请求来构建缓存（从数据库读取数据，再将数据更新到 Redis 里），当缓存构建完成后，再释放锁。未能获取互斥锁的请求，要么等待锁释放后重新读取缓存，要么就返回空值或者默认值。实现互斥锁的时候，最好设置超时时间，不然第一个请求拿到了锁，然后这个请求发生了某种意外而一直阻塞，一直不释放锁，这时其他请求也一直拿不到锁，整个系统就会出现无响应的现象。3. 双 key 策略我们对缓存数据可以使用两个 key，一个是主 key，会设置过期时间，一个是备 key，不会设置过期，它们只是 key 不一样，但是 value 值是一样的，相当于给缓存数据做了个副本。当业务线程访问不到「主 key」的缓存数据时，就直接返回「备 key」的缓存数据，然后在更新缓存的时候，同时更新「主 key」和「备 key」的数据。双 key 策略的好处是，当主 key 过期了，有大量请求获取缓存数据的时候，直接返回备 key 的数据，这样可以快速响应请求。而不用因为 key 失效而导致大量请求被锁阻塞住（采用了互斥锁，仅一个请求来构建缓存），后续再通知后台线程，重新构建主 key 的数据。4. 后台更新缓存业务线程不再负责更新缓存，缓存也不设置有效期，而是让缓存“永久有效”，并将更新缓存的工作交由后台线程定时更新。事实上，缓存数据不设置有效期，并不是意味着数据一直能在内存里，因为当系统内存紧张的时候，有些缓存数据会被“淘汰”，而在缓存被“淘汰”到下一次后台定时更新缓存的这段时间内，业务线程读取缓存失败就返回空值，业务的视角就以为是数据丢失了。解决上面的问题的方式有两种。第一种方式，后台线程不仅负责定时更新缓存，而且也负责频繁地检测缓存是否有效，检测到缓存失效了，原因可能是系统紧张而被淘汰的，于是就要马上从数据库读取数据，并更新到缓存。这种方式的检测时间间隔不能太长，太长也导致用户获取的数据是一个空值而不是真正的数据，所以检测的间隔最好是毫秒级的，但是总归是有个间隔时间，用户体验一般。第二种方式，在业务线程发现缓存数据失效后（缓存数据被淘汰），通过消息队列发送一条消息通知后台线程更新缓存，后台线程收到消息后，在更新缓存前可以判断缓存是否存在，存在就不执行更新缓存操作；不存在就读取数据库数据，并将数据加载到缓存。这种方式相比第一种方式缓存的更新会更及时，用户体验也比较好。在业务刚上线的时候，我们最好提前把数据缓存起来，而不是等待用户访问才来触发缓存构建，这就是所谓的缓存预热，后台更新缓存的机制刚好也适合干这个事情。Redis 故障宕机针对 Redis 故障宕机而引发的缓存雪崩问题，常见的应对方法有下面这几种：服务熔断或请求限流机制；构建 Redis 缓存高可靠集群；1. 服务熔断或请求限流机制因为 Redis 故障宕机而导致缓存雪崩问题时，我们可以启动服务熔断机制，暂停业务应用对缓存服务的访问，直接返回错误，不用再继续访问数据库，从而降低对数据库的访问压力，保证数据库系统的正常运行，然后等到 Redis 恢复正常后，再允许业务应用访问缓存服务。服务熔断机制是保护数据库的正常运行，但是暂停了业务应用访问缓存服系统，全部业务都无法正常工作为了减少对业务的影响，我们可以启用请求限流机制，只将少部分请求发送到数据库进行处理，再多的请求就在入口直接拒绝服务，等到 Redis 恢复正常并把缓存预热完后，再解除请求限流的机制。2. 构建 Redis 缓存高可靠集群服务熔断或请求限流机制是缓存雪崩发生后的应对方案，我们最好通过主从节点的方式构建 Redis 缓存高可靠集群。如果 Redis 缓存的主节点故障宕机，从节点可以切换成为主节点，继续提供缓存服务，避免了由于 Redis 故障宕机而导致的缓存雪崩问题。缓存击穿我们的业务通常会有几个数据会被频繁地访问，比如秒杀活动，这类被频繁访问的数据被称为热点数据。如果缓存中的某个热点数据过期了，此时大量的请求访问了该热点数据，就无法从缓存中读取，直接访问数据库，数据库很容易就被高并发的请求冲垮，这就是缓存击穿的问题。可以发现缓存击穿跟缓存雪崩很相似，你可以认为缓存击穿是缓存雪崩的一个子集。应对缓存击穿可以采取前面说到两种方案：互斥锁方案，保证同一时间只有一个业务线程更新缓存，未能获取互斥锁的请求，要么等待锁释放后重新读取缓存，要么就返回空值或者默认值。不给热点数据设置过期时间，由后台异步更新缓存，或者在热点数据准备要过期前，提前通知后台线程更新缓存以及重新设置过期时间；缓存穿透当发生缓存雪崩或击穿时，数据库中还是保存了应用要访问的数据，一旦缓存恢复相对应的数据，就可以减轻数据库的压力，而缓存穿透就不一样了。当用户访问的数据，既不在缓存中，也不在数据库中，导致请求在访问缓存时，发现缓存缺失，再去访问数据库时，发现数据库中也没有要访问的数据，没办法构建缓存数据，来服务后续的请求。那么当有大量这样的请求到来时，数据库的压力骤增，这就是缓存穿透的问题。缓存穿透的发生一般有这两种情况：业务误操作，缓存中的数据和数据库中的数据都被误删除了，所以导致缓存和数据库中都没有数据；黑客恶意攻击，故意大量访问某些读取不存在数据的业务；应对缓存穿透的方案，常见的方案有三种。第一种方案，非法请求的限制；第二种方案，缓存空值或者默认值；第三种方案，使用布隆过滤器快速判断数据是否存在，避免通过查询数据库来判断数据是否存在；第一种方案，非法请求的限制当有大量恶意请求访问不存在的数据的时候，也会发生缓存穿透，因此在 API 入口处我们要判断出请求参数是否合理，请求参数是否含有非法值、请求字段是否存在，如果判断出是恶意请求就直接返回错误，避免进一步访问缓存和数据库。第二种方案，缓存空值或者默认值当我们线上业务发现缓存穿透的现象时，可以针对查询的数据，在缓存中设置一个空值或者默认值，这样后续请求就可以从缓存中读取到空值或者默认值，返回给应用，而不会继续查询数据库。第三种方案，使用布隆过滤器快速判断数据是否存在，避免通过查询数据库来判断数据是否存在。我们可以在写入数据库数据时，使用布隆过滤器做个标记，然后在用户请求到来时，业务线程确认缓存失效后，可以通过查询布隆过滤器快速判断数据是否存在，如果不存在，就不用通过查询数据库来判断数据是否存在。即使发生了缓存穿透，大量请求只会查询 Redis 和布隆过滤器，而不会查询数据库，保证了数据库能正常运行，Redis 自身也是支持布隆过滤器的。那问题来了，布隆过滤器是如何工作的呢？接下来，我介绍下。布隆过滤器由「初始值都为 0 的位图数组」和「N 个哈希函数」两部分组成。当我们在写入数据库数据时，在布隆过滤器里做个标记，这样下次查询数据是否在数据库时，只需要查询布隆过滤器，如果查询到数据没有被标记，说明不在数据库中。布隆过滤器会通过 3 个操作完成标记：第一步，使用 N 个哈希函数分别对数据做哈希计算，得到 N 个哈希值；第二步，将第一步得到的 N 个哈希值对位图数组的长度取模，得到每个哈希值在位图数组的对应位置。第三步，将每个哈希值在位图数组的对应位置的值设置为 1；举个例子，假设有一个位图数组长度为 8，哈希函数 3 个的布隆过滤器。在数据库写入数据 x 后，把数据 x 标记在布隆过滤器时，数据 x 会被 3 个哈希函数分别计算出 3 个哈希值，然后在对这 3 个哈希值对 8 取模，假设取模的结果为 1、4、6，然后把位图数组的第 1、4、6 位置的值设置为 1。当应用要查询数据 x 是否数据库时，通过布隆过滤器只要查到位图数组的第 1、4、6 位置的值是否全为 1，只要有一个为 0，就认为数据 x 不在数据库中。布隆过滤器由于是基于哈希函数实现查找的，高效查找的同时存在哈希冲突的可能性，比如数据 x 和数据 y 可能都落在第 1、4、6 位置，而事实上，可能数据库中并不存在数据 y，存在误判的情况。所以，查询布隆过滤器说数据存在，并不一定证明数据库中存在这个数据，但是查询到数据不存在，数据库中一定就不存在这个数据。总结缓存异常会面临的三个问题：缓存雪崩、击穿和穿透。其中，缓存雪崩和缓存击穿主要原因是数据不在缓存中，而导致大量请求访问了数据库，数据库压力骤增，容易引发一系列连锁反应，导致系统奔溃。不过，一旦数据被重新加载回缓存，应用又可以从缓存快速读取数据，不再继续访问数据库，数据库的压力也会瞬间降下来。因此，缓存雪崩和缓存击穿应对的方案比较类似。而缓存穿透主要原因是数据既不在缓存也不在数据库中。因此，缓存穿透与缓存雪崩、击穿应对的方案不太一样。我这里整理了表格，你可以从下面这张表格很好的知道缓存雪崩、击穿和穿透的区别以及应对方案。
AI 摘要 文章演示了Docker安装ZooKeeper及常用命令（create/set/get/ls/stat/delete），并详解Curator连接、节点增删改查、监听器的代码实现，覆盖持久/临时/顺序节点与重试策略。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结08 ZooKeeper常用命令+ Curator使用详解guide-rpc-framework 使用了 Zookeeper 来存储服务的相关信息 ，并且使用的是 ZooKeeper Java客户端 Curator 来对 ZooKeeper 进行增删改查等操作。所以，本文就简单介绍一下 ZooKeeper常用命令 以及 Curator 的基本使用。前言这篇文章简单给演示一下 ZooKeeper 常见命令的使用以及 ZooKeeper Java客户端 Curator 的基本使用。介绍到的内容都是最基本的操作，能满足日常工作的基本需要。如果文章有任何需要改善和完善的地方，欢迎在评论区指出，共同进步！ZooKeeper 安装和使用使用Docker 安装 zookeepera.使用 Docker 下载 ZooKeeperdocker pull zookeeper:3.5.8b.运行 ZooKeeperdocker run -d --name zookeeper -p 2181:2181 zookeeper:3.5.8连接 ZooKeeper 服务a.进入ZooKeeper容器中先使用 docker ps 查看 ZooKeeper 的 ContainerID，然后使用 docker exec -it ContainerID /bin/bash 命令进入容器中。b.先进入 bin 目录,然后通过 **./zkCli.sh -server 127.0.0.1:2181**命令连接ZooKeeper 服务root@eaf70fc620cb:/apache-zookeeper-3.5.8-bin# cd bin如果你看到控制台成功打印出如下信息的话，说明你已经成功连接 ZooKeeper 服务。常用命令演示查看常用命令(help 命令)通过 help 命令查看 ZooKeeper 常用命令创建节点(create 命令)通过 create 命令在根目录创建了 node1 节点，与它关联的字符串是"node1"[zk: 127.0.0.1:2181(CONNECTED) 34] create /node1 “node1”通过 create 命令在根目录创建了 node1 节点，与它关联的内容是数字 123[zk: 127.0.0.1:2181(CONNECTED) 1] create /node1/node1.1 123 Created /node1/node1.1更新节点数据内容(set 命令)[zk: 127.0.0.1:2181(CONNECTED) 11] set /node1 "set node1"获取节点的数据(get 命令)get 命令可以获取指定节点的数据内容和节点的状态,可以看出我们通过 set 命令已经将节点数据内容改为 "set node1"。set node1 cZxid = 0x47 ctime = Sun Jan 20 10:22:59 CST 2019 mZxid = 0x4b mtime = Sun Jan 20 10:41:10 CST 2019 pZxid = 0x4a cversion = 1 dataVersion = 1 aclVersion = 0 ephemeralOwner = 0x0 dataLength = 9 numChildren = 1查看某个目录下的子节点(ls 命令)通过 ls 命令查看根目录下的节点[zk: 127.0.0.1:2181(CONNECTED) 37] ls / [dubbo, ZooKeeper, node1]通过 ls 命令查看 node1 目录下的节点[zk: 127.0.0.1:2181(CONNECTED) 5] ls /node1 [node1.1]ZooKeeper 中的 ls 命令和 linux 命令中的 ls 类似， 这个命令将列出绝对路径 path 下的所有子节点信息（列出 1 级，并不递归）查看节点状态(stat 命令)通过 stat 命令查看节点状态[zk: 127.0.0.1:2181(CONNECTED) 10] stat /node1 cZxid = 0x47 ctime = Sun Jan 20 10:22:59 CST 2019 mZxid = 0x47 mtime = Sun Jan 20 10:22:59 CST 2019 pZxid = 0x4a cversion = 1 dataVersion = 0 aclVersion = 0 ephemeralOwner = 0x0 dataLength = 11 numChildren = 1上面显示的一些信息比如 cversion、aclVersion、numChildren 等等，我在上面 “ZooKeeper 相关概念总结(入门)” 这篇文章中已经介绍到。查看节点信息和状态(ls2 命令)ls2 命令更像是 ls 命令和 stat 命令的结合。 ls2 命令返回的信息包括 2 部分：子节点列表当前节点的 stat 信息。[zk: 127.0.0.1:2181(CONNECTED) 7] ls2 /node1 [node1.1] cZxid = 0x47 ctime = Sun Jan 20 10:22:59 CST 2019 mZxid = 0x47 mtime = Sun Jan 20 10:22:59 CST 2019 pZxid = 0x4a cversion = 1 dataVersion = 0 aclVersion = 0 ephemeralOwner = 0x0 dataLength = 11 numChildren = 1删除节点(delete 命令)这个命令很简单，但是需要注意的一点是如果你要删除某一个节点，那么这个节点必须无子节点才行。[zk: 127.0.0.1:2181(CONNECTED) 3] delete /node1/node1.1在后面我会介绍到 Java 客户端 API 的使用以及开源 ZooKeeper 客户端 ZkClient 和 Curator 的使用。ZooKeeper Java客户端 Curator简单使用Curator 是Netflix公司开源的一套 ZooKeeper Java客户端框架，相比于 Zookeeper 自带的客户端 zookeeper 来说，Curator 的封装更加完善，各种 API 都可以比较方便地使用。下面我们就来简单地演示一下 Curator 的使用吧！Curator4.0+版本对ZooKeeper 3.5.x支持比较好。开始之前，请先将下面的依赖添加进你的项目。<dependency> <groupId>org.apache.curator</groupId> <artifactId>curator-framework</artifactId> <version>4.2.0</version> </dependency> <dependency> <groupId>org.apache.curator</groupId> <artifactId>curator-recipes</artifactId> <version>4.2.0</version> </dependency>连接 ZooKeeper 客户端通过 CuratorFrameworkFactory 创建 CuratorFramework 对象，然后再调用 CuratorFramework 对象的 start() 方法即可！private static final int BASE_SLEEP_TIME = 1000; private static final int MAX_RETRIES = 3; // Retry strategy. Retry 3 times, and will increase the sleep time between retries. RetryPolicy retryPolicy = new ExponentialBackoffRetry(BASE_SLEEP_TIME, MAX_RETRIES); CuratorFramework zkClient = CuratorFrameworkFactory.builder() // the server to connect to (can be a server list) .connectString("127.0.0.1:2181") .retryPolicy(retryPolicy) .build(); zkClient.start();对于一些基本参数的说明：baseSleepTimeMs：重试之间等待的初始时间maxRetries ：最大重试次数connectString ：要连接的服务器列表retryPolicy ：重试策略数据节点的增删改查创建节点我们在 ZooKeeper常见概念解读 中介绍到，我们通常是将 znode 分为 4 大类：持久（PERSISTENT）节点 ：一旦创建就一直存在即使 ZooKeeper 集群宕机，直到将其删除。临时（EPHEMERAL）节点 ：临时节点的生命周期是与 客户端会话（session） 绑定的，会话消失则节点消失 。并且，临时节点 只能做叶子节点 ，不能创建子节点。持久顺序（PERSISTENT_SEQUENTIAL）节点 ：除了具有持久（PERSISTENT）节点的特性之外， 子节点的名称还具有顺序性。比如 /node1/app0000000001 、/node1/app0000000002 。临时顺序（EPHEMERAL_SEQUENTIAL）节点 ：除了具备临时（EPHEMERAL）节点的特性之外，子节点的名称还具有顺序性。你在使用的 ZooKeeper 的时候，会发现 CreateMode 类中实际有 7种 znode 类型 ，但是用的最多的还是上面介绍的 4 种。a.创建持久化节点你可以通过下面两种方式创建持久化的节点。//注意:下面的代码会报错，下文说了具体原因 zkClient.create().forPath("/node1/00001"); zkClient.create().withMode(CreateMode.PERSISTENT).forPath("/node1/00002");但是，你运行上面的代码会报错，这是因为的父节点node1还未创建。你可以先创建父节点 node1 ，然后再执行上面的代码就不会报错了。zkClient.create().forPath("/node1");更推荐的方式是通过下面这行代码， **creatingParentsIfNeeded()** 可以保证父节点不存在的时候自动创建父节点，这是非常有用的。zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/node1/00001");b.创建临时节点zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath("/node1/00001");c.创建节点并指定数据内容zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath("/node1/00001","java".getBytes()); zkClient.getData().forPath("/node1/00001");//获取节点的数据内容，获取到的是 byte数组d.检测节点是否创建成功zkClient.checkExists().forPath("/node1/00001");//不为null的话，说明节点创建成功删除节点a.删除一个子节点zkClient.delete().forPath("/node1/00001");b.删除一个节点以及其下的所有子节点zkClient.delete().deletingChildrenIfNeeded().forPath("/node1");获取/更新节点数据内容zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath("/node1/00001","java".getBytes()); zkClient.getData().forPath("/node1/00001");//获取节点的数据内容 zkClient.setData().forPath("/node1/00001","c++".getBytes());//更新节点数据内容获取某个节点的所有子节点路径List<String> childrenPaths = zkClient.getChildren().forPath("/node1");监听器下面简单演示一下如何给某个节点注册子节点监听器 。注册了监听器之后，这个节点的子节点发生变化比如增加、减少或者更新的时候，你可以自定义回调操作。String path = "/node1"; PathChildrenCache pathChildrenCache = new PathChildrenCache(zkClient, path, true); PathChildrenCacheListener pathChildrenCacheListener = (curatorFramework, pathChildrenCacheEvent) -> { // do something }; pathChildrenCache.getListenable().addListener(pathChildrenCacheListener); pathChildrenCache.start();如果你要获取节点事件类型的话，可以通过：pathChildrenCacheEvent.getType()一共有下面几种类型： public static enum Type { CHILD_ADDED,//子节点增加 CHILD_UPDATED,//子节点更新 CHILD_REMOVED,//子节点被删除 CONNECTION_SUSPENDED, CONNECTION_RECONNECTED, CONNECTION_LOST, INITIALIZED; private Type() { } }
AI 摘要 一致性哈希将节点与数据映射到2³²环，顺时针寻址，扩容缩容仅影响相邻节点，减少数据迁移；引入虚拟节点解决分布不均，提升均衡与稳定性。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结如何分配请求？大多数网站背后肯定不是只有一台服务器提供服务，因为单机的并发量和数据量都是有限的，所以都会用多台服务器构成集群来对外提供服务。但是问题来了，现在有那么多个节点（后面统称服务器为节点，因为少一个字），要如何分配客户端的请求呢？其实这个问题就是「负载均衡问题」。解决负载均衡问题的算法很多，不同的负载均衡算法，对应的就是不同的分配策略，适应的业务场景也不同。最简单的方式，引入一个中间的负载均衡层，让它将外界的请求「轮流」的转发给内部的集群。比如集群有 3 个节点，外界请求有 3 个，那么每个节点都会处理 1 个请求，达到了分配请求的目的。考虑到每个节点的硬件配置有所区别，我们可以引入权重值，将硬件配置更好的节点的权重值设高，然后根据各个节点的权重值，按照一定比重分配在不同的节点上，让硬件配置更好的节点承担更多的请求，这种算法叫做加权轮询。加权轮询算法使用场景是建立在每个节点存储的数据都是相同的前提。所以，每次读数据的请求，访问任意一个节点都能得到结果。但是，加权轮询算法是无法应对「分布式系统」的，因为分布式系统中，每个节点存储的数据是不同的。当我们想提高系统的容量，就会将数据水平切分到不同的节点来存储，也就是将数据分布到了不同的节点。比如一个分布式 KV（key-valu）缓存系统，某个 key 应该到哪个或者哪些节点上获得，应该是确定的，不是说任意访问一个节点都可以得到缓存结果的。因此，我们要想一个能应对分布式系统的负载均衡算法。使用哈希算法有什么问题？有的同学可能很快就想到了：哈希算法。因为对同一个关键字进行哈希计算，每次计算都是相同的值，这样就可以将某个 key 确定到一个节点了，可以满足分布式系统的负载均衡需求。哈希算法最简单的做法就是进行取模运算，比如分布式系统中有 3 个节点，基于 hash(key) % 3 公式对数据进行了映射。如果客户端要获取指定 key 的数据，通过下面的公式可以定位节点：hash(key) % 3如果经过上面这个公式计算后得到的值是 0，就说明该 key 需要去第一个节点获取。但是有一个很致命的问题，如果节点数量发生了变化，也就是在对系统做扩容或者缩容时，必须迁移改变了映射关系的数据，否则会出现查询不到数据的问题。举个例子，假设我们有一个由 A、B、C 三个节点组成分布式 KV 缓存系统，基于计算公式 hash(key) % 3 将数据进行了映射，每个节点存储了不同的数据：现在有 3 个查询 key 的请求，分别查询 key-01，key-02，key-03 的数据，这三个 key 分别经过 hash() 函数计算后的值为 hash( key-01) = 6、hash( key-02) = 7、hash(key-03) = 8，然后再对这些值进行取模运算。通过这样的哈希算法，每个 key 都可以定位到对应的节点。当 3 个节点不能满足业务需求了，这时我们增加了一个节点，节点的数量从 3 变化为 4，意味取模哈希函数中基数的变化，这样会导致大部分映射关系改变，如下图：比如，之前的 hash(key-01) % 3 = 0，就变成了 hash(key-01) % 4 = 2，查询 key-01 数据时，寻址到了节点 C，而 key-01 的数据是存储在节点 A 上的，不是在节点 C，所以会查询不到数据。同样的道理，如果我们对分布式系统进行缩容，比如移除一个节点，也会因为取模哈希函数中基数的变化，可能出现查询不到数据的问题。要解决这个问题的办法，就需要我们进行迁移数据，比如节点的数量从 3 变化为 4 时，要基于新的计算公式 hash(key) % 4，重新对数据和节点做映射。假设总数据条数为 M，哈希算法在面对节点数量变化时，最坏情况下所有数据都需要迁移，所以它的数据迁移规模是 O(M)，这样数据的迁移成本太高了。所以，我们应该要重新想一个新的算法，来避免分布式系统在扩容或者缩容时，发生过多的数据迁移。使用一致性哈希算法有什么问题？一致性哈希算法就很好地解决了分布式系统在扩容或者缩容时，发生过多的数据迁移的问题。一致哈希算法也用了取模运算，但与哈希算法不同的是，哈希算法是对节点的数量进行取模运算，而一致哈希算法是对 2^32 进行取模运算，是一个固定的值。我们可以把一致哈希算法是对 2^32 进行取模运算的结果值组织成一个圆环，就像钟表一样，钟表的圆可以理解成由 60 个点组成的圆，而此处我们把这个圆想象成由 2^32 个点组成的圆，这个圆环被称为哈希环，如下图：一致性哈希要进行两步哈希：第一步：对存储节点进行哈希计算，也就是对存储节点做哈希映射，比如根据节点的 IP 地址进行哈希；第二步：当对数据进行存储或访问时，对数据进行哈希映射；所以，一致性哈希是指将「存储节点」和「数据」都映射到一个首尾相连的哈希环上。问题来了，对「数据」进行哈希映射得到一个结果要怎么找到存储该数据的节点呢？答案是，映射的结果值往顺时针的方向的找到第一个节点，就是存储该数据的节点。举个例子，有 3 个节点经过哈希计算，映射到了如下图的位置：接着，对要查询的 key-01 进行哈希计算，确定此 key-01 映射在哈希环的位置，然后从这个位置往顺时针的方向找到第一个节点，就是存储该 key-01 数据的节点。比如，下图中的 key-01 映射的位置，往顺时针的方向找到第一个节点就是节点 A。所以，当需要对指定 key 的值进行读写的时候，要通过下面 2 步进行寻址：首先，对 key 进行哈希计算，确定此 key 在环上的位置；然后，从这个位置沿着顺时针方向走，遇到的第一节点就是存储 key 的节点。知道了一致哈希寻址的方式，我们来看看，如果增加一个节点或者减少一个节点会发生大量的数据迁移吗？假设节点数量从 3 增加到了 4，新的节点 D 经过哈希计算后映射到了下图中的位置：你可以看到，key-01、key-03 都不受影响，只有 key-02 需要被迁移节点 D。假设节点数量从 3 减少到了 2，比如将节点 A 移除：你可以看到，key-02 和 key-03 不会受到影响，只有 key-01 需要被迁移节点 B。因此，在一致哈希算法中，如果增加或者移除一个节点，仅影响该节点在哈希环上顺时针相邻的后继节点，其它数据也不会受到影响。上面这些图中 3 个节点映射在哈希环还是比较分散的，所以看起来请求都会「均衡」到每个节点。但是一致性哈希算法并不保证节点能够在哈希环上分布均匀，这样就会带来一个问题，会有大量的请求集中在一个节点上。比如，下图中 3 个节点的映射位置都在哈希环的右半边：这时候有一半以上的数据的寻址都会找节点 A，也就是访问请求主要集中的节点 A 上，这肯定不行的呀，说好的负载均衡呢，这种情况一点都不均衡。另外，在这种节点分布不均匀的情况下，进行容灾与扩容时，哈希环上的相邻节点容易受到过大影响，容易发生雪崩式的连锁反应。比如，上图中如果节点 A 被移除了，当节点 A 宕机后，根据一致性哈希算法的规则，其上数据应该全部迁移到相邻的节点 B 上，这样，节点 B 的数据量、访问量都会迅速增加很多倍，一旦新增的压力超过了节点 B 的处理能力上限，就会导致节点 B 崩溃，进而形成雪崩式的连锁反应。所以，一致性哈希算法虽然减少了数据迁移量，但是存在节点分布不均匀的问题。如何通过虚拟节点提高均衡度？要想解决节点能在哈希环上分配不均匀的问题，就是要有大量的节点，节点数越多，哈希环上的节点分布的就越均匀。但问题是，实际中我们没有那么多节点。所以这个时候我们就加入虚拟节点，也就是对一个真实节点做多个副本。具体做法是，不再将真实节点映射到哈希环上，而是将虚拟节点映射到哈希环上，并将虚拟节点映射到实际节点，所以这里有「两层」映射关系。比如对每个节点分别设置 3 个虚拟节点：对节点 A 加上编号来作为虚拟节点：A-01、A-02、A-03对节点 B 加上编号来作为虚拟节点：B-01、B-02、B-03对节点 C 加上编号来作为虚拟节点：C-01、C-02、C-03引入虚拟节点后，原本哈希环上只有 3 个节点的情况，就会变成有 9 个虚拟节点映射到哈希环上，哈希环上的节点数量多了 3 倍。你可以看到，节点数量多了后，节点在哈希环上的分布就相对均匀了。这时候，如果有访问请求寻址到「A-01」这个虚拟节点，接着再通过「A-01」虚拟节点找到真实节点 A，这样请求就能访问到真实节点 A 了。上面为了方便你理解，每个真实节点仅包含 3 个虚拟节点，这样能起到的均衡效果其实很有限。而在实际的工程中，虚拟节点的数量会大很多，比如 Nginx 的一致性哈希算法，每个权重为 1 的真实节点就含有 160 个虚拟节点。另外，虚拟节点除了会提高节点的均衡度，还会提高系统的稳定性。当节点变化时，会有不同的节点共同分担系统的变化，因此稳定性更高。比如，当某个节点被移除时，对应该节点的多个虚拟节点均会移除，而这些虚拟节点按顺时针方向的下一个虚拟节点，可能会对应不同的真实节点，即这些不同的真实节点共同分担了节点变化导致的压力。而且，有了虚拟节点后，还可以为硬件配置更好的节点增加权重，比如对权重更高的节点增加更多的虚拟机节点即可。因此，带虚拟节点的一致性哈希方法不仅适合硬件配置不同的节点的场景，而且适合节点规模会发生变化的场景。总结不同的负载均衡算法适用的业务场景也不同的。轮询这类的策略只能适用与每个节点的数据都是相同的场景，访问任意节点都能请求到数据。但是不适用分布式系统，因为分布式系统意味着数据水平切分到了不同的节点上，访问数据的时候，一定要寻址存储该数据的节点。哈希算法虽然能建立数据和节点的映射关系，但是每次在节点数量发生变化的时候，最坏情况下所有数据都需要迁移，这样太麻烦了，所以不适用节点数量变化的场景。为了减少迁移的数据量，就出现了一致性哈希算法。一致性哈希是指将「存储节点」和「数据」都映射到一个首尾相连的哈希环上，如果增加或者移除一个节点，仅影响该节点在哈希环上顺时针相邻的后继节点，其它数据也不会受到影响。但是一致性哈希算法不能够均匀的分布节点，会出现大量请求都集中在一个节点的情况，在这种情况下进行容灾与扩容时，容易出现雪崩的连锁反应。为了解决一致性哈希算法不能够均匀的分布节点的问题，就需要引入虚拟节点，对一个真实节点做多个副本。不再将真实节点映射到哈希环上，而是将虚拟节点映射到哈希环上，并将虚拟节点映射到实际节点，所以这里有「两层」映射关系。引入虚拟节点后，可以会提高节点的均衡度，还会提高系统的稳定性。所以，带虚拟节点的一致性哈希方法不仅适合硬件配置不同的节点的场景，而且适合节点规模会发生变化的场景。完！
AI 摘要 文章系统梳理分布式消息队列基础：定义、生产者-消费者模型，三大核心价值（异步提速、削峰限流、解耦），并指出可用性下降、复杂度上升、一致性风险等代价；对比JMS/AMQP协议差异，RPC与MQ场景定位，最后盘点Kafka、RocketMQ、RabbitMQ、Pulsar等主流产品选型要点。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结title: 消息队列基础知识总结category: 高性能tag:消息队列::: tip这篇文章中的消息队列主要指的是分布式消息队列。:::“RabbitMQ？”“Kafka？”“RocketMQ？”...在日常学习与开发过程中，我们常常听到消息队列这个关键词。我也在我的多篇文章中提到了这个概念。可能你是熟练使用消息队列的老手，又或者你是不懂消息队列的新手，不论你了不了解消息队列，本文都将带你搞懂消息队列的一些基本理论。如果你是老手，你可能从本文学到你之前不曾注意的一些关于消息队列的重要概念，如果你是新手，相信本文将是你打开消息队列大门的一板砖。什么是消息队列？我们可以把消息队列看作是一个存放消息的容器，当我们需要使用消息的时候，直接从容器中取出消息供自己使用即可。由于队列 Queue 是一种先进先出的数据结构，所以消费消息时也是按照顺序来消费的。参与消息传递的双方称为 生产者 和 消费者 ，生产者负责发送消息，消费者负责处理消息。我们知道操作系统中的进程通信的一种很重要的方式就是消息队列。我们这里提到的消息队列稍微有点区别，更多指的是各个服务以及系统内部各个组件/模块之前的通信，属于一种 中间件 。维基百科是这样介绍中间件的：中间件（英语：Middleware），又译中间件、中介层，是一类提供系统软件和应用软件之间连接、便于软件各部件之间的沟通的软件，应用软件可以借助中间件在不同的技术架构之间共享信息与资源。中间件位于客户机服务器的操作系统之上，管理着计算资源和网络通信。简单来说：中间件就是一类为应用软件服务的软件，应用软件是为用户服务的，用户不会接触或者使用到中间件。除了消息队列之外，常见的中间件还有 RPC 框架、分布式组件、HTTP 服务器、任务调度框架、配置中心、数据库层的分库分表工具和数据迁移工具等等。关于中间件比较详细的介绍可以参考阿里巴巴淘系技术的一篇回答：https://www.zhihu.com/question/19730582/answer/1663627873 。随着分布式和微服务系统的发展，消息队列在系统设计中有了更大的发挥空间，使用消息队列可以降低系统耦合性、实现任务异步、有效地进行流量削峰，是分布式和微服务系统中重要的组件之一。消息队列有什么用？通常来说，使用消息队列能为我们的系统带来下面三点好处：通过异步处理提高系统性能（减少响应所需时间）削峰/限流降低系统耦合性。如果在面试的时候你被面试官问到这个问题的话，一般情况是你在你的简历上涉及到消息队列这方面的内容，这个时候推荐你结合你自己的项目来回答。通过异步处理提高系统性能（减少响应所需时间）将用户的请求数据存储到消息队列之后就立即返回结果。随后，系统再对消息进行消费。因为用户请求数据写入消息队列之后就立即返回给用户了，但是请求数据在后续的业务校验、写数据库等操作中可能失败。因此，使用消息队列进行异步处理之后，需要适当修改业务流程进行配合，比如用户在提交订单之后，订单数据写入消息队列，不能立即返回用户订单提交成功，需要在消息队列的订单消费者进程真正处理完该订单之后，甚至出库后，再通过电子邮件或短信通知用户订单成功，以免交易纠纷。这就类似我们平时手机订火车票和电影票。削峰/限流先将短时间高并发产生的事务消息存储在消息队列中，然后后端服务再慢慢根据自己的能力去消费这些消息，这样就避免直接把后端服务打垮掉。举例：在电子商务一些秒杀、促销活动中，合理使用消息队列可以有效抵御促销活动刚开始大量订单涌入对系统的冲击。如下图所示：降低系统耦合性使用消息队列还可以降低系统耦合性。我们知道如果模块之间不存在直接调用，那么新增模块或者修改模块就对其他模块影响较小，这样系统的可扩展性无疑更好一些。还是直接上图吧：生产者（客户端）发送消息到消息队列中去，接受者（服务端）处理消息，需要消费的系统直接去消息队列取消息进行消费即可而不需要和其他系统有耦合，这显然也提高了系统的扩展性。消息队列使用发布-订阅模式工作，消息发送者（生产者）发布消息，一个或多个消息接受者（消费者）订阅消息。 从上图可以看到消息发送者（生产者）和消息接受者（消费者）之间没有直接耦合，消息发送者将消息发送至分布式消息队列即结束对消息的处理，消息接受者从分布式消息队列获取该消息后进行后续处理，并不需要知道该消息从何而来。对新增业务，只要对该类消息感兴趣，即可订阅该消息，对原有系统和业务没有任何影响，从而实现网站业务的可扩展性设计。例如，我们商城系统分为用户、订单、财务、仓储、消息通知、物流、风控等多个服务。用户在完成下单后，需要调用财务（扣款）、仓储（库存管理）、物流（发货）、消息通知（通知用户发货）、风控（风险评估）等服务。使用消息队列后，下单操作和后续的扣款、发货、通知等操作就解耦了，下单完成发送一个消息到消息队列，需要用到的地方去订阅这个消息进行消息即可。另外，为了避免消息队列服务器宕机造成消息丢失，会将成功发送到消息队列的消息存储在消息生产者服务器上，等消息真正被消费者服务器处理后才删除消息。在消息队列服务器宕机后，生产者服务器会选择分布式消息队列服务器集群中的其他服务器发布消息。备注： 不要认为消息队列只能利用发布-订阅模式工作，只不过在解耦这个特定业务环境下是使用发布-订阅模式的。除了发布-订阅模式，还有点对点订阅模式（一个消息只有一个消费者），我们比较常用的是发布-订阅模式。另外，这两种消息模型是 JMS 提供的，AMQP 协议还提供了另外 5 种消息模型。实现分布式事务我们知道分布式事务的解决方案之一就是 MQ 事务。RocketMQ、 Kafka、Pulsar、QMQ 都提供了事务相关的功能。事务允许事件流应用将消费，处理，生产消息整个过程定义为一个原子操作。详细介绍可以查看 分布式事务详解(付费) 这篇文章。使用消息队列会带来哪些问题？系统可用性降低： 系统可用性在某种程度上降低，为什么这样说呢？在加入 MQ 之前，你不用考虑消息丢失或者说 MQ 挂掉等等的情况，但是，引入 MQ 之后你就需要去考虑了！系统复杂性提高： 加入 MQ 之后，你需要保证消息没有被重复消费、处理消息丢失的情况、保证消息传递的顺序性等等问题！一致性问题： 我上面讲了消息队列可以实现异步，消息队列带来的异步确实可以提高系统响应速度。但是，万一消息的真正消费者并没有正确消费消息怎么办？这样就会导致数据不一致的情况了!JMS 和 AMQPJMS 是什么？JMS（JAVA Message Service,java 消息服务）是 Java 的消息服务，JMS 的客户端之间可以通过 JMS 服务进行异步的消息传输。JMS（JAVA Message Service，Java 消息服务）API 是一个消息服务的标准或者说是规范，允许应用程序组件基于 JavaEE 平台创建、发送、接收和读取消息。它使分布式通信耦合度更低，消息服务更加可靠以及异步性。JMS 定义了五种不同的消息正文格式以及调用的消息类型，允许你发送并接收以一些不同形式的数据：StreamMessage：Java 原始值的数据流MapMessage：一套名称-值对TextMessage：一个字符串对象ObjectMessage：一个序列化的 Java 对象BytesMessage：一个字节的数据流ActiveMQ（已被淘汰） 就是基于 JMS 规范实现的。JMS 两种消息模型点到点（P2P）模型使用队列（Queue）作为消息通信载体；满足生产者与消费者模式，一条消息只能被一个消费者使用，未被消费的消息在队列中保留直到被消费或超时。比如：我们生产者发送 100 条消息的话，两个消费者来消费一般情况下两个消费者会按照消息发送的顺序各自消费一半（也就是你一个我一个的消费。）发布/订阅（Pub/Sub）模型发布订阅模型（Pub/Sub） 使用主题（Topic）作为消息通信载体，类似于广播模式；发布者发布一条消息，该消息通过主题传递给所有的订阅者。AMQP 是什么？AMQP，即 Advanced Message Queuing Protocol，一个提供统一消息服务的应用层标准 高级消息队列协议（二进制应用层协议），是应用层协议的一个开放标准，为面向消息的中间件设计，兼容 JMS。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件同产品，不同的开发语言等条件的限制。RabbitMQ 就是基于 AMQP 协议实现的。JMS vs AMQP对比方向JMSAMQP定义Java API协议跨语言否是跨平台否是支持消息类型提供两种消息模型：①Peer-2-Peer;②Pub/sub提供了五种消息模型：①direct exchange；②fanout exchange；③topic change；④headers exchange；⑤system exchange。本质来讲，后四种和 JMS 的 pub/sub 模型没有太大差别，仅是在路由机制上做了更详细的划分；支持消息类型支持多种消息类型 ，我们在上面提到过byte[]（二进制）总结：AMQP 为消息定义了线路层（wire-level protocol）的协议，而 JMS 所定义的是 API 规范。在 Java 体系中，多个 client 均可以通过 JMS 进行交互，不需要应用修改代码，但是其对跨平台的支持较差。而 AMQP 天然具有跨平台、跨语言特性。JMS 支持 TextMessage、MapMessage 等复杂的消息类型；而 AMQP 仅支持 byte[] 消息类型（复杂的类型可序列化后发送）。由于 Exchange 提供的路由算法，AMQP 可以提供多样化的路由方式来传递消息到消息队列，而 JMS 仅支持 队列 和 主题/订阅 方式两种。RPC 和消息队列的区别RPC 和消息队列都是分布式微服务系统中重要的组件之一，下面我们来简单对比一下两者：从用途来看：RPC 主要用来解决两个服务的远程通信问题，不需要了解底层网络的通信机制。通过 RPC 可以帮助我们调用远程计算机上某个服务的方法，这个过程就像调用本地方法一样简单。消息队列主要用来降低系统耦合性、实现任务异步、有效地进行流量削峰。从通信方式来看：RPC 是双向直接网络通讯，消息队列是单向引入中间载体的网络通讯。从架构上来看：消息队列需要把消息存储起来，RPC 则没有这个要求，因为前面也说了 RPC 是双向直接网络通讯。从请求处理的时效性来看：通过 RPC 发出的调用一般会立即被处理，存放在消息队列中的消息并不一定会立即被处理。RPC 和消息队列本质上是网络通讯的两种不同的实现机制，两者的用途不同，万不可将两者混为一谈。分布式消息队列技术选型常见的消息队列有哪些？KafkaKafka 是 LinkedIn 开源的一个分布式流式处理平台，已经成为 Apache 顶级项目，早期被用来用于处理海量的日志，后面才慢慢发展成了一款功能全面的高性能消息队列。流式处理平台具有三个关键功能：消息队列：发布和订阅消息流，这个功能类似于消息队列，这也是 Kafka 也被归类为消息队列的原因。容错的持久方式存储记录消息流：Kafka 会把消息持久化到磁盘，有效避免了消息丢失的风险。流式处理平台： 在消息发布的时候进行处理，Kafka 提供了一个完整的流式处理类库。Kafka 是一个分布式系统，由通过高性能 TCP 网络协议进行通信的服务器和客户端组成，可以部署在在本地和云环境中的裸机硬件、虚拟机和容器上。在 Kafka 2.8 之前，Kafka 最被大家诟病的就是其重度依赖于 Zookeeper 做元数据管理和集群的高可用。在 Kafka 2.8 之后，引入了基于 Raft 协议的 KRaft 模式，不再依赖 Zookeeper，大大简化了 Kafka 的架构，让你可以以一种轻量级的方式来使用 Kafka。不过，要提示一下：如果要使用 KRaft 模式的话，建议选择较高版本的 Kafka，因为这个功能还在持续完善优化中。Kafka 3.3.1 版本是第一个将 KRaft（Kafka Raft）共识协议标记为生产就绪的版本。Kafka 官网：http://kafka.apache.org/Kafka 更新记录（可以直观看到项目是否还在维护）：https://kafka.apache.org/downloadsRocketMQRocketMQ 是阿里开源的一款云原生“消息、事件、流”实时数据处理平台，借鉴了 Kafka，已经成为 Apache 顶级项目。RocketMQ 的核心特性（摘自 RocketMQ 官网）：云原生：生与云，长与云，无限弹性扩缩，K8s 友好高吞吐：万亿级吞吐保证，同时满足微服务与大数据场景。流处理：提供轻量、高扩展、高性能和丰富功能的流计算引擎。金融级：金融级的稳定性，广泛用于交易核心链路。架构极简：零外部依赖，Shared-nothing 架构。生态友好：无缝对接微服务、实时计算、数据湖等周边生态。根据官网介绍：Apache RocketMQ 自诞生以来，因其架构简单、业务功能丰富、具备极强可扩展性等特点被众多企业开发者以及云厂商广泛采用。历经十余年的大规模场景打磨，RocketMQ 已经成为业内共识的金融级可靠业务消息首选方案，被广泛应用于互联网、大数据、移动互联网、物联网等领域的业务场景。RocketMQ 官网：https://rocketmq.apache.org/ （文档很详细，推荐阅读）RocketMQ 更新记录（可以直观看到项目是否还在维护）：https://github.com/apache/rocketmq/releasesRabbitMQRabbitMQ 是采用 Erlang 语言实现 AMQP(Advanced Message Queuing Protocol，高级消息队列协议）的消息中间件，它最初起源于金融系统，用于在分布式系统中存储转发消息。RabbitMQ 发展到今天，被越来越多的人认可，这和它在易用性、扩展性、可靠性和高可用性等方面的卓著表现是分不开的。RabbitMQ 的具体特点可以概括为以下几点：可靠性： RabbitMQ 使用一些机制来保证消息的可靠性，如持久化、传输确认及发布确认等。灵活的路由： 在消息进入队列之前，通过交换器来路由消息。对于典型的路由功能，RabbitMQ 己经提供了一些内置的交换器来实现。针对更复杂的路由功能，可以将多个交换器绑定在一起，也可以通过插件机制来实现自己的交换器。这个后面会在我们讲 RabbitMQ 核心概念的时候详细介绍到。扩展性： 多个 RabbitMQ 节点可以组成一个集群，也可以根据实际业务情况动态地扩展集群中节点。高可用性： 队列可以在集群中的机器上设置镜像，使得在部分节点出现问题的情况下队列仍然可用。支持多种协议： RabbitMQ 除了原生支持 AMQP 协议，还支持 STOMP、MQTT 等多种消息中间件协议。多语言客户端： RabbitMQ 几乎支持所有常用语言，比如 Java、Python、Ruby、PHP、C#、JavaScript 等。易用的管理界面： RabbitMQ 提供了一个易用的用户界面，使得用户可以监控和管理消息、集群中的节点等。在安装 RabbitMQ 的时候会介绍到，安装好 RabbitMQ 就自带管理界面。插件机制： RabbitMQ 提供了许多插件，以实现从多方面进行扩展，当然也可以编写自己的插件。感觉这个有点类似 Dubbo 的 SPI 机制RabbitMQ 官网：https://www.rabbitmq.com/ 。RabbitMQ 更新记录（可以直观看到项目是否还在维护）：https://www.rabbitmq.com/news.htmlPulsarPulsar 是下一代云原生分布式消息流平台，最初由 Yahoo 开发 ，已经成为 Apache 顶级项目。Pulsar 集消息、存储、轻量化函数式计算为一体，采用计算与存储分离架构设计，支持多租户、持久化存储、多机房跨区域数据复制，具有强一致性、高吞吐、低延时及高可扩展性等流数据存储特性，被看作是云原生时代实时消息流传输、存储和计算最佳解决方案。Pulsar 的关键特性如下（摘自官网）：是下一代云原生分布式消息流平台。Pulsar 的单个实例原生支持多个集群，可跨机房在集群间无缝地完成消息复制。极低的发布延迟和端到端延迟。可无缝扩展到超过一百万个 topic。简单的客户端 API，支持 Java、Go、Python 和 C++。主题的多种订阅模式（独占、共享和故障转移）。通过 Apache BookKeeper 提供的持久化消息存储机制保证消息传递 。由轻量级的 serverless 计算框架 Pulsar Functions 实现流原生的数据处理。基于 Pulsar Functions 的 serverless connector 框架 Pulsar IO 使得数据更易移入、移出 Apache Pulsar。分层式存储可在数据陈旧时，将数据从热存储卸载到冷/长期存储（如 S3、GCS）中。Pulsar 官网：https://pulsar.apache.org/Pulsar 更新记录（可以直观看到项目是否还在维护）：https://github.com/apache/pulsar/releasesActiveMQ目前已经被淘汰，不推荐使用，不建议学习。如何选择？参考《Java 工程师面试突击第 1 季-中华石杉老师》对比方向概要吞吐量万级的 ActiveMQ 和 RabbitMQ 的吞吐量（ActiveMQ 的性能最差）要比十万级甚至是百万级的 RocketMQ 和 Kafka 低一个数量级。可用性都可以实现高可用。ActiveMQ 和 RabbitMQ 都是基于主从架构实现高可用性。RocketMQ 基于分布式架构。 Kafka 也是分布式的，一个数据多个副本，少数机器宕机，不会丢失数据，不会导致不可用时效性RabbitMQ 基于 Erlang 开发，所以并发能力很强，性能极其好，延时很低，达到微秒级，其他几个都是 ms 级。功能支持Pulsar 的功能更全面，支持多租户、多种消费模式和持久性模式等功能，是下一代云原生分布式消息流平台。消息丢失ActiveMQ 和 RabbitMQ 丢失的可能性非常低， Kafka、RocketMQ 和 Pulsar 理论上可以做到 0 丢失。总结：ActiveMQ 的社区算是比较成熟，但是较目前来说，ActiveMQ 的性能比较差，而且版本迭代很慢，不推荐使用，已经被淘汰了。RabbitMQ 在吞吐量方面虽然稍逊于 Kafka、RocketMQ 和 Pulsar，但是由于它基于 Erlang 开发，所以并发能力很强，性能极其好，延时很低，达到微秒级。但是也因为 RabbitMQ 基于 Erlang 开发，所以国内很少有公司有实力做 Erlang 源码级别的研究和定制。如果业务场景对并发量要求不是太高（十万级、百万级），那这几种消息队列中，RabbitMQ 或许是你的首选。RocketMQ 和 Pulsar 支持强一致性，对消息一致性要求比较高的场景可以使用。RocketMQ 阿里出品，Java 系开源项目，源代码我们可以直接阅读，然后可以定制自己公司的 MQ，并且 RocketMQ 有阿里巴巴的实际业务场景的实战考验。Kafka 的特点其实很明显，就是仅仅提供较少的核心功能，但是提供超高的吞吐量，ms 级的延迟，极高的可用性以及可靠性，而且分布式可以任意扩展。同时 Kafka 最好是支撑较少的 topic 数量即可，保证其超高吞吐量。Kafka 唯一的一点劣势是有可能消息重复消费，那么对数据准确性会造成极其轻微的影响，在大数据领域中以及日志采集中，这点轻微影响可以忽略这个特性天然适合大数据实时计算以及日志收集。如果是大数据领域的实时计算、日志采集等场景，用 Kafka 是业内标准的，绝对没问题，社区活跃度很高，绝对不会黄，何况几乎是全世界这个领域的事实性规范。参考《大型网站技术架构 》KRaft: Apache Kafka Without ZooKeeper：https://developer.confluent.io/learn/kraft/消息队列的使用场景是什么样的？：https://mp.weixin.qq.com/s/4V1jI6RylJr7Jr9JsQe73A
AI 摘要 文章通过短信发送示例，对比演示静态代理、JDK动态代理与CGLIB动态代理的编码步骤、优缺点及适用场景：静态代理编译期硬编码，维护困难；JDK动态代理基于接口生成运行时代理类；CGLIB用继承生成子类，可代理无接口类；总结三者灵活性、性能与框架应用差异。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结07 静态代理+JDK/CGLIB 动态代理实战代理模式代理模式是一种比较好的理解的设计模式。简单来说就是 我们使用代理对象来代替对真实对象(real object)的访问，这样就可以在不修改原目标对象的前提下，提供额外的功能操作，扩展目标对象的功能。代理模式的主要作用是扩展目标对象的功能，比如说在目标对象的某个方法执行前后你可以增加一些自定义的操作。举个例子：你找了小红来帮你问话，小红就可以看作是代理你的代理对象，代理的行为（方法）是问话。https://medium.com/@mithunsasidharan/understanding-the-proxy-design-pattern-5e63fe38052a代理模式有静态代理和动态代理两种实现方式，我们 先来看一下静态代理模式的实现。静态代理静态代理中，我们对目标对象的每个方法的增强都是手动完成的（_后面会具体演示代码_），非常不灵活（_比如接口一旦新增加方法，目标对象和代理对象都要进行修改_）且麻烦(_需要对每个目标类都单独写一个代理类_)。 实际应用场景非常非常少，日常开发几乎看不到使用静态代理的场景。上面我们是从实现和应用角度来说的静态代理，从 JVM 层面来说， 静态代理在编译时就将接口、实现类、代理类这些都变成了一个个实际的 class 文件。静态代理实现步骤:定义一个接口及其实现类；创建一个代理类同样实现这个接口将目标对象注入进代理类，然后在代理类的对应方法调用目标类中的对应方法。这样的话，我们就可以通过代理类屏蔽对目标对象的访问，并且可以在目标方法执行前后做一些自己想做的事情。下面通过代码展示！1.定义发送短信的接口public interface SmsService { String send(String message); }2.实现发送短信的接口public class SmsServiceImpl implements SmsService { public String send(String message) { System.out.println("send message:" + message); return message; } }3.创建代理类并同样实现发送短信的接口public class SmsProxy implements SmsService { private final SmsService smsService; public SmsProxy(SmsService smsService) { this.smsService = smsService; } @Override public String send(String message) { //调用方法之前，我们可以添加自己的操作 System.out.println("before method send()"); smsService.send(message); //调用方法之后，我们同样可以添加自己的操作 System.out.println("after method send()"); return null; } }4.实际使用public class Main { public static void main(String[] args) { SmsService smsService = new SmsServiceImpl(); SmsProxy smsProxy = new SmsProxy(smsService); smsProxy.send("java"); } }运行上述代码之后，控制台打印出：before method send() send message:java after method send()可以输出结果看出，我们已经增强了 SmsServiceImpl 的send()方法。动态代理相比于静态代理来说，动态代理更加灵活。我们不需要针对每个目标类都单独创建一个代理类，并且也不需要我们必须实现接口，我们可以直接代理实现类( _CGLIB 动态代理机制_)。从 JVM 角度来说，动态代理是在运行时动态生成类字节码，并加载到 JVM 中的。说到动态代理，Spring AOP、RPC 框架应该是两个不得不的提的，它们的实现都依赖了动态代理。动态代理在我们日常开发中使用的相对较小，但是在框架中的几乎是必用的一门技术。学会了动态代理之后，对于我们理解和学习各种框架的原理也非常有帮助。就 Java 来说，动态代理的实现方式有很多种，比如 JDK 动态代理、CGLIB 动态代理等等。guide-rpc-framework 使用的是 JDK 动态代理，我们先来看看 JDK 动态代理的使用。另外，虽然 guide-rpc-framework 没有用到 CGLIB 动态代理 ，我们这里还是简单介绍一下其使用以及和JDK 动态代理的对比。JDK 动态代理机制介绍在 Java 动态代理机制中 **InvocationHandler** 接口和 **Proxy** 类是核心。Proxy 类中使用频率最高的方法是：newProxyInstance() ，这个方法主要用来生成一个代理对象。 public static Object newProxyInstance(ClassLoader loader, Class<?>[] interfaces, InvocationHandler h) throws IllegalArgumentException { ...... }这个方法一共有 3 个参数：loader :类加载器，用于加载代理对象。interfaces : 被代理类实现的一些接口；h : 实现了 InvocationHandler 接口的对象；要实现动态代理的话，还必须需要实现InvocationHandler 来自定义处理逻辑。 当我们的动态代理对象调用一个方法时候，这个方法的调用就会被转发到实现InvocationHandler 接口类的 invoke 方法来调用。public interface InvocationHandler { /** * 当你使用代理对象调用方法的时候实际会调用到这个方法 */ public Object invoke(Object proxy, Method method, Object[] args) throws Throwable; }invoke() 方法有下面三个参数：proxy :动态生成的代理类method : 与代理类对象调用的方法相对应args : 当前 method 方法的参数也就是说：你通过**Proxy** 类的 **newProxyInstance()** 创建的代理对象在调用方法的时候，实际会调用到实现**InvocationHandler** 接口的类的 **invoke()**方法。 你可以在 invoke() 方法中自定义处理逻辑，比如在方法执行前后做什么事情。JDK 动态代理类使用步骤定义一个接口及其实现类；自定义 InvocationHandler 并重写invoke方法，在 invoke 方法中我们会调用原生方法（被代理类的方法）并自定义一些处理逻辑；通过 Proxy.newProxyInstance(ClassLoader loader,Class<?>[] interfaces,InvocationHandler h) 方法创建代理对象；代码示例这样说可能会有点空洞和难以理解，我上个例子，大家感受一下吧！1.定义发送短信的接口public interface SmsService { String send(String message); }2.实现发送短信的接口public class SmsServiceImpl implements SmsService { public String send(String message) { System.out.println("send message:" + message); return message; } }3.定义一个 JDK 动态代理类import java.lang.reflect.InvocationHandler; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; /** * @author shuang.kou * @createTime 2020年05月11日 11:23:00 */ public class DebugInvocationHandler implements InvocationHandler { /** * 代理类中的真实对象 */ private final Object target; public DebugInvocationHandler(Object target) { this.target = target; } public Object invoke(Object proxy, Method method, Object[] args) throws InvocationTargetException, IllegalAccessException { //调用方法之前，我们可以添加自己的操作 System.out.println("before method " + method.getName()); Object result = method.invoke(target, args); //调用方法之后，我们同样可以添加自己的操作 System.out.println("after method " + method.getName()); return result; } }invoke() 方法: 当我们的动态代理对象调用原生方法的时候，最终实际上调用到的是 invoke() 方法，然后 invoke() 方法代替我们去调用了被代理对象的原生方法。4.获取代理对象的工厂类public class JdkProxyFactory { public static Object getProxy(Object target) { return Proxy.newProxyInstance( target.getClass().getClassLoader(), // 目标类的类加载 target.getClass().getInterfaces(), // 代理需要实现的接口，可指定多个 new DebugInvocationHandler(target) // 代理对象对应的自定义 InvocationHandler ); } }getProxy() ：主要通过Proxy.newProxyInstance（）方法获取某个类的代理对象5.实际使用SmsService smsService = (SmsService) JdkProxyFactory.getProxy(new SmsServiceImpl()); smsService.send("java");运行上述代码之后，控制台打印出：before method send send message:java after method sendCGLIB 动态代理机制介绍JDK 动态代理有一个最致命的问题是其只能代理实现了接口的类。为了解决这个问题，我们可以用 CGLIB 动态代理机制来避免。CGLIB(_Code Generation Library_)是一个基于ASM的字节码生成库，它允许我们在运行时对字节码进行修改和动态生成。CGLIB 通过继承方式实现代理。很多知名的开源框架都使用到了CGLIB， 例如 Spring 中的 AOP 模块中：如果目标对象实现了接口，则默认采用 JDK 动态代理，否则采用 CGLIB 动态代理。在 CGLIB 动态代理机制中 **MethodInterceptor** 接口和 **Enhancer** 类是核心。你需要自定义 MethodInterceptor 并重写 intercept 方法，intercept 用于拦截增强被代理类的方法。public interface MethodInterceptor extends Callback{ // 拦截被代理类中的方法 public Object intercept(Object obj, java.lang.reflect.Method method, Object[] args, MethodProxy proxy) throws Throwable; }obj :被代理的对象（需要增强的对象）method :被拦截的方法（需要增强的方法）args :方法入参methodProxy :用于调用原始方法你可以通过 Enhancer类来动态获取被代理类，当代理类调用方法的时候，实际调用的是 MethodInterceptor 中的 intercept 方法。CGLIB 动态代理类使用步骤定义一个类；自定义 MethodInterceptor 并重写 intercept 方法，intercept 用于拦截增强被代理类的方法，和 JDK 动态代理中的 invoke 方法类似；通过 Enhancer 类的 create()创建代理类；代码示例不同于 JDK 动态代理不需要额外的依赖。CGLIB(_Code Generation Library_) 实际是属于一个开源项目，如果你要使用它的话，需要手动添加相关依赖。<dependency> <groupId>cglib</groupId> <artifactId>cglib</artifactId> <version>3.3.0</version> </dependency>1.实现一个使用阿里云发送短信的类package github.javaguide.dynamicProxy.cglibDynamicProxy; public class AliSmsService { public String send(String message) { System.out.println("send message:" + message); return message; } }2.自定义 **MethodInterceptor**（方法拦截器）import net.sf.cglib.proxy.MethodInterceptor; import net.sf.cglib.proxy.MethodProxy; import java.lang.reflect.Method; /** * 自定义MethodInterceptor */ public class DebugMethodInterceptor implements MethodInterceptor { /** * @param o 代理对象（增强的对象） * @param method 被拦截的方法（需要增强的方法） * @param args 方法入参 * @param methodProxy 用于调用原始方法 */ @Override public Object intercept(Object o, Method method, Object[] args, MethodProxy methodProxy) throws Throwable { //调用方法之前，我们可以添加自己的操作 System.out.println("before method " + method.getName()); Object object = methodProxy.invokeSuper(o, args); //调用方法之后，我们同样可以添加自己的操作 System.out.println("after method " + method.getName()); return object; } }3.获取代理类import net.sf.cglib.proxy.Enhancer; public class CglibProxyFactory { public static Object getProxy(Class<?> clazz) { // 创建动态代理增强类 Enhancer enhancer = new Enhancer(); // 设置类加载器 enhancer.setClassLoader(clazz.getClassLoader()); // 设置被代理类 enhancer.setSuperclass(clazz); // 设置方法拦截器 enhancer.setCallback(new DebugMethodInterceptor()); // 创建代理类 return enhancer.create(); } }4.实际使用AliSmsService aliSmsService = (AliSmsService) CglibProxyFactory.getProxy(AliSmsService.class); aliSmsService.send("java");运行上述代码之后，控制台打印出：before method send send message:java after method sendJDK 动态代理和 CGLIB 动态代理对比JDK 动态代理只能只能代理实现了接口的类，而 CGLIB 可以代理未实现任何接口的类。 另外， CGLIB 动态代理是通过生成一个被代理类的子类来拦截被代理类的方法调用，因此不能代理声明为 final 类型的类和方法。就二者的效率来说，大部分情况都是 JDK 动态代理更优秀，随着 JDK 版本的升级，这个优势更加明显。静态代理和动态代理的对比灵活性 ：动态代理更加灵活，不需要必须实现接口，可以直接代理实现类，并且可以不需要针对每个目标类都创建一个代理类。另外，静态代理中，接口一旦新增加方法，目标对象和代理对象都要进行修改，这是非常麻烦的！JVM 层面 ：静态代理在编译时就将接口、实现类、代理类这些都变成了一个个实际的 class 文件。而动态代理是在运行时动态生成类字节码，并加载到 JVM 中的。总结这篇文章中主要介绍了代理模式的两种实现：静态代理以及动态代理。涵盖了静态代理和动态代理实战、静态代理和动态代理的区别、JDK 动态代理和 Cglib 动态代理区别等内容。文中涉及到的所有源码，你可以在这里找到：https://github.com/Snailclimb/guide-rpc-framework-learning/tree/master/src/main/java/github/javaguide/proxy 。