28 实战：分布式锁详解与代码
什么是锁？
锁是一种常用的并发控制机制，用于保证一项资源在任何时候只能被一个线程使用，如果其他线程也要使用同样的资源，必须排队等待上一个线程使用完。
锁的示意图，如下所示：
什么是分布式锁？
上面说的锁指的是程序级别的锁，例如 Java 语言中的 synchronized 和 ReentrantLock 在单应用中使用不会有任何问题，但如果放到分布式环境下就不适用了，这个时候我们就要使用分布式锁。
分布式锁比较好理解就是用于分布式环境下并发控制的一种机制，用于控制某个资源在同一时刻只能被一个应用所使用。
分布式锁示意图，如下所示：
怎么实现分布式锁？
分布式锁比较常见的实现方式有三种：
Memcached 实现的分布式锁：使用 add 命令，添加成功的情况下，表示创建分布式锁成功。
Memcached是一个高性能的分布式内存对象缓存系统，它可以用于实现分布式锁。在Memcached中，可以使用add命令来添加一个新的键值对，如果键已经存在，则add命令会失败。
ZooKeeper 实现的分布式锁：使用 ZooKeeper 顺序临时节点来实现分布式锁。
Redis 实现的分布式锁。
本文要重点来说的是第三种，也就是 Redis 分布式锁的实现方式。
Redis 分布式锁的实现思路是使用 setnx（set if not exists），如果创建成功则表明此锁创建成功，否则代表这个锁已经被占用创建失败。
分布式锁实现 1 2 3 4 5 127.0.0.1:6379> setnx lock true (integer) 1 #创建锁成功 # 逻辑业务处理... 127.0.0.1:6379> del lock (integer) 1 #释放锁
从以上代码可以看出，释放锁使用 del lock 即可，如果在锁未被删除之前，其他程序再来执行 setnx 是不会创建成功的，结果如下：
1 2 127.0.0.1:6379> setnx lock true (integer) 0
执行结果为 0 表示失败。
setnx 的问题 setnx 虽然可以成功地创建分布式锁，但存在一个问题，如果此程序在创建了锁之后，程序异常退出了，那么这个锁将永远不会被释放，就造成了死锁的问题 。
这个时候有人想到，我们可以使用 expire key seconds 设置超时时间，即使出现程序中途崩溃的情况，超过超时时间之后，这个锁也会解除，不会出现死锁的情况了，实现命令如下：
1 2 3 4 5 6 7 127.0.0.1:6379> setnx lock true (integer) 1 127.0.0.1:6379> expire lock 30 (integer) 1 # 逻辑业务处理... 127.0.0.1:6379> del lock (integer) 1 #释放锁
但这样依然会有问题，因为命令 setnx 和 expire 处理是一前一后非原子性的，因此如果在它们执行之间，出现断电和 Redis 异常退出的情况，因为超时时间未设置，依然会造成死锁。
带参数的 Set 因为 setnx 和 expire 存在原子性的问题，所以之后出现了很多类库用于解决此问题的，这样就增加了使用的成本，意味着你不但要添加 Redis 本身的客户端，并且为了解决 setnx 分布式锁的问题，还需要额外第三方类库。
然而，这个问题到 Redis 2.6.12 时得到了解决，因为这个版本可以使用 set 并设置超时和非空判定等参数了。
这样我们就可以使用 set 命令来设置分布式锁，并设置超时时间了，而且 set 命令可以保证原子性，实现命令如下所示：
1 2 3 4 127.0.0.1:6379> set lock true ex 30 nx OK #创建锁成功 127.0.0.1:6379> set lock true ex 30 nx (nil) #在锁被占用的时候，企图获取锁失败
其中， ex n 为设置超时时间，nx 为元素非空判断，用来判断是否能正常使用锁的。
分布式锁的执行超时问题
使用 set 命令之后好像所有问题都解决了，然后真相是“没那么简单”。使用 set 命令只解决创建锁的问题，那执行中的极端问题，和释放锁极端问题，我们依旧要考虑。
例如，我们设置锁的最大超时时间是 30s，但业务处理使用了 35s，这就会导致原有的业务还未执行完成，锁就被释放了，新的程序和旧程序一起操作就会带来线程安全的问题。
此执行流程如下图所示：
执行超时的问题处理带来线程安全问题之外，还引发了另一个问题：锁被误删 。
假设锁的最大超时时间是 30s，应用 1 执行了 35s，然而应用 2 在 30s，锁被自动释放之后，用重新获取并设置了锁，然后在 35s 时，应用 1 执行完之后，就会把应用 2 创建的锁给删除掉，如下图所示：
锁被误删的解决方案是在使用 set 命令创建锁时，给 value 值设置一个归属人标识，例如给应用关联一个 UUID，每次在删除之前先要判断 UUID 是不是属于当前的线程，如果属于在删除，这样就避免了锁被误删的问题。
注意：如果是在代码中执行删除，不能使用先判断再删除的方法，伪代码如下：
1 2 3 if (xxx.equals(xxx)){ del(luck); }
因为判断代码和删除代码不具备原子性，因此也不能这样使用，这个时候可以使用 Lua 脚本来执行判断和删除的操作，因为多条 Lua 命令可以保证原子性，Java 实现代码如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 public static boolean unLock (Jedis jedis, String lockKey, String flagId) { String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end" ; Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(flagId)); if ("1L" .equals(result)) { return true ; } return false ; }
其中，Collections.singletonList() 方法的作用是将 String 转成 List，因为 jedis.eval() 最后两个参数的类型必须是 List。
说完了锁误删的解决方案，咱们回过头来看如何解决执行超时的问题，执行超时的问题可以从以下两方面来解决：
把执行比较耗时的任务不要放到加锁的方法内，锁内的方法尽量控制执行时长；
把最大超时时间可以适当的设置长一点，正常情况下锁用完之后会被手动的删除掉，因此适当的把最大超时时间设置的长一点，也是可行的。
代码实战
下面我们使用 Java 代码来实现分布式锁，代码如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 import org.apache.commons.lang3.StringUtils;import redis.clients.jedis.Jedis;import redis.clients.jedis.params.SetParams;import utils.JedisUtils;import java.util.Collections;public class LockExample { static final String _LOCKKEY = "REDISLOCK" ; static final String _FLAGID = "UUID:6379" ; static final Integer _TimeOut = 90 ; public static void main (String[] args) { Jedis jedis = JedisUtils.getJedis(); boolean lockResult = lock(jedis, _LOCKKEY, _FLAGID, _TimeOut); if (lockResult) { System.out.println("加锁成功" ); } else { System.out.println("加锁失败" ); } if (unLock(jedis, _LOCKKEY, _FLAGID)) { System.out.println("锁释放成功" ); } else { System.out.println("锁释放成功" ); } } public static boolean lock (Jedis jedis, String key, String flagId, Integer secondsTime) { SetParams params = new SetParams (); params.ex(secondsTime); params.nx(); String res = jedis.set(key, flagId, params); if (StringUtils.isNotBlank(res) && res.equals("OK" )) return true ; return false ; } public static boolean unLock (Jedis jedis, String lockKey, String flagId) { String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end" ; Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(flagId)); if ("1L" .equals(result)) { return true ; } return false ; } }
以上代码执行结果如下所示：
小结
本文介绍了锁和分布式锁的概念，锁其实就是用来保证同一时刻只有一个程序可以去操作某一个资源，以此来保证并发时程序能正常执行的。使用 Redis 来实现分布式锁不能使用 setnx 命令，因为它可能会带来死锁的问题，因此我们可以使用 Redis 2.6.12 带来的多参数的 set 命令来申请锁，但在使用的时候也要注意锁内的业务流程执行的时间，不能大于锁设置的最大超时时间，不然会带来线程安全问题和锁误删的问题。
29 实战：布隆过滤器安装与使用及原理分析
我们前面有讲到过 HyperLogLog 可以用来做基数统计，但它没提供判断一个值是否存在的查询方法，那我们如何才能查询一个值是否存在于海量数据之中呢？
如果使用传统的方式，例如 SQL 中的传统查询，因为数据量太多，查询效率又低有占用系统的资源，因此我们需要一个优秀的算法和功能来实现这个需求，这是我们今天要讲的——布隆过滤器。
原理
介绍 ：布隆过滤器（Bloom Filter） 是一种 空间效率很高 的 概率型数据结构 ，用于判断一个元素是否存在于一个集合中。布隆过滤器能够以较小的内存占用来快速判断某个元素可能存在或一定不存在。不过，它有个特点：可能会误判元素存在 ，但不会误判元素不存在。
原理 ：布隆过滤器使用 位数组 和 多个哈希函数 来表示集合中的元素。它通过哈希函数将元素映射到位数组中的若干个位置，并将这些位置标记为 1。在查询元素时，布隆过滤器使用相同的哈希函数检查这些位置：
如果所有相应位置都是 1，则说明该元素可能存在 （但可能存在误判）。
如果有任意一个位置为 0，则说明该元素一定不存在 。
工作流程
插入元素：
初始化一个位数组 bit array，长度为 m，所有位初始化为 0。
对于每个要插入的元素，使用 k 个哈希函数 对元素进行哈希运算，将其映射到位数组的 k 个位置 。
将哈希函数映射到的这些位置对应的位设置为 1。
查询元素：
对查询的元素，使用同样的 k 个哈希函数 进行哈希运算，检查元素对应的 k 个位置 。
如果所有这些位置均为 1，则该元素可能存在 （存在一定的误判概率）。
如果有任意一个位置为 0，则该元素一定不存在 。
误判原因：
布隆过滤器的误判主要是因为哈希碰撞 ：多个不同的元素可能会被哈希函数映射到相同的位，导致某些位被设置为 1。因此，当查询一个不存在的元素时，它经过哈希函数映射可能会碰巧落在所有为 1 的位置上，从而误判该元素存在。
不过，布隆过滤器的误判率可以通过以下方式控制：
位数组的长度 （m）：位数组越长，哈希碰撞的概率越低。哈希函数的数量 （k）：哈希函数越多，元素映射的位数越多，误判率越低，但也会增加计算复杂度。插入的元素个数 （n）：元素越多，哈希碰撞的概率越高，误判率会随之增加。
解决误判
增加位数组长度 ：更长的位数组可以减少哈希冲突。增加哈希函数数量 ：合理数量的哈希函数可以减少误判，但不是越多越好。分层布隆过滤器 ：通过多层过滤进一步降低误判概率。可计数布隆过滤器 ：允许删除元素并减少误判。选择更好的哈希函数 ：减少哈希冲突，提高准确性。分布式布隆过滤器 ：适合大规模分布式系统，减少单个过滤器的负担。结合其他数据结构验证 ：与哈希表或数据库结合，进一步验证元素是否存在。使用 Golomb-Coded Sets ：在空间效率和误判率方面具有优势。
开启布隆过滤器
在 Redis 中不能直接使用布隆过滤器，但我们可以通过 Redis 4.0 版本之后提供的 modules（扩展模块）的方式引入，本文提供两种方式的开启方式。
方式一：编译方式 1. 下载并安装布隆过滤器
1 2 3 git clone https://github.com/RedisLabsModules/redisbloom.git cd redisbloommake
编译正常执行完，会在根目录生成一个 redisbloom.so 文件。
2. 启动 Redis 服务器
1 > ./src/redis-server redis.conf --loadmodule ./src/modules/RedisBloom-master/redisbloom.so
其中 --loadmodule 为加载扩展模块的意思，后面跟的是 redisbloom.so 文件的目录。
方式二：Docker 方式 1 2 docker pull redislabs/rebloom docker run -p6379:6379 redislabs/rebloom
启动验证 服务启动之后，我们需要判断布隆过滤器是否正常开启，此时我们只需使用 redis-cli 连接到服务端，输入 bf.add 看有没有命令提示，就可以判断是否正常启动了，如下图所示：
如果有命令提示则表名 Redis 服务器已经开启了布隆过滤器。
布隆过滤器的使用
布隆过滤器的命令不是很多，主要包含以下几个：
bf.add：添加元素
bf.exists：判断某个元素是否存在
bf.madd：添加多个元素
bf.mexists：判断多个元素是否存在
bf.reserve：设置布隆过滤器的准确率
具体使用如下所示：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 127.0.0.1:6379> bf.add user xiaoming (integer) 1 127.0.0.1:6379> bf.add user xiaohong (integer) 1 127.0.0.1:6379> bf.add user laowang (integer) 1 127.0.0.1:6379> bf.exists user laowang (integer) 1 127.0.0.1:6379> bf.exists user lao (integer) 0 127.0.0.1:6379> bf.madd user huahua feifei 1) (integer) 1 2) (integer) 1 127.0.0.1:6379> bf.mexists user feifei laomiao 1) (integer) 1 2) (integer) 0
可以看出以上结果没有任何误差，我们再来看一下准确率 bf.reserve 的使用：
1 2 3 4 127.0.0.1:6379> bf.reserve user 0.01 200 (error) ERR item exists #已经存的 key 设置会报错 127.0.0.1:6379> bf.reserve userlist 0.9 10 OK
可以看出此命令必须在元素刚开始执行，否则会报错，它有三个参数：key、error_rate 和 initial_size。
其中：
error_rate：允许布隆过滤器的错误率，这个值越低过滤器占用空间也就越大，以为此值决定了位数组的大小，位数组是用来存储结果的，它的空间占用的越大（存储的信息越多），错误率就越低，它的默认值是 0.01。
initial_size：布隆过滤器存储的元素大小，实际存储的值大于此值，准确率就会降低，它的默认值是 100。
后面原理部分会讲到 error_rate 和 initial_size 对准确率影响的具体原因。
代码实战
下面我们用 Java 客户端来实现布隆过滤器的操作，因为 Jedis 没有直接操作布隆过滤器的方法，所以我们使用 Jedis 操作 Lua 脚本的方式来实现布隆过滤器，代码如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 import redis.clients.jedis.Jedis;import utils.JedisUtils;import java.util.Arrays;public class BloomExample { private static final String _KEY = "user" ; public static void main (String[] args) { Jedis jedis = JedisUtils.getJedis(); for (int i = 1 ; i < 10001 ; i++) { bfAdd(jedis, _KEY, "user_" + i); boolean exists = bfExists(jedis, _KEY, "user_" + i); if (!exists) { System.out.println("未找到数据 i=" + i); break ; } } System.out.println("执行完成" ); } public static boolean bfAdd (Jedis jedis, String key, String value) { String luaStr = "return redis.call('bf.add', KEYS[1], KEYS[2])" ; Object result = jedis.eval(luaStr, Arrays.asList(key, value), Arrays.asList()); if (result.equals(1L )) { return true ; } return false ; } public static boolean bfExists (Jedis jedis, String key, String value) { String luaStr = "return redis.call('bf.exists', KEYS[1], KEYS[2])" ; Object result = jedis.eval(luaStr, Arrays.asList(key, value), Arrays.asList()); if (result.equals(1L )) { return true ; } return false ; } }
但我们发现执行了半天，执行的结果竟然是：
没有任何误差，奇怪了，于是我们在循环次数后面又加了一个 0，执行了大半天之后，发现依旧是相同的结果，依旧没有任何误差。
这是因为对于布隆过滤器来说，它说没有的值一定没有，它说有的值有可能没有 。
于是我们把程序改一下，重新换一个 key 值，把条件改为查询存在的数据，代码如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 import redis.clients.jedis.Jedis;import utils.JedisUtils;import java.util.Arrays;public class BloomExample { private static final String _KEY = "userlist" ; public static void main (String[] args) { Jedis jedis = JedisUtils.getJedis(); for (int i = 1 ; i < 100001 ; i++) { bfAdd(jedis, _KEY, "user_" + i); boolean exists = bfExists(jedis, _KEY, "user_" + (i + 1 )); if (exists) { System.out.println("找到了" + i); break ; } } System.out.println("执行完成" ); } public static boolean bfAdd (Jedis jedis, String key, String value) { String luaStr = "return redis.call('bf.add', KEYS[1], KEYS[2])" ; Object result = jedis.eval(luaStr, Arrays.asList(key, value), Arrays.asList()); if (result.equals(1L )) { return true ; } return false ; } public static boolean bfExists (Jedis jedis, String key, String value) { String luaStr = "return redis.call('bf.exists', KEYS[1], KEYS[2])" ; Object result = jedis.eval(luaStr, Arrays.asList(key, value), Arrays.asList()); if (result.equals(1L )) { return true ; } return false ; } }
这次我们发现执行不一会就出现了如下信息：
说明循环执行了一会之后就出现误差了，代码执行也符合布隆过滤器的预期。
原理
上面我们学会了布隆过滤器的使用，下面我们就来看下它的实现原理。
Redis 布隆过滤器的实现，依靠的是它数据结构中的一个位数组，每次存储键值的时候，不是直接把数据存储在数据结构中，因为这样太占空间了，它是利用几个不同的无偏哈希函数，把此元素的 hash 值均匀的存储在位数组中，也就是说，每次添加时会通过几个无偏哈希函数算出它的位置，把这些位置设置成 1 就完成了添加操作。
当进行元素判断时，查询此元素的几个哈希位置上的值是否为 1，如果全部为 1，则表示此值存在，如果有一个值为 0，则表示不存在。因为此位置是通过 hash 计算得来的，所以即使这个位置是 1，并不能确定是那个元素把它标识为 1 的，因此布隆过滤器查询此值存在时，此值不一定存在，但查询此值不存在时，此值一定不存在 。
并且当位数组存储值比较稀疏的时候，查询的准确率越高，而当位数组存储的值越来越多时，误差也会增大。
位数组和 key 之间的关系，如下图所示：
布隆过滤器使用场景
它的经典使用场景包括以下几个：
垃圾邮件过滤
爬虫里的 URL 去重
判断一个元素在亿级数据中是否存在
布隆过滤器在数据库领域的使用也比较广泛，例如：HBase、Cassandra、LevelDB、RocksDB 内部都有使用布隆过滤器。
小结
通过本文我们知道可以使用 Redis 4.0 之后提供的 modules 方式来开启布隆过滤器，并学习了布隆过滤器的三个重要操作方法 bf.add 添加元素、bf.exists 查询元素是否存在，还有 bf.reserve 设置布隆过滤器的准确率，其中 bf.reserve 有 2 个重要的参数：错误率和数组大小，错误率设置的越低，数组设置的越大，需要存储的空间就越大，相对来说查询的错误率也越低，需要如何设置需要使用者根据实际情况进行调整。我们也知道布隆过滤器的特点：当它查询有数据时，此数据不一定真的存在，当它查询没有此数据时，此数据一定不存在。
通俗说 ：
布隆过滤器是一种空间效率极高的概率性数据结构，用于判断一个元素是否在集合中。Redis 的布隆过滤器实现就是依靠位数组和无偏哈希函数来完成的。
比如我们有一个空的布隆过滤器，位数组的长度为10，初始状态下所有位都是0：
现在我们要添加一个元素“Redis”。我们选择两个哈希函数，分别计算出“Redis”的哈希值为2和7。于是我们把位数组的第2位和第7位设置为1：
这样就完成了“Redis”的添加操作。
然后，当我们要判断一个元素是否在集合中时，只需要用同样的哈希函数计算出元素的哈希值，然后查看位数组对应的位置是否都为1。如果都为1，那么元素可能在集合中；如果有任何一位不为1，那么元素肯定不在集合中。
需要注意的是，布隆过滤器可能会有假阳性，也就是说，有些并不存在集合中的元素可能会被判断为存在。这是因为不同的元素可能会被哈希到同样的位置。但布隆过滤器不会有假阴性，如果它说某元素不在集合中，那么这个元素肯定不在集合中。
30 完整案例：实现延迟队列的两种方法
延迟队列是指把当前要做的事情，往后推迟一段时间再做。
延迟队列在实际工作中和面试中都比较常见，它的实现方式有很多种，然而每种实现方式也都有它的优缺点，接下来我们来看。
延迟队列的使用场景
延迟队列的常见使用场景有以下几种：
超过 30 分钟未支付的订单，将会被取消
外卖商家超过 5 分钟未接单的订单，将会被取消
在平台注册但 30 天内未登录的用户，发短信提醒
等类似的应用场景，都可以使用延迟队列来实现。
常见实现方式
Redis 延迟队列实现的思路、优点：目前市面上延迟队列的实现方式基本分为三类，第一类是通过程序的方式实现，例如 JDK 自带的延迟队列 DelayQueue，第二类是通过 MQ 框架来实现，例如 RabbitMQ 可以通过 rabbitmq-delayed-message-exchange 插件来实现延迟队列，第三类就是通过 Redis 的方式来实现延迟队列。
程序实现方式 JDK 自带的 DelayQueue 实现延迟队列，代码如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 public class DelayTest { public static void main (String[] args) throws InterruptedException { DelayQueue delayQueue = new DelayQueue (); delayQueue.put(new DelayElement (1000 )); delayQueue.put(new DelayElement (3000 )); delayQueue.put(new DelayElement (5000 )); System.out.println("开始时间：" + DateFormat.getDateTimeInstance().format(new Date ())); while (!delayQueue.isEmpty()){ System.out.println(delayQueue.take()); } System.out.println("结束时间：" + DateFormat.getDateTimeInstance().format(new Date ())); } static class DelayElement implements Delayed { long delayTime = System.currentTimeMillis(); public DelayElement (long delayTime) { this .delayTime = (this .delayTime + delayTime); } @Override public long getDelay (TimeUnit unit) { return unit.convert(delayTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS); } @Override public int compareTo (Delayed o) { if (this .getDelay(TimeUnit.MILLISECONDS) > o.getDelay(TimeUnit.MILLISECONDS)) { return 1 ; } else if (this .getDelay(TimeUnit.MILLISECONDS) < o.getDelay(TimeUnit.MILLISECONDS)) { return -1 ; } else { return 0 ; } } @Override public String toString () { return DateFormat.getDateTimeInstance().format(new Date (delayTime)); } } }
程序执行结果：
1 2 3 4 5 开始时间：2019-6-13 20 :40:38 2019-6 -13 20 :40:39 2019-6 -13 20 :40:41 2019-6 -13 20 :40:43 结束时间：2019-6-13 20 :40:43
优点
开发比较方便，可以直接在代码中使用
代码实现比较简单
缺点
不支持持久化保存
不支持分布式系统
MQ 实现方式 RabbitMQ 本身并不支持延迟队列，但可以通过添加插件 rabbitmq-delayed-message-exchange 来实现延迟队列。
优点
支持分布式
支持持久化
缺点
框架比较重，需要搭建和配置 MQ。
Redis 实现方式 Redis 是通过有序集合（ZSet）的方式来实现延迟消息队列的，ZSet 有一个 Score 属性可以用来存储延迟执行的时间。
优点
灵活方便，Redis 是互联网公司的标配，无序额外搭建相关环境；
可进行消息持久化，大大提高了延迟队列的可靠性；
分布式支持，不像 JDK 自身的 DelayQueue；
高可用性，利用 Redis 本身高可用方案，增加了系统健壮性。
缺点
需要使用无限循环的方式来执行任务检查，会消耗少量的系统资源。
结合以上优缺点，我们决定使用 Redis 来实现延迟队列，具体实现代码如下。
代码实战
本文我们使用 Java 语言来实现延迟队列，延迟队列的实现有两种方式：第一种是利用 zrangebyscore 查询符合条件的所有待处理任务，循环执行队列任务。第二种实现方式是每次查询最早的一条消息，判断这条信息的执行时间是否小于等于此刻的时间，如果是则执行此任务，否则继续循环检测。
方式一 一次性查询所有满足条件的任务，循环执行，代码如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 import redis.clients.jedis.Jedis;import utils.JedisUtils;import java.time.Instant;import java.util.Set;public class DelayQueueExample { private static final String _KEY = "myDelayQueue" ; public static void main (String[] args) throws InterruptedException { Jedis jedis = JedisUtils.getJedis(); long delayTime = Instant.now().plusSeconds(30 ).getEpochSecond(); jedis.zadd(_KEY, delayTime, "order_1" ); jedis.zadd(_KEY, Instant.now().plusSeconds(2 ).getEpochSecond(), "order_2" ); jedis.zadd(_KEY, Instant.now().plusSeconds(2 ).getEpochSecond(), "order_3" ); jedis.zadd(_KEY, Instant.now().plusSeconds(7 ).getEpochSecond(), "order_4" ); jedis.zadd(_KEY, Instant.now().plusSeconds(10 ).getEpochSecond(), "order_5" ); doDelayQueue(jedis); } public static void doDelayQueue (Jedis jedis) throws InterruptedException { while (true ) { Instant nowInstant = Instant.now(); long lastSecond = nowInstant.plusSeconds(-1 ).getEpochSecond(); long nowSecond = nowInstant.getEpochSecond(); Set<String> data = jedis.zrangeByScore(_KEY, lastSecond, nowSecond); for (String item : data) { System.out.println("消费：" + item); } jedis.zremrangeByScore(_KEY, lastSecond, nowSecond); Thread.sleep(1000 ); } } }
以上程序执行结果如下：
1 2 3 4 5 消费：order_2 消费：order_3 消费：order_4 消费：order_5 消费：order_1
方式二 每次查询最早的一条任务，与当前时间判断，决定是否需要执行，实现代码如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 import redis.clients.jedis.Jedis;import utils.JedisUtils;import java.time.Instant;import java.util.Set;public class DelayQueueExample { private static final String _KEY = "myDelayQueue" ; public static void main (String[] args) throws InterruptedException { Jedis jedis = JedisUtils.getJedis(); long delayTime = Instant.now().plusSeconds(30 ).getEpochSecond(); jedis.zadd(_KEY, delayTime, "order_1" ); jedis.zadd(_KEY, Instant.now().plusSeconds(2 ).getEpochSecond(), "order_2" ); jedis.zadd(_KEY, Instant.now().plusSeconds(2 ).getEpochSecond(), "order_3" ); jedis.zadd(_KEY, Instant.now().plusSeconds(7 ).getEpochSecond(), "order_4" ); jedis.zadd(_KEY, Instant.now().plusSeconds(10 ).getEpochSecond(), "order_5" ); doDelayQueue2(jedis); } public static void doDelayQueue2 (Jedis jedis) throws InterruptedException { while (true ) { long nowSecond = Instant.now().getEpochSecond(); Set<String> data = jedis.zrange(_KEY, 0 , 0 ); if (data.size() == 1 ) { String firstValue = data.iterator().next(); Double score = jedis.zscore(_KEY, firstValue); if (nowSecond >= score) { System.out.println("消费消息：" + firstValue); jedis.zrem(_KEY, firstValue); } } Thread.sleep(100 ); } } }
以上程序执行结果和实现方式一相同，结果如下：
1 2 3 4 5 消费：order_2 消费：order_3 消费：order_4 消费：order_5 消费：order_1
其中，执行间隔代码 Thread.sleep(100) 可根据实际的业务情况删减或配置。
小结
本文我们介绍了延迟队列的使用场景以及各种实现方案，其中 Redis 的方式是最符合我们需求的，它主要是利用有序集合的 score 属性来存储延迟执行时间，再开启一个无限循环来判断是否有符合要求的任务，如果有的话执行相关逻辑，没有的话继续循环检测。
31 实战：定时任务案例
我在开发的时候曾经遇到了这样一个问题，产品要求给每个在线预约看病的患者，距离预约时间的前一天发送一条提醒推送，以防止患者错过看病的时间。这个时候就要求我们给每个人设置一个定时任务，用前面文章说的延迟队列也可以实现，但延迟队列的实现方式需要开启一个无限循环任务，那有没有其他的实现方式呢？
答案是肯定的，接下来我们就用 Keyspace Notifications（键空间通知）来实现定时任务，定时任务指的是指定一个时间来执行某个任务，就叫做定时任务 。
开启键空间通知
默认情况下 Redis 服务器端是不开启键空间通知的，需要我们手动开启。
键空间开启分为两种方式：
接下来，我们分别来看。
命令设置方式 使用 redis-cli 连接到服务器端之后，输入 config set notify-keyspace-events Ex 命令，可以直接开启键空间通知功能，返回“OK”则表示开启成功，如下命令所示：
1 2 127.0.0.1:6379> config set notify-keyspace-events Ex OK
优点：
缺点：
这种方式设置的配置信息是存储在内存中的，重启 Redis 服务之后，配置项会丢失。
配置文件设置方式 找到 Redis 的配置文件 redis.conf，设置配置项 notify-keyspace-events Ex，然后重启 Redis 服务器。
优点：
无论 Redis 服务器重启多少次，配置都不会丢失。
缺点：
配置说明 可以看出无论是那种方式，都是设置 notify-keyspace-events Ex，其中 Ex 表示开启键事件通知里面的 key 过期事件。
更多配置项说明如下：
K：键空间通知，所有通知以 __keyspace@<db>__ 为前缀
E：键事件通知，所有通知以 __keyevent@<db>__ 为前缀
g：DEL、EXPIRE、RENAME 等类型无关的通用命令的通知
$：字符串命令的通知
l：列表命令的通知
s：集合命令的通知
h：哈希命令的通知
z：有序集合命令的通知
x：过期事件，每当有过期键被删除时发送
e：驱逐（evict）事件，每当有键因为 maxmemory 政策而被删除时发送
A：参数 g$lshzxe 的别名
以上配置项可以自由组合，例如我们订阅列表事件就是 El，但需要注意的是，如果 notify-keyspace-event 的值设置为空，则表示不开启任何通知，有值则表示开启通知 。
功能实现
我们要实现定时任务需要使用 Pub/Sub 订阅者和发布者的功能，使用订阅者订阅元素的过期事件，然后再执行固定的任务，这就是定时任务的实现思路。
以本文开头的问题为例，我们是这样实现此定时任务的，首先根据每个患者预约的时间往前推一天，然后再计算出当前时间和目标时间（预约前一天的时间）的毫秒值，把这个值作为元素的过期时间设置到 Redis 中，当这个键过期的时候，我们使用订阅者模式就可以订阅到此信息，然后再发提醒消息给此用户，这样就实现了给每个患者开启一个单独的分布式定时任务的功能。
我们先用命令的模式来模拟一下此功能的实现，首先，我们使用 redis-cli 开启一个客户端，监听 __keyevent@0__:expired 键过期事件，此监听值 __keyevent@0__:expired 为固定的写法，其中 0 表示第一个数据库，我们知道 Redis 中一共有 16 个数据，默认使用的是第 0 个，我们建议新开一个非 0 的数据库专门用来实现定时任务，这样就可以避免很多无效的事件监听。
命令监听如下：
1 2 3 4 127.0.0.1:6379> psubscribe __keyevent@0__:expired 1) "psubscribe" 2) "__keyevent@0__:expired" 3) (integer) 1
此时我们开启另一个客户端，添加两条测试数据试试，命令如下：
1 2 3 4 127.0.0.1:6379> set key value ex 3 OK 127.0.0.1:6379> set user xiaoming ex 3 OK
等过去 3 秒钟之后，我们去看监听结果如下：
1 2 3 4 5 6 7 8 9 10 11 12 127.0.0.1:6379> psubscribe __keyevent@0__:expired 1) "psubscribe" 2) "__keyevent@0__:expired" 3) (integer) 1 1) "pmessage" 2) "__keyevent@0__:expired" 3) "__keyevent@0__:expired" 4) "key" #接收到过期信息 key 1) "pmessage" 2) "__keyevent@0__:expired" 3) "__keyevent@0__:expired" 4) "user" #接收到过期信息 user
已经成功的介绍到两条过期信息了。
代码实战
本文我们使用 Jedis 来实现定时任务，代码如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 import redis.clients.jedis.Jedis;import redis.clients.jedis.JedisPubSub;import utils.JedisUtils;public class TaskExample { public static final String _TOPIC = "__keyevent@0__:expired" ; public static void main (String[] args) { Jedis jedis = JedisUtils.getJedis(); doTask(jedis); } public static void doTask (Jedis jedis) { jedis.psubscribe(new JedisPubSub () { @Override public void onPMessage (String pattern, String channel, String message) { System.out.println("收到消息：" + message); } }, _TOPIC); } }
小结
本文我们通过开启 Keyspace Notifications 和 Pub/Sub 消息订阅的方式，可以拿到每个键值过期的事件，我们利用这个机制实现了给每个人开启一个定时任务的功能，过期事件中我们可以获取到过期键的 key 值，在 key 值中我们可以存储每个用户的 id，例如“user_1001”的方式，其中数字部分表示用户的编号，通过此编号就可以完成给对应人发送消息通知的功能。
32 实战：RediSearch 高性能的全文搜索引擎
RediSearch 是一个高性能的全文搜索引擎，它可以作为一个 Redis Module（扩展模块）运行在 Redis 服务器上。
RediSearch 主要特性如下：
基于文档的多个字段全文索引
高性能增量索引
文档排序（由用户在索引时手动提供）
在子查询之间使用 AND 或 NOT 操作符的复杂布尔查询
可选的查询子句
基于前缀的搜索
支持字段权重设置
自动完成建议（带有模糊前缀建议）
精确的短语搜索
在许多语言中基于词干分析的查询扩展
支持用于查询扩展和评分的自定义函数
将搜索限制到特定的文档字段
数字过滤器和范围
使用 Redis 自己的地理命令进行地理过滤
Unicode 支持（需要 UTF-8 字符集）
检索完整的文档内容或只是 ID 的检索
支持文档删除和更新与索引垃圾收集
支持部分更新和条件文档更新
安装
和前面讲到布隆过滤器的引入方式一样，我们可以使用 RediSearch 官方推荐的 Docker 方式来安装并启动 RediSearch 功能，操作命令如下：
1 docker run -p 6379:6379 redislabs/redisearch:latest
安装完成之后使用 redis-cli 来检查 RediSearch 模块是否加载成功 ，使用 Docker 启动 redis-cli，命令如下：
1 docker exec -it myredis redis-cli
其中“myredis”为 Redis 服务器的名称，执行结果如下：
1 2 3 4 5 127.0.0.1:6379> module list 1) 1) "name" 2) "ft" 3) "ver" 4) (integer) 10610
返回数组存在“ft”，表明 RediSearch 模块已经成功加载。
源码方式安装 如果不想使用 Docker，我们也可以使用源码的方式进行安装，安装命令如下：
1 2 3 git clone https://github.com/RedisLabsModules/RediSearch.git cd RediSearch # 进入模块目录 make all
安装完成之后，可以使用如下命令启动 Redis 并加载 RediSearch 模块，命令如下：
1 src/redis-server redis.conf --loadmodule ../RediSearch/src/redisearch.so
使用
我们先使用 redis-cli 来对 RediSearch 进行相关的操作。
创建索引和字段 1 2 127.0.0.1:6379> ft.create myidx schema title text weight 5.0 desc text OK
其中“myidx”为索引的ID，此索引包含了两个字段“title”和“desc”，“weight”为权重，默认值为 1.0。
将内容添加到索引 1 2 127.0.0.1:6379> ft.add myidx doc1 1.0 fields title "He urged her to study English" desc "good idea" OK
其中“doc1”为文档 ID（docid），“1.0”为评分（score）。
根据关键查询 1 2 3 4 5 6 7 127.0.0.1:6379> ft.search myidx "english" limit 0 10 1) (integer) 1 2) "doc1" 3) 1) "title" 2) "He urged her to study English" 3) "desc" 4) "good idea"
可以看出我们使用 title 字段中的关键字“english”查询出了一条满足查询条件的数据。
中文搜索 首先我们需要先给索引中，添加一条中文数据，执行命令如下：
1 2 127.0.0.1:6379> ft.add myidx doc2 1.0 language "chinese" fields title "Java 14 发布了！新功能速览" desc "Java 14 在 2020.3.17 日发布正式版了，但现在很多公司还在使用 Java 7 或 Java 8" OK
注意：这里必须要设置语言编码为中文，也就是“language “chinese”” ，默认是英文编码，如果不设置则无法支持中文查询（无法查出结果）。
我们使用之前的查询方式，命令如下：
1 2 127.0.0.1:6379> ft.search myidx "正式版" 1) (integer) 0
我们发现并没有查到任何信息，这是因为我们没有指定搜索的语言，不但保存时候要指定编码，查询时也需要指定，查询命令如下：
1 2 3 4 5 6 7 127.0.0.1:6379> ft.search myidx "发布了" language "chinese" 1) (integer) 1 2) "doc2" 3) 1) "desc" 2) "Java 14 \xe5\x9c\xa8 2020.3.17 \xe6\x97\xa5\xe5\x8f\x91\xe5\xb8\x83\xe6\xad\xa3\xe5\xbc\x8f\xe7\x89\x88\xe4\xba\x86\xef\xbc\x8c\xe4\xbd\x86\xe7\x8e\xb0\xe5\x9c\xa8\xe5\xbe\x88\xe5\xa4\x9a\xe5\x85\xac\xe5\x8f\xb8\xe8\xbf\x98\xe5\x9c\xa8\xe4\xbd\xbf\xe7\x94\xa8 Java 7 \xe6\x88\x96 Java 8" 3) "title" 4) "Java 14 \xe5\x8f\x91\xe5\xb8\x83\xe4\xba\x86\xef\xbc\x81\xe6\x96\xb0\xe5\x8a\x9f\xe8\x83\xbd\xe9\x80\x9f\xe8\xa7\x88"
从结果可以看出中文信息已经被顺利的查询出来了。
删除索引的数据 1 2 127.0.0.1:6379> ft.del myidx doc1 (integer) 1
我们使用索引加文档 ID 就可以实现删除数据的功能。
删除索引 我们可以使用“ft.drop”关键字删除整个索引，执行命令如下：
1 2 127.0.0.1:6379> ft.drop myidx OK
查询索引详细信息 我们可以使用“ft.info ”关键查询索引相关信息，执行命令如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 127.0.0.1:6379> ft.info myidx 1) index_name 2) myidx 3) index_options 4) (empty list or set) 5) fields 6) 1) 1) title 2) type 3) TEXT 4) WEIGHT 5) "5" 2) 1) desc 2) type 3) TEXT 4) WEIGHT 5) "1" 7) num_docs 8) "2" 9) max_doc_id 10) "2" 11) num_terms 12) "9" 13) num_records 14) "18" 15) inverted_sz_mb 16) "0.000102996826171875" 17) total_inverted_index_blocks 18) "29" 19) offset_vectors_sz_mb 20) "1.71661376953125e-05" 21) doc_table_size_mb 22) "0.000164031982421875" 23) sortable_values_size_mb 24) "0" 25) key_table_size_mb 26) "8.0108642578125e-05" 27) records_per_doc_avg 28) "9" 29) bytes_per_record_avg 30) "6" 31) offsets_per_term_avg 32) "1" 33) offset_bits_per_record_avg 34) "8" 35) gc_stats 36) 1) bytes_collected 2) "0" 3) total_ms_run 4) "16" 5) total_cycles 6) "14" 7) avarage_cycle_time_ms 8) "1.1428571428571428" 9) last_run_time_ms 10) "2" 11) gc_numeric_trees_missed 12) "0" 13) gc_blocks_denied 14) "0" 37) cursor_stats 38) 1) global_idle 2) (integer) 0 3) global_total 4) (integer) 0 5) index_capacity 6) (integer) 128 7) index_total 8) (integer) 0
其中“num_docs”表示存储的数据数量。
代码实战
本文我们使用 JRediSearch 来实现全文搜索的功能，首先在 pom.xml 添加 JRediSearch 引用：
1 2 3 4 5 6 <dependency > <groupId > com.redislabs</groupId > <artifactId > jredisearch</artifactId > <version > 1.3.0</version > </dependency >
完整的操作代码如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 import io.redisearch.client.AddOptions; import io.redisearch.client.Client; import io.redisearch.Document; import io.redisearch.SearchResult; import io.redisearch.Query; import io.redisearch.Schema; public class RediSearchExample { public static void main(String[] args) { // 连接 Redis 服务器和指定索引 Client client = new Client("myidx", "127.0.0.1", 6379); // 定义索引 Schema schema = new Schema().addTextField("title", 5.0).addTextField("desc", 1.0); // 删除索引 client.dropIndex(); // 创建索引 client.createIndex(schema, Client.IndexOptions.Default()); // 设置中文编码 AddOptions addOptions = new AddOptions(); addOptions.setLanguage("chinese"); // 添加数据 Document document = new Document("doc1"); document.set("title", "天气预报"); document.set("desc", "今天的天气很好，是个阳光明媚的大晴天，有蓝蓝的天空和白白的云朵。"); // 向索引中添加文档 client.addDocument(document,addOptions); // 查询 Query q = new Query("天气") // 设置查询条件 .setLanguage("chinese") // 设置为中文编码 .limit(0,5); // 返回查询结果 SearchResult res = client.search(q); // 输出查询结果 System.out.println(res.docs); } }
以上程序执行结果如下：
1 [{"id" :"doc1" ,"score" :1.0,"properties" :{"title" :"天气预报" ,"desc" :"今天的天气很好，是个阳光明媚的大晴天，有蓝蓝的天空和白白的云朵。" }}]
可以看出添加的中文数据，被正确的查询出来了。
小结
本文我们使用 Docker 和 源码编译的方式成功的启动了 RediSearch 功能，要使用 RediSearch 的全文搜索功能，必须先要创建一个索引，然后再索引中添加数据，再使用 ft.search 命令进行全文搜索，如果要查询中文内容的话，需要在添加数据时设置中文编码，并且在查询时也要设置中文编码，指定“language “chinese””。
33 实战：Redis 性能测试
为什么需要性能测试？
性能测试的使用场景有很多，例如以下几个：
技术选型，比如测试 Memcached 和 Redis；
对比单机 Redis 和集群 Redis 的吞吐量；
评估不同类型的存储性能，例如集合和有序集合；
对比开启持久化和关闭持久化的吞吐量；
对比调优和未调优的吞吐量；
对比不同 Redis 版本的吞吐量，作为是否升级的一个参考标准。
等等，诸如此类的情况，我们都需要进行性能测试。
性能测试的几种方式
既然性能测试使用场景那么多，那要怎么进行性能测试呢？
目前比较主流的性能测试分为两种：
编写代码模拟并发进行性能测试；
使用 redis-benchmark 进行测试。
因为自己编写代码进行性能测试的方式不够灵活，且很难短时间内模拟大量的并发数，所有作者并不建议使用这种方式。幸运的是 Redis 本身给我们提供了性能测试工具 redis-benchmark（Redis 基准测试），因此我们本文重点来介绍 redis-benchmark 的使用。
基准测试实战
redis-benchmark 位于 Redis 的 src 目录下，我们可以使用 ./redis-benchmark -h 来查看基准测试的使用，执行结果如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 Usage: redis-benchmark [-h <host>] [-p <port>] [-c <clients>] [-n <requests>] [-k <boolean>] -h <hostname> Server hostname (default 127.0.0.1) -p <port> Server port (default 6379) -s <socket> Server socket (overrides host and port) -a <password> Password for Redis Auth -c <clients> Number of parallel connections (default 50) -n <requests> Total number of requests (default 100000) -d <size> Data size of SET/GET value in bytes (default 3) --dbnum <db> SELECT the specified db number (default 0) -k <boolean> 1=keep alive 0=reconnect (default 1) -r <keyspacelen> Use random keys for SET/GET/INCR, random values for SADD Using this option the benchmark will expand the string __rand_int__ inside an argument with a 12 digits number in the specified range from 0 to keyspacelen-1. The substitution changes every time a command is executed. Default tests use this to hit random keys in the specified range. -P <numreq> Pipeline <numreq> requests. Default 1 (no pipeline). -e If server replies with errors, show them on stdout. (no more than 1 error per second is displayed) -q Quiet. Just show query/sec values --csv Output in CSV format -l Loop. Run the tests forever -t <tests> Only run the comma separated list of tests. The test names are the same as the ones produced as output. -I Idle mode. Just open N idle connections and wait.
可以看出 redis-benchmark 支持以下选项：
-h <hostname>：服务器的主机名（默认值为 127.0.0.1）。-p <port>：服务器的端口号（默认值为 6379）。-s <socket>：服务器的套接字（会覆盖主机名和端口号）。-a <password>：登录 Redis 时进行身份验证的密码。-c <clients>：并发的连接数量（默认值为 50）。-n <requests>：发出的请求总数（默认值为 100000）。-d <size>：SET/GET 命令所操作的值的数据大小，以字节为单位（默认值为 2）。–dbnum <db>：选择用于性能测试的数据库的编号（默认值为 0）。-k <boolean>：1 = 保持连接；0 = 重新连接（默认值为 1）。-r <keyspacelen>：SET/GET/INCR 命令使用随机键，SADD 命令使用随机值。通过这个选项，基准测试会将参数中的 __rand_int__ 字符串替换为一个 12 位的整数，这个整数的取值范围从 0 到 keyspacelen-1。每次执行一条命令的时候，用于替换的整数值都会改变。通过这个参数，默认的测试方案会在指定范围之内尝试命中随机键。-P <numreq>：使用管道机制处理 <numreq> 条 Redis 请求。默认值为 1（不使用管道机制）。-q：静默测试，只显示 QPS 的值。–csv：将测试结果输出为 CSV 格式的文件。-l：循环测试。基准测试会永远运行下去。-t <tests>：基准测试只会运行列表中用逗号分隔的命令。测试命令的名称和结果输出产生的名称相同。-I：空闲模式，只会打开 N 个空闲的连接，然后等待。
可以看出 redis-benchmark 带的功能还是比较全的。
基本使用 在安装 Redis 服务端的机器上，我们可以不带任何参数直接执行 ./redis-benchmark 执行结果如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 [@iZ2ze0nc5n41zomzyqtksmZ:src]$ ./redis-benchmark ====== PING_INLINE ====== 100000 requests completed in 1.26 seconds 50 parallel clients 3 bytes payload keep alive: 1 99.81% <= 1 milliseconds 100.00% <= 2 milliseconds 79302.14 requests per second ====== PING_BULK ====== 100000 requests completed in 1.29 seconds 50 parallel clients 3 bytes payload keep alive: 1 99.83% <= 1 milliseconds 100.00% <= 1 milliseconds 77459.34 requests per second ====== SET ====== 100000 requests completed in 1.26 seconds 50 parallel clients 3 bytes payload keep alive: 1 99.80% <= 1 milliseconds 99.99% <= 2 milliseconds 100.00% <= 2 milliseconds 79239.30 requests per second ====== GET ====== 100000 requests completed in 1.19 seconds 50 parallel clients 3 bytes payload keep alive: 1 99.72% <= 1 milliseconds 99.95% <= 15 milliseconds 100.00% <= 16 milliseconds 100.00% <= 16 milliseconds 84104.29 requests per second ====== INCR ====== 100000 requests completed in 1.17 seconds 50 parallel clients 3 bytes payload keep alive: 1 99.86% <= 1 milliseconds 100.00% <= 1 milliseconds 85397.09 requests per second ====== LPUSH ====== 100000 requests completed in 1.22 seconds 50 parallel clients 3 bytes payload keep alive: 1 99.79% <= 1 milliseconds 100.00% <= 1 milliseconds 82169.27 requests per second ====== RPUSH ====== 100000 requests completed in 1.22 seconds 50 parallel clients 3 bytes payload keep alive: 1 99.71% <= 1 milliseconds 100.00% <= 1 milliseconds 81900.09 requests per second ====== LPOP ====== 100000 requests completed in 1.29 seconds 50 parallel clients 3 bytes payload keep alive: 1 99.78% <= 1 milliseconds 99.95% <= 13 milliseconds 99.97% <= 14 milliseconds 100.00% <= 14 milliseconds 77399.38 requests per second ====== RPOP ====== 100000 requests completed in 1.25 seconds 50 parallel clients 3 bytes payload keep alive: 1 99.82% <= 1 milliseconds 100.00% <= 1 milliseconds 80192.46 requests per second ====== SADD ====== 100000 requests completed in 1.25 seconds 50 parallel clients 3 bytes payload keep alive: 1 99.74% <= 1 milliseconds 100.00% <= 1 milliseconds 80192.46 requests per second ====== HSET ====== 100000 requests completed in 1.21 seconds 50 parallel clients 3 bytes payload keep alive: 1 99.86% <= 1 milliseconds 100.00% <= 1 milliseconds 82440.23 requests per second ====== SPOP ====== 100000 requests completed in 1.22 seconds 50 parallel clients 3 bytes payload keep alive: 1 99.92% <= 1 milliseconds 100.00% <= 1 milliseconds 81699.35 requests per second ====== LPUSH (needed to benchmark LRANGE) ====== 100000 requests completed in 1.26 seconds 50 parallel clients 3 bytes payload keep alive: 1 99.69% <= 1 milliseconds 99.95% <= 13 milliseconds 99.99% <= 14 milliseconds 100.00% <= 14 milliseconds 79176.56 requests per second ====== LRANGE_100 (first 100 elements) ====== 100000 requests completed in 1.25 seconds 50 parallel clients 3 bytes payload keep alive: 1 99.57% <= 1 milliseconds 99.98% <= 2 milliseconds 100.00% <= 2 milliseconds 80128.20 requests per second ====== LRANGE_300 (first 300 elements) ====== 100000 requests completed in 1.25 seconds 50 parallel clients 3 bytes payload keep alive: 1 99.91% <= 1 milliseconds 100.00% <= 1 milliseconds 80064.05 requests per second ====== LRANGE_500 (first 450 elements) ====== 100000 requests completed in 1.30 seconds 50 parallel clients 3 bytes payload keep alive: 1 99.78% <= 1 milliseconds 100.00% <= 1 milliseconds 76863.95 requests per second ====== LRANGE_600 (first 600 elements) ====== 100000 requests completed in 1.20 seconds 50 parallel clients 3 bytes payload keep alive: 1 99.85% <= 1 milliseconds 100.00% <= 1 milliseconds 83263.95 requests per second ====== MSET (10 keys) ====== 100000 requests completed in 1.27 seconds 50 parallel clients 3 bytes payload keep alive: 1 99.65% <= 1 milliseconds 100.00% <= 1 milliseconds 78740.16 requests per second
可以看出以上都是对常用的方法 Set、Get、Incr 等进行测试，基本能达到每秒 8W 的处理级别。
精简测试 我们可以使用 ./redis-benchmark -t set,get,incr -n 1000000 -q 命令，来对 Redis 服务器进行精简测试，测试结果如下：
1 2 3 4 [@iZ2ze0nc5n41zomzyqtksmZ:src]$ ./redis-benchmark -t set,get,incr -n 1000000 -q SET: 81726.05 requests per second GET: 81466.40 requests per second INCR: 82481.03 requests per second
可以看出以上测试展示的结果非常的精简，这是因为我们设置了 -q 参数，此选项的意思是设置输出结果为精简模式，其中 -t 表示指定测试指令，-n 设置每个指令测试 100w 次。
管道测试 本课程的前面章节介绍了 Pipeline（管道）的知识，它是用于客户端把命令批量发给服务器端执行的，以此来提高程序的整体执行效率，那接下来我们测试一下 Pipeline 的吞吐量能到达多少，执行命令如下：
1 2 3 4 [@iZ2ze0nc5n41zomzyqtksmZ:src]$ ./redis-benchmark -t set,get,incr -n 1000000 -q -P 10 SET: 628535.50 requests per second GET: 654450.25 requests per second INCR: 647249.19 requests per second
我们发现 Pipeline 的测试很快就执行完了，同样是每个指令执行 100w 次，可以看出 Pipeline 的性能几乎是普通命令的 8 倍 ， -P 10 表示每次执行 10 个 Redis 命令。
基准测试的影响元素
为什么每次执行 10 个 Redis 命令，Pipeline 的效率为什么达不到普通命令的 10 倍呢？
这是因为基准测试会受到很大外部因素的影响，例如以下几个：
网络带宽和网络延迟可能是 Redis 操作最大的性能瓶颈，比如有 10w q/s，平均每个请求负责传输 8 KB 的字符，那我们需要的理论带宽是 7.6 Gbits/s，如果服务器配置的是 1 Gbits/s，那么一定会有很多信息在排队等候传输，因此运行效率可想而知，这也是很多 Redis 生产坏境之所以效率不高的原因；
CPU 可能是 Redis 运行的另一个重要的影响因素，如果 CPU 的计算能力跟不上 Redis 要求的话，也会影响 Redis 的运行效率；
如果 Redis 运行在虚拟设备上，性能也会受影响，因为普通操作在虚拟设备上会有额外的消耗；
普通操作和批量操作（Pipeline）对 Redis 的吞吐量也有很大的影响。
小结
本文介绍了 Redis 自带的性能测试工具 redis-benchmark 也是 Redis 主流的性能测试工具，我们可以轻松模拟指定并发量和指定命令的测试条件，也可以模拟管道测试。测试的结果对于我们做技术选型、版本选择以及数据类型的选择上都有一定的指导意义，但需要注意 Redis 的吞吐量还受到其他因素的影响，例如带宽、CPU 等因素
34 实战：Redis 慢查询
Redis 慢查询作用和 MySQL 慢查询作用类似，都是为我们查询出不合理的执行命令，然后让开发人员和运维人员一起来规避这些耗时的命令，从而让服务器更加高效和健康的运行。对于单线程的 Redis 来说，不合理的使用更是致命的，因此掌握 Redis 慢查询技能对我们来说非常的关键。
如何进行慢查询？
在开始之前，我们先要了解一下 Redis 中和慢查询相关的配置项，Redis 慢查询重要的配置项有以下两个：
slowlog-log-slower-than：用于设置慢查询的评定时间，也就是说超过此配置项的命令，将会被当成慢操作记录在慢查询日志中，它执行单位是微秒（1 秒等于 1000000 微秒）；
slowlog-max-len：用来配置慢查询日志的最大记录数。
我们先来看它们的默认配置值：
1 2 3 4 5 6 127.0.0.1:6379> config get slowlog-log-slower-than #慢查询判断时间 1) "slowlog-log-slower-than" 2) "10000" 127.0.0.1:6379> config get slowlog-max-len #慢查询最大记录条数 1) "slowlog-max-len" 2) "128"
可以看出慢查询的临界值是 10000 微秒，默认保存 128 条慢查询记录。
修改配置项 slowlog-log-slower-than 和 slowlog-max-len 可以通过 config set xxx 的模式来修改，例如 config set slowlog-max-len 200 设置慢查询最大记录数为 200 条。
慢查询演示 我们先来设置慢查询的判断时间为 0 微秒，这样所有的执行命令都会被记录，设置命令如下：
1 2 127.0.0.1:6379> config set slowlog-log-slower-than 0 OK
接下来我们执行两条插入命令：
1 2 3 4 127.0.0.1:6379> set msg xiaoming OK 127.0.0.1:6379> set lang java OK
最后我们使用 slowlog show 来查询慢日志，结果如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 127.0.0.1:6379> slowlog get #慢日志查询 1) 1) (integer) 2 #慢日志下标 2) (integer) 1581994139 #执行时间 3) (integer) 5 #花费时间 (单位微秒) 4) 1) "set" #执行的具体命令 2) "lang" 3) "java" 5) "127.0.0.1:47068" 6) "" 2) 1) (integer) 1 2) (integer) 1581994131 3) (integer) 6 4) 1) "set" 2) "msg" 3) "xiaoming" 5) "127.0.0.1:47068" 6) "" 3) 1) (integer) 0 2) (integer) 1581994093 3) (integer) 5 4) 1) "config" 2) "set" 3) "slowlog-log-slower-than" 4) "0" 5) "127.0.0.1:47068" 6) ""
加上本身的设置命令一共有三条“慢操作”记录，按照插入的顺序倒序存入慢查询日志中。
小贴士：当慢查询日志超过设定的最大存储条数之后，会把最早的执行命令依次舍弃。
慢查询其他相关命令
查询指定条数慢日志 语法：slowlog get n。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 127.0.0.1:6379> slowlog get 2 #查询两条 1) 1) (integer) 20 2) (integer) 1581997567 3) (integer) 14 4) 1) "slowlog" 2) "get" 3) "4" 5) "127.0.0.1:47068" 6) "" 2) 1) (integer) 19 2) (integer) 1581997544 3) (integer) 11 4) 1) "slowlog" 2) "get" 3) "3" 5) "127.0.0.1:47068" 6) "" 127.0.0.1:6379> slowlog get 3 #查询三条 1) 1) (integer) 22 2) (integer) 1581997649 3) (integer) 25 4) 1) "set" 2) "msg" 3) "hi" 5) "127.0.0.1:47068" 6) "" 2) 1) (integer) 21 2) (integer) 1581997613 3) (integer) 9 4) 1) "slowlog" 2) "get" 3) "2" 5) "127.0.0.1:47068" 6) "" 3) 1) (integer) 20 2) (integer) 1581997567 3) (integer) 14 4) 1) "slowlog" 2) "get" 3) "4" 5) "127.0.0.1:47068" 6) ""
获取慢查询队列长度 语法：slowlog len。
1 2 127.0.0.1:6379> slowlog len (integer) 16
清空慢查询日志 使用 slowlog reset 来清空所有的慢查询日志，执行命令如下：
1 2 127.0.0.1:6379> slowlog reset OK
代码实战
本文我们使用 Java 来实现慢查询日志的操作，代码如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 import redis.clients.jedis.Jedis;import redis.clients.jedis.util.Slowlog;import utils.JedisUtils;import java.util.List;public class SlowExample { public static void main (String[] args) { Jedis jedis = JedisUtils.getJedis(); jedis.set("db" , "java" ); jedis.set("lang" , "java" ); long logLen = jedis.slowlogLen(); List<Slowlog> list = jedis.slowlogGet(); for (Slowlog item : list) { System.out.println("慢查询命令：" + item.getArgs()+ " 执行了：" +item.getExecutionTime()+" 微秒" ); } jedis.slowlogReset(); } }
以上代码执行结果如下：
1 2 3 4 5 慢查询命令：[SLOWLOG, len] 执行了：1 微秒 慢查询命令：[SET, lang, java] 执行了：2 微秒 慢查询命令：[SET, db, java] 执行了：4 微秒 慢查询命令：[SLOWLOG, reset] 执行了：155 微秒
小结
本文我们介绍了慢查询相关的两个重要参数 slowlog-log-slower-than（用于设置慢查询的评定时间）和 slowlog-max-len 用来配置慢查询日志的最大记录数，然后通过修改 config set slowlog-log-slower-than 0 把所有操作都记录在慢日志进行相关测试。我们可以使用 slowlog get [n] 查询慢操作日志，使用 slowlog reset 清空慢查询日志。最后给大家一个建议，可以定期检查慢查询日志，及时发现和改进 Redis 运行中不合理的操作。
35 实战：Redis 性能优化方案
Redis 是基于单线程模型实现的，也就是 Redis 是使用一个线程来处理所有的客户端请求的，尽管 Redis 使用了非阻塞式 IO，并且对各种命令都做了优化（大部分命令操作时间复杂度都是 O(1)），但由于 Redis 是单线程执行的特点，因此它对性能的要求更加苛刻，本文我们将通过一些优化手段，让 Redis 更加高效的运行。
本文我们将使用以下手段，来提升 Redis 的运行速度：
缩短键值对的存储长度；
使用 lazy free（延迟删除）特性；
设置键值的过期时间；
禁用耗时长的查询命令；
使用 slowlog 优化耗时命令；
使用 Pipeline 批量操作数据；
避免大量数据同时失效；
客户端使用优化；
限制 Redis 内存大小；
使用物理机而非虚拟机安装 Redis 服务；
检查数据持久化策略；
使用分布式架构来增加读写速度。
缩短键值对的存储长度
键值对的长度是和性能成反比的，比如我们来做一组写入数据的性能测试，执行结果如下：
数据量
key 大小
value 大小
string:set 平均耗时
hash:hset 平均耗时
100w
20byte
512byte
1.13 微秒
10.28 微秒
100w
20byte
200byte
0.74 微秒
8.08 微秒
100w
20byte
100byte
0.65 微秒
7.92 微秒
100w
20byte
50byte
0.59 微秒
6.74 微秒
100w
20byte
20byte
0.55 微秒
6.60 微秒
100w
20byte
5byte
0.53 微秒
6.53 微秒
从以上数据可以看出，在 key 不变的情况下，value 值越大操作效率越慢，因为 Redis 对于同一种数据类型会使用不同的内部编码进行存储，比如字符串的内部编码就有三种：int（整数编码）、raw（优化内存分配的字符串编码）、embstr（动态字符串编码），这是因为 Redis 的作者是想通过不同编码实现效率和空间的平衡，然而数据量越大使用的内部编码就越复杂，而越是复杂的内部编码存储的性能就越低。
这还只是写入时的速度，当键值对内容较大时，还会带来另外几个问题：
内容越大需要的持久化时间就越长，需要挂起的时间越长，Redis 的性能就会越低；
内容越大在网络上传输的内容就越多，需要的时间就越长，整体的运行速度就越低；
内容越大占用的内存就越多，就会更频繁地触发内存淘汰机制，从而给 Redis 带来了更多的运行负担。
因此在保证完整语义的同时，我们要尽量地缩短键值对的存储长度，必要时要对数据进行序列化和压缩再存储，以 Java 为例，序列化我们可以使用 protostuff 或 kryo，压缩我们可以使用 snappy。
使用 lazy free 特性
lazy free 特性是 Redis 4.0 新增的一个非常实用的功能，它可以理解为惰性删除或延迟删除。意思是在删除的时候提供异步延时释放键值的功能，把键值释放操作放在 BIO（Background I/O）单独的子线程处理中，以减少删除对 Redis 主线程的阻塞，可以有效地避免删除 big key 时带来的性能和可用性问题。
lazy free 对应了 4 种场景，默认都是关闭的：
1 2 3 4 lazyfree-lazy-eviction no lazyfree-lazy-expire no lazyfree-lazy-server-del no slave-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 会帮你自动清除过期的键值对，以节约对内存的占用，以避免键值过多的堆积，频繁的触发内存淘汰策略。
禁用耗时长的查询命令
Redis 绝大多数读写命令的时间复杂度都在 O(1) 到 O(N) 之间，在官方文档对每命令都有时间复杂度说明，地址：
https://redis.io/commands
其中 O(1) 表示可以安全使用的，而 O(N) 就应该当心了，N 表示不确定，数据越大查询的速度可能会越慢。因为 Redis 只用一个线程来做数据查询，如果这些指令耗时很长，就会阻塞 Redis，造成大量延时。
要避免 O(N) 命令对 Redis 造成的影响，可以从以下几个方面入手改造：
决定禁止使用 keys 命令；
避免一次查询所有的成员，要使用 scan 命令进行分批的，游标式的遍历；
通过机制严格控制 Hash、Set、Sorted Set 等结构的数据大小；
将排序、并集、交集等操作放在客户端执行，以减少 Redis 服务器运行压力；
删除（del）一个大数据的时候，可能会需要很长时间，所以建议用异步删除的方式 unlink，它会启动一个新的线程来删除目标数据，而不阻塞 Redis 的主线程。
使用 slowlog 优化耗时命令
我们可以使用 slowlog 功能找出最耗时的 Redis 命令进行相关的优化，以提升 Redis 的运行速度，慢查询有两个重要的配置项：
slowlog-log-slower-than：用于设置慢查询的评定时间，也就是说超过此配置项的命令，将会被当成慢操作记录在慢查询日志中，它执行单位是微秒（1 秒等于 1000000 微秒）；
slowlog-max-len：用来配置慢查询日志的最大记录数。
我们可以根据实际的业务情况进行相应的配置，其中慢日志是按照插入的顺序倒序存入慢查询日志中，我们可以使用 slowlog get n 来获取相关的慢查询日志，再找到这些慢查询对应的业务进行相关的优化。
使用 Pipeline 批量操作数据
Pipeline（管道技术）是客户端提供的一种批处理技术，用于一次处理多个 Redis 命令，从而提高整个交互的性能。
我们使用 Java 代码来测试一下 Pipeline 和普通操作的性能对比，Pipeline 的测试代码如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 public class PipelineExample { public static void main (String[] args) { Jedis jedis = new Jedis ("127.0.0.1" , 6379 ); long beginTime = System.currentTimeMillis(); Pipeline pipe = jedis.pipelined(); for (int i = 0 ; i < 100 ; i++) { pipe.set("key" + i, "val" + i); pipe.del("key" +i); } pipe.sync(); long endTime = System.currentTimeMillis(); System.out.println("执行耗时：" + (endTime - beginTime) + "毫秒" ); } }
以上程序执行结果为：
普通的操作代码如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 public class PipelineExample { public static void main (String[] args) { Jedis jedis = new Jedis ("127.0.0.1" , 6379 ); long beginTime = System.currentTimeMillis(); for (int i = 0 ; i < 100 ; i++) { jedis.set("key" + i, "val" + i); jedis.del("key" +i); } long endTime = System.currentTimeMillis(); System.out.println("执行耗时：" + (endTime - beginTime) + "毫秒" ); } }
以上程序执行结果为：
从以上的结果可以看出，管道的执行时间是 297 毫秒，而普通命令执行时间是 17276 毫秒，管道技术要比普通的执行大约快了 58 倍。
避免大量数据同时失效
Redis 过期键值删除使用的是贪心策略，它每秒会进行 10 次过期扫描，此配置可在 redis.conf 进行配置，默认值是 hz 10，Redis 会随机抽取 20 个值，删除这 20 个键中过期的键，如果过期 key 的比例超过 25%，重复执行此流程，如下图所示：
如果在大型系统中有大量缓存在同一时间同时过期，那么会导致 Redis 循环多次持续扫描删除过期字典，直到过期字典中过期键值被删除的比较稀疏为止，而在整个执行过程会导致 Redis 的读写出现明显的卡顿，卡顿的另一种原因是内存管理器需要频繁回收内存页，因此也会消耗一定的 CPU。
为了避免这种卡顿现象的产生，我们需要预防大量的缓存在同一时刻一起过期，最简单的解决方案就是在过期时间的基础上添加一个指定范围的随机数。
客户端使用优化
在客户端的使用上我们除了要尽量使用 Pipeline 的技术外，还需要注意要尽量使用 Redis 连接池，而不是频繁创建销毁 Redis 连接，这样就可以减少网络传输次数和减少了非必要调用指令。
限制 Redis 内存大小
在 64 位操作系统中 Redis 的内存大小是没有限制的，也就是配置项 maxmemory <bytes> 是被注释掉的，这样就会导致在物理内存不足时，使用 swap 空间既交换空间，而当操心系统将 Redis 所用的内存分页移至 swap 空间时，将会阻塞 Redis 进程，导致 Redis 出现延迟，从而影响 Redis 的整体性能。因此我们需要限制 Redis 的内存大小为一个固定的值，当 Redis 的运行到达此值时会触发内存淘汰策略，内存淘汰策略在 Redis 4.0 之后有 8 种 ：
noeviction ：不淘汰任何数据，当内存不足时，新增操作会报错，Redis 默认内存淘汰策略；allkeys-lru ：淘汰整个键值中最久未使用的键值；allkeys-random ：随机淘汰任意键值;volatile-lru ：淘汰所有设置了过期时间的键值中最久未使用的键值；volatile-random ：随机淘汰设置了过期时间的任意键值；volatile-ttl ：优先淘汰更早过期的键值。
在 Redis 4.0 版本中又新增了 2 种淘汰策略：
volatile-lfu ：淘汰所有设置了过期时间的键值中，最少使用的键值；allkeys-lfu ：淘汰整个键值中最少使用的键值。
其中 allkeys-xxx 表示从所有的键值中淘汰数据，而 volatile-xxx 表示从设置了过期键的键值中淘汰数据。
我们可以根据实际的业务情况进行设置，默认的淘汰策略不淘汰任何数据，在新增时会报错。
使用物理机而非虚拟机
在虚拟机中运行 Redis 服务器，因为和物理机共享一个物理网口，并且一台物理机可能有多个虚拟机在运行，因此在内存占用上和网络延迟方面都会有很糟糕的表现，我们可以通过 ./redis-cli --intrinsic-latency 100 命令查看延迟时间，如果对 Redis 的性能有较高要求的话，应尽可能在物理机上直接部署 Redis 服务器。
检查数据持久化策略
Redis 的持久化策略是将内存数据复制到硬盘上，这样才可以进行容灾恢复或者数据迁移，但维护此持久化的功能，需要很大的性能开销。
在 Redis 4.0 之后，Redis 有 3 种持久化的方式：
RDB（Redis DataBase，快照方式）将某一个时刻的内存数据，以二进制的方式写入磁盘；
AOF（Append Only File，文件追加方式），记录所有的操作命令，并以文本的形式追加到文件中；
混合持久化方式，Redis 4.0 之后新增的方式，混合持久化是结合了 RDB 和 AOF 的优点，在写入的时候，先把当前的数据以 RDB 的形式写入文件的开头，再将后续的操作命令以 AOF 的格式存入文件，这样既能保证 Redis 重启时的速度，又能减低数据丢失的风险。
RDB 和 AOF 持久化各有利弊，RDB 可能会导致一定时间内的数据丢失，而 AOF 由于文件较大则会影响 Redis 的启动速度，为了能同时拥有 RDB 和 AOF 的优点，Redis 4.0 之后新增了混合持久化的方式，因此我们在必须要进行持久化操作时，应该选择混合持久化的方式。
查询是否开启混合持久化可以使用 config get aof-use-rdb-preamble 命令。
其中 yes 表示已经开启混合持久化，no 表示关闭，Redis 5.0 默认值为 yes。
1 2 3 127.0.0.1:6379> config get aof-use-rdb-preamble 1) "aof-use-rdb-preamble" 2) "yes"
如果是其他版本的 Redis 首先需要检查一下，是否已经开启了混合持久化，如果关闭的情况下，可以通过以下两种方式开启：
通过命令行开启
通过修改 Redis 配置文件开启
通过命令行开启 使用命令 config set aof-use-rdb-preamble yes 执行结果如下图所示：
1 2 127.0.0.1:6379> config set aof-use-rdb-preamble yes OK
命令行设置配置的缺点是重启 Redis 服务之后，设置的配置就会失效。
通过修改 Redis 配置文件开启 在 Redis 的根路径下找到 redis.conf 文件，把配置文件中的 aof-use-rdb-preamble no 改为 aof-use-rdb-preamble yes 如下图所示：
配置完成之后，需要重启 Redis 服务器，配置才能生效，但修改配置文件的方式，在每次重启 Redis 服务之后，配置信息不会丢失。
需要注意的是，在非必须进行持久化的业务中，可以关闭持久化，这样可以有效地提升 Redis 的运行速度，不会出现间歇性卡顿的困扰。
使用分布式架构来增加读写速度
Redis 分布式架构有三个重要的手段：
主从同步
哨兵模式
Redis Cluster 集群
使用主从同步功能我们可以把写入放到主库上执行，把读功能转移到从服务上，因此就可以在单位时间内处理更多的请求，从而提升的 Redis 整体的运行速度。
而哨兵模式是对于主从功能的升级，但当主节点奔溃之后，无需人工干预就能自动恢复 Redis 的正常使用。
Redis Cluster 是 Redis 3.0 正式推出的，Redis 集群是通过将数据分散存储到多个节点上，来平衡各个节点的负载压力。
Redis Cluster 采用虚拟哈希槽分区，所有的键根据哈希函数映射到 0~16383 整数槽内，计算公式：
1 slot = CRC16(key) & 16383
每一个节点负责维护一部分槽以及槽所映射的键值数据。这样 Redis 就可以把读写压力从一台服务器，分散给多台服务器了，因此性能会有很大的提升。
在这三个功能中，我们只需要使用一个就行了，毫无疑问 Redis Cluster 应该是首选的实现方案，它可以把读写压力自动地分担给更多的服务器，并且拥有自动容灾的能力。
36 实战：Redis 主从同步
主从同步（主从复制）是 Redis 高可用服务的基石，也是多机运行中最基础的一个。我们把主要存储数据的节点叫做主节点 (master），把其他通过复制主节点数据的副本节点叫做从节点 (slave），如下图所示：
在 Redis 中一个主节点可以拥有多个从节点，一个从节点也可以是其他服务器的主节点，如下图所示：
主从同步的优点
主从同步具有以下三个优点：
性能方面：有了主从同步之后，可以把查询任务分配给从服务器，用主服务器来执行写操作，这样极大的提高了程序运行的效率，把所有压力分摊到各个服务器了；
高可用：当有了主从同步之后，当主服务器节点宕机之后，可以很迅速的把从节点提升为主节点，为 Redis 服务器的宕机恢复节省了宝贵的时间；
防止数据丢失：当主服务器磁盘坏掉之后，其他从服务器还保留着相关的数据，不至于数据全部丢失。
既然主从同步有这么多的优点，那接下来我们来看如何开启和使用主从同步功能。
开启主从同步
运行中设置从服务器 在 Redis 运行过程中，我们可以使用 replicaof host port 命令，把自己设置为目标 IP 的从服务器，执行命令如下：
1 2 127.0.0.1:6379> replicaof 127.0.0.1 6380 OK
如果主服务设置了密码，需要在从服务器输入主服务器的密码，使用 config set masterauth 主服务密码 命令的方式，例如：
1 2 127.0.0.1:6377> config set masterauth pwd654321 OK
1. 执行流程
在执行完 replicaof 命令之后，从服务器的数据会被清空，主服务会把它的数据副本同步给从服务器。
2. 测试同步功能
主从服务器设置完同步之后，我们来测试一下主从数据同步，首先我们先在主服务器上执行保存数据操作，再去从服务器查询。
主服务器执行命令：
1 2 127.0.0.1:6379> set lang redis OK
从服务执行查询：
1 2 127.0.0.1:6379> get lang "redis"
可以看出数据已经被正常同步过来了。
启动时设置从服务器 我们可以使用命令 redis-server --port 6380 --replicaof 127.0.0.1 6379 将自己设置成目标服务器的从服务器。
数据同步
完整数据同步 当有新的从服务器连接时，为了保障多个数据库的一致性，主服务器会执行一次 bgsave 命令生成一个 RDB 文件，然后再以 Socket 的方式发送给从服务器，从服务器收到 RDB 文件之后再把所有的数据加载到自己的程序中，就完成了一次全量的数据同步。
部分数据同步 在 Redis 2.8 之前每次从服务器离线再重新上线之前，主服务器会进行一次完整的数据同步，然后这种情况如果发生在离线时间比较短的情况下，只有少量的数据不同步却要同步所有的数据是非常笨拙和不划算的，在 Redis 2.8 这个功能得到了优化。
Redis 2.8 的优化方法是当从服务离线之后，主服务器会把离线之后的写入命令，存储在一个特定大小的队列中，队列是可以保证先进先出的执行顺序的，当从服务器重写恢复上线之后，主服务会判断离线这段时间内的命令是否还在队列中，如果在就直接把队列中的数据发送给从服务器，这样就避免了完整同步的资源浪费。
小贴士：存储离线命令的队列大小默认是 1MB，使用者可以自行修改队列大小的配置项 repl-backlog-size。
无盘数据同步 从前面的内容我们可以得知，在第一次主从连接的时候，会先产生一个 RDB 文件，再把 RDB 文件发送给从服务器，如果主服务器是非固态硬盘的时候，系统的 I/O 操作是非常高的，为了缓解这个问题，Redis 2.8.18 新增了无盘复制功能，无盘复制功能不会在本地创建 RDB 文件，而是会派生出一个子进程，然后由子进程通过 Socket 的方式，直接将 RDB 文件写入到从服务器，这样主服务器就可以在不创建RDB文件的情况下，完成与从服务器的数据同步。
要使用无须复制功能，只需把配置项 repl-diskless-sync 的值设置为 yes 即可，它默认配置值为 no。
查询服务器的角色
我们使用 role 命令，来查询当前服务器的主从角色信息。
主服务查看 在主服务器上执行 role 结果如下：
1 2 3 4 5 6 127.0.0.1:6379> role 1) "master" 2) (integer) 546 3) 1) 1) "172.17.0.1" 2) "6379" 3) "546"
master 表示主服务器，底下是从服务器的 IP、端口和连接时间。
从服务器查看 在从服务器执行 role 命令，执行结果如下：
1 2 3 4 5 6 127.0.0.1:6379> role 1) "slave" 2) "192.168.1.71" 3) (integer) 6380 4) "connected" 5) (integer) 14
slave 表示从服务器，底下主服务器的 IP、端口和连接时间。
关闭主从同步
我们可以使用 replicaof no one 命令来停止从服务器的复制，操作命令如下：
1 2 3 4 5 6 7 8 9 10 11 12 127.0.0.1:6379> role #查询当前角色 1) "slave" #从服务器 2) "192.168.1.71" 3) (integer) 6380 4) "connected" 5) (integer) 14 127.0.0.1:6379> replicaof no one #关闭同步 OK 127.0.0.1:6379> role #查询当前角色 1) "master" #主服务器 2) (integer) 1097 3) (empty list or set)
可以看出执行了 replicaof no one 命令之后，自己就从服务器变成主服务器了。
小贴士：服务器类型的转换并不会影响数据，这台服务器的数据将会被保留。
注意事项
主从同步有一些需要注意的点，我们来看一下。
数据一致性问题 当从服务器已经完成和主服务的数据同步之后，再新增的命令会以异步的方式发送至从服务器，在这个过程中主从同步会有短暂的数据不一致，如在这个异步同步发生之前主服务器宕机了，会造成数据不一致。
从服务器只读性 默认在情况下，处于复制模式的主服务器既可以执行写操作也可以执行读操作，而从服务器则只能执行读操作。
可以在从服务器上执行 config set replica-read-only no 命令，使从服务器开启写模式，但需要注意以下几点：
在从服务器上写的数据不会同步到主服务器；
当键值相同时主服务器上的数据可以覆盖从服务器；
在进行完整数据同步时，从服务器数据会被清空。
复制命令的变化 Redis 5.0 之前使用的复制命令是 slaveof，在 Redis 5.0 之后复制命令才被改为 replicaof，在高版本（Redis 5+）中我们应该尽量使用 replicaof，因为 slaveof 命令可能会被随时废弃掉。
小结
本文我们了解了 Redis 多机运行的基础功能主从同步，主从同步可以通过 replicaof host port 命令开启，知道了同步的三种方式：完整数据同步（第一次全量 RDB 同步），部分数据同步（Redis 2.8 对于短时间离线的同步功能优化），无盘同步（非 RDB 生成的方式同步数据），我们也可以使用 replicaof no one 命令来停止从服务器的复制功能。
37 实战：Redis哨兵模式（上）
上一篇我们讲了主从复制模式，它是属于 Redis 多机运行的基础，但这种模式本身存在一个致命的问题，当主节点奔溃之后，需要人工干预才能恢复 Redis 的正常使用。
例如，我们有 3 台服务器做了主从复制，一个主服务器 A 和两个从服务器 B、C，当 A 发生故障之后，需要人工把 B 服务器设置为主服务器，同时再去 C 服务器设置成从服务器并且从主服务器 B 同步数据，如果是发生在晚上或者从服务器节点很多的情况下，对于人工来说想要立即实现恢复的难度很多，所以我们需要一个自动的工具——Redis Sentinel（哨兵模式）来把手动的过程变成自动的，让 Redis 拥有自动容灾恢复（failover）的能力。
哨兵模式如下所示：
小贴士：Redis Sentinel 的最小分配单位是一主一从。
Redis Sentinel 搭建
Redis 官方提供了 Redis Sentinel 的功能，它的运行程序保存在 src 目录下.
我们需要使用命令 ./src/redis-sentinel sentinel.conf 来启动 Sentinel，可以看出我们在启动它时必须设置一个 sentinel.conf 文件，这个配置文件中必须包含监听的主节点信息：
1 sentinel monitor master-name ip port quorum
例如：
1 sentinel monitor mymaster 127.0 .0 .1 6379 1
其中：
master-name 表示给监视的主节点起一个名称；
ip 表示主节点的 IP；
port 表示主节点的端口；
quorum 表示确认主节点下线的 Sentinel 数量，如果 quorum 设置为 1 表示只要有一台 Sentinel 判断它下线了，就可以确认它真的下线了。
注意：如果主节点 Redis 服务器有密码，还必须在 sentinel.conf 中添加主节点的密码，不然会导致 Sentinel 不能自动监听到主节点下面的从节点。
所以如果 Redis 有密码，sentinel.conf 必须包含以下内容：
1 2 sentinel monitor mymaster 127.0 .0 .1 6379 1 sentinel auth-pass mymaster pwd654321
当我们配置好 sentinel.conf 并执行启动命令 ./src/redis-sentinel sentinel.conf 之后，Redis Sentinel 就会被启动，如下图所示：
从上图可以看出 Sentinel 只需配置监听主节点的信息，它会自动监听对应的从节点。
启动 Sentinel 集群
上面我们演示了单个 Sentinel 的启动，但生产环境我们不会只启动一台 Sentinel，因为如果启动一台 Sentinel 假如它不幸宕机的话，就不能提供自动容灾的服务了，不符合我们高可用的宗旨，所以我们会在不同的物理机上启动多个 Sentinel 来组成 Sentinel 集群，来保证 Redis 服务的高可用。
启动 Sentinel 集群的方法很简单，和上面启动单台的方式一样，我们只需要把多个 Sentinel 监听到一个主服务器节点，那么多个 Sentinel 就会自动发现彼此，并组成一个 Sentinel 集群。
我们启动第二个 Sentinel 来试一下，执行结果如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 [@iZ2ze0nc5n41zomzyqtksmZ:redis2]$ ./src/redis-sentinel sentinel.conf 5547:X 19 Feb 2020 20:29:30.047 # oO0OoO0OoO0Oo Redis is starting oO0OoO0OoO0Oo 5547:X 19 Feb 2020 20:29:30.047 # Redis version=5.0.5, bits=64, commit=00000000, modified=0, pid=5547, just started 5547:X 19 Feb 2020 20:29:30.047 # Configuration loaded _._ _.-``__ ''-._ _.-`` `. `_. ''-._ Redis 5.0.5 (00000000/0) 64 bit .-`` .-```. ```\/ _.,_ ''-._ ( ' , .-` | `, ) Running in sentinel mode |`-._`-...-` __...-.``-._|'` _.-'| Port: 26377 | `-._ `._ / _.-' | PID: 5547 `-._ `-._ `-./ _.-' _.-' |`-._`-._ `-.__.-' _.-'_.-'| | `-._`-._ _.-'_.-' | http://redis.io `-._ `-._`-.__.-'_.-' _.-' |`-._`-._ `-.__.-' _.-'_.-'| | `-._`-._ _.-'_.-' | `-._ `-._`-.__.-'_.-' _.-' `-._ `-.__.-' _.-' `-._ _.-' `-.__.-' 5547:X 19 Feb 2020 20:29:30.049 # WARNING: The TCP backlog setting of 511 cannot be enforced because /proc/sys/net/core/somaxconn is set to the lower value of 128. 5547:X 19 Feb 2020 20:29:30.049 # Sentinel ID is 6455f2f74614a71ce0a63398b2e48d6cd1cf0d06 5547:X 19 Feb 2020 20:29:30.049 # +monitor master mymaster 127.0.0.1 6379 quorum 1 5547:X 19 Feb 2020 20:29:30.049 * +slave slave 127.0.0.1:6377 127.0.0.1 6377 @ mymaster 127.0.0.1 6379 5547:X 19 Feb 2020 20:29:30.052 * +slave slave 127.0.0.1:6378 127.0.0.1 6378 @ mymaster 127.0.0.1 6379 5547:X 19 Feb 2020 20:29:30.345 * +sentinel sentinel 6455f2f74614a71ce0a63398b2e48d6cd1cf0d08 127.0.0.1 26379 @ mymaster 127.0.0.1 6379
从以上启动命令可以看出，比单机模式多了最后一行发现其他 Sentinel 服务器的命令，说明这两个 Sentinel 已经组成一个集群了。
Sentinel 集群示意图如下：
一般情况下 Sentinel 集群的数量取大于 1 的奇数，例如 3、5、7、9，而 quorum 的配置要根据 Sentinel 的数量来发生变化，例如 Sentinel 是 3 台，那么对应的 quorum 最好是 2，如果 Sentinel 是 5 台，那么 quorum 最好是 3，它表示当有 3 台 Sentinel 都确认主节点下线了，就可以确定主节点真的下线了。
与 quorum 参数相关的有两个概念：主观下线和客观下线。
当 Sentinel 集群中，有一个 Sentinel 认为主服务器已经下线时，它会将这个主服务器标记为主观下线（Subjectively Down，SDOWN），然后询问集群中的其他 Sentinel，是否也认为该服务器已下线，当同意主服务器已下线的 Sentinel 数量达到 quorum 参数所指定的数量时，Sentinel 就会将相应的主服务器标记为客观下线（Objectively down，ODOWN），然后开始对其进行故障转移。
自动容灾测试
前面我们已经搭建了 Redis Sentinel，接下来我们就尝试一下自动容灾的功能，为了模拟故障我们先把主节点手动 kill 掉，执行命令如下：
1 2 3 4 5 6 7 8 9 10 [@iZ2ze0nc5n41zomzyqtksmZ:~]$ ps -ef|grep redis #找到主节点的进程id root 5186 1 0 16:54 ? 00:00:23 ./src/redis-server *:6377 root 5200 1 0 16:56 ? 00:00:22 ./src/redis-server *:6378 root 5304 5287 0 17:31 pts/2 00:00:00 redis-cli -a pwd654321 root 5395 5255 0 18:26 pts/1 00:00:19 ./src/redis-sentinel *:26379 [sentinel] root 5547 5478 0 20:29 pts/4 00:00:02 ./src/redis-sentinel *:26377 [sentinel] root 5551 5517 0 20:29 pts/5 00:00:00 redis-cli -h 127.0.0.1 -p 26377 -a pwd654321 root 5568 5371 0 20:48 pts/0 00:00:00 grep --color=auto redis root 28517 1 0 Feb13 ? 00:15:33 ./src/redis-server *:6379 [@iZ2ze0nc5n41zomzyqtksmZ:~]$ kill -9 28517 #关闭主节点服务
这个时候我们在连接上另一台 Redis 服务器，查看当前主从服务器信息，执行命令如下：
1 2 3 4 5 6 7 [@iZ2ze0nc5n41zomzyqtksmZ:~]$ redis-cli -h 127.0.0.1 -p 6377 -a pwd654321 2>/dev/null 127.0.0.1:6377> role 1) "master" 2) (integer) 770389 3) 1) 1) "127.0.0.1" 2) "6378" 3) "770389"
可以看出之前的从服务 6377 被提升为主服务器了，还剩下一台从服务 6378，而之前的主服务器 6379 被我们手动下线了，可以看出 Sentinel 已经完美的完成的它的故障自动转移的任务。
主服务竞选规则
上面我们模拟了 Redis Sentinel 自动容灾恢复，那接下来我们来看一下，主服务器竞选的规则和相关设置项。
新主节点竞选优先级设置 我们可以 redis.conf 中的 replica-priority 选项来设置竞选新主节点的优先级，它的默认值是 100，它的最大值也是 100，这个值越小它的权重就越高，例如从节点 A 的 replica-priority 值为 100，从节点 B 的值为 50，从节点 C 的值为 5，那么在竞选时从节点 C 会作为新的主节点。
新主节点竞选规则 新主节点的竞选会排除不符合条件的从节点，然后再剩余的从节点按照优先级来挑选。首先来说，存在以下条件的从节点会被排除：
排除所有已经下线以及长时间没有回复心跳检测的疑似已下线从服务器；
排除所有长时间没有与主服务器通信，数据状态过时的从服务器；
排除所有优先级（replica-priority）为 0 的服务器。
符合条件的从节点竞选顺序：
优先级最高的从节点将会作为新主节点；
优先级相等则判断复制偏移量，偏移量最大的从节点获胜；
如果以上两个条件都相同，选择 Redis 运行时随机生成 ID 最小那个为新的主服务器。
旧主节点恢复上线 如果之前的旧主节点恢复上线，会作为从节点运行在主从服务器模式中。
哨兵工作原理
哨兵的工作原理是这样的，首先每个 Sentinel 会以每秒钟 1 次的频率，向已知的主服务器、从服务器和以及其他 Sentinel 实例，发送一个 PING 命令。
如果最后一次有效回复 PING 命令的时间超过 down-after-milliseconds 所配置的值（默认 30s），那么这个实例会被 Sentinel 标记为主观下线。
如果一个主服务器被标记为主观下线，那么正在监视这个主服务器的所有 Sentinel 节点，要以每秒 1 次的频率确认 主服务器的确进入了主观下线状态。
如果有足够数量（quorum 配置值）的 Sentinel 在指定的时间范围内同意这一判断，那么这个主服务器被标记为客观下线。此时所有的 Sentinel 会按照规则协商自动选出新的主节点。
注意：一个有效的 PING 回复可以是：+PONG、-LOADING 或者 -MASTERDOWN。如果返回值非以上三种回复，或者在指定时间内没有回复 PING 命令， 那么 Sentinel 认为服务器返回的回复无效（non-valid)。
小结
本文我们讲了主从模式的步骤，需要手动切换故障服务器的弊端，引出了 Sentinel 模式，可以实现监控和自动容灾，我们通过 Redis 提供的 Redis-Sentinel 来启动哨兵模式，当我们启动多个哨兵模式监视同一个主节点时，它们就会彼此发现形成一个新的高可用的 Sentinel 网络。同时我们讲了 Sentinel 的工作原理是通过 PING 命令来检查节点是否存活的，并通过配置项和复制偏移量 ID 来确定新主节点，下文我们讲一下哨兵管理命令和代码实战。
38 实战：Redis 哨兵模式（下）
上一篇我们介绍了 Redis Sentinel 的搭建和运行原理，本文我们重点来看下 Sentinel 的命令操作和代码实战。
Sentinel 命令操作
要使用 Sentinel 实现要连接到 Sentinel 服务器，和连接 Redis 服务相同，我们可以使用 redis-cli 来连接 Sentinel，如下命令所示：
1 2 3 [@iZ2ze0nc5n41zomzyqtksmZ:~]$ redis-cli -h 127.0.0.1 -p 26379 -a pwd654321 127.0.0.1:26379> ping PONG
其中：
-h 后面输入的是 Sentinel 的 IP；
-p 后面输入的是 Sentinel 的端口，默认是 26379；
-a 后面输入的是密码。
Sentinel 的端口号可以在 sentinel.conf 里面配置，通过 port 选项设置。
注意：Sentinel 可以监视多台主节点，而不是只能监视一台服务器 。想要监视多台主节点只需要在配置文件中设置多个 sentinel monitor master-name ip port quorum 即可，我们通过 master-name 来区分不同的主节点。
查询所有被监控的主服务器信息 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 127.0.0.1:26379> sentinel masters 1) 1) "name" 2) "mymaster" 3) "ip" 4) "127.0.0.1" 5) "port" 6) "6377" 7) "runid" 8) "eb3552c6fc8974f91466c4ada90fe23ef30fd89c" 9) "flags" 10) "master" 11) "link-pending-commands" 12) "0" 13) "link-refcount" 14) "1" 15) "last-ping-sent" 16) "0" 17) "last-ok-ping-reply" 18) "400" 19) "last-ping-reply" 20) "400" 21) "down-after-milliseconds" 22) "30000" 23) "info-refresh" 24) "5731" 25) "role-reported" 26) "master" 27) "role-reported-time" 28) "75963321" 29) "config-epoch" 30) "7" 31) "num-slaves" 32) "2" 33) "num-other-sentinels" 34) "1" 35) "quorum" 36) "2" 37) "failover-timeout" 38) "180000" 39) "parallel-syncs" 40) "1"
相关语法：
因为我们配置的 Sentinel 只监视了一台主服务器，所以只有一台服务器的信息。
查询某个主节点的信息 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 127.0.0.1:26379> sentinel master mymaster 1) "name" 2) "mymaster" 3) "ip" 4) "127.0.0.1" 5) "port" 6) "6377" 7) "runid" 8) "eb3552c6fc8974f91466c4ada90fe23ef30fd89c" 9) "flags" 10) "master" 11) "link-pending-commands" 12) "0" 13) "link-refcount" 14) "1" 15) "last-ping-sent" 16) "0" 17) "last-ok-ping-reply" 18) "250" 19) "last-ping-reply" 20) "250" 21) "down-after-milliseconds" 22) "30000" 23) "info-refresh" 24) "8191" 25) "role-reported" 26) "master" 27) "role-reported-time" 28) "76096303" 29) "config-epoch" 30) "7" 31) "num-slaves" 32) "2" 33) "num-other-sentinels" 34) "1" 35) "quorum" 36) "2" 37) "failover-timeout" 38) "180000" 39) "parallel-syncs" 40) "1"
相关语法：
1 sentinel master master-name
查看某个主节点的 IP 和端口 1 2 3 127.0.0.1:26379> sentinel get-master-addr-by-name mymaster 1) "127.0.0.1" 2) "6377"
相关语法：
1 sentinel get -master-addr-by -name master-name
查询从节点的信息 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 127.0.0.1:26379> sentinel slaves mymaster #获取方式一 1) 1) "name" 2) "127.0.0.1:6379" 3) "ip" 4) "127.0.0.1" 5) "port" 6) "6379" 7) "runid" 8) "14734d6065d745d89f115ca4735e7eeeeaa1a59b" 9) "flags" 10) "slave" 11) "link-pending-commands" 12) "0" 13) "link-refcount" 14) "1" 15) "last-ping-sent" 16) "0" 17) "last-ok-ping-reply" 18) "389" 19) "last-ping-reply" 20) "389" 21) "down-after-milliseconds" 22) "30000" 23) "info-refresh" 24) "390" 25) "role-reported" 26) "slave" 27) "role-reported-time" 28) "982798" 29) "master-link-down-time" 30) "1582192784000" 31) "master-link-status" 32) "err" 33) "master-host" 34) "127.0.0.1" 35) "master-port" 36) "6377" 37) "slave-priority" 38) "100" 39) "slave-repl-offset" 40) "1" 2) 1) "name" 2) "127.0.0.1:6378" 3) "ip" 4) "127.0.0.1" 5) "port" 6) "6378" 7) "runid" 8) "f9d69479ace6c9eb4a6dffa58ebc1ddf3de456e0" 9) "flags" 10) "slave" 11) "link-pending-commands" 12) "0" 13) "link-refcount" 14) "1" 15) "last-ping-sent" 16) "0" 17) "last-ok-ping-reply" 18) "390" 19) "last-ping-reply" 20) "390" 21) "down-after-milliseconds" 22) "30000" 23) "info-refresh" 24) "4004" 25) "role-reported" 26) "slave" 27) "role-reported-time" 28) "76212633" 29) "master-link-down-time" 30) "0" 31) "master-link-status" 32) "ok" 33) "master-host" 34) "127.0.0.1" 35) "master-port" 36) "6377" 37) "slave-priority" 38) "100" 39) "slave-repl-offset" 40) "10811245" 127.0.0.1:26379> sentinel replicas mymaster #获取方式二 1) 1) "name" 2) "127.0.0.1:6379" 3) "ip" 4) "127.0.0.1" 5) "port" 6) "6379" 7) "runid" 8) "14734d6065d745d89f115ca4735e7eeeeaa1a59b" 9) "flags" 10) "slave" 11) "link-pending-commands" 12) "0" 13) "link-refcount" 14) "1" 15) "last-ping-sent" 16) "0" 17) "last-ok-ping-reply" 18) "100" 19) "last-ping-reply" 20) "100" 21) "down-after-milliseconds" 22) "30000" 23) "info-refresh" 24) "100" 25) "role-reported" 26) "slave" 27) "role-reported-time" 28) "1071687" 29) "master-link-down-time" 30) "1582192873000" 31) "master-link-status" 32) "err" 33) "master-host" 34) "127.0.0.1" 35) "master-port" 36) "6377" 37) "slave-priority" 38) "100" 39) "slave-repl-offset" 40) "1" 2) 1) "name" 2) "127.0.0.1:6378" 3) "ip" 4) "127.0.0.1" 5) "port" 6) "6378" 7) "runid" 8) "f9d69479ace6c9eb4a6dffa58ebc1ddf3de456e0" 9) "flags" 10) "slave" 11) "link-pending-commands" 12) "0" 13) "link-refcount" 14) "1" 15) "last-ping-sent" 16) "0" 17) "last-ok-ping-reply" 18) "100" 19) "last-ping-reply" 20) "100" 21) "down-after-milliseconds" 22) "30000" 23) "info-refresh" 24) "2496" 25) "role-reported" 26) "slave" 27) "role-reported-time" 28) "76301522" 29) "master-link-down-time" 30) "0" 31) "master-link-status" 32) "ok" 33) "master-host" 34) "127.0.0.1" 35) "master-port" 36) "6377" 37) "slave-priority" 38) "100" 39) "slave-repl-offset" 40) "10823208"
相关语法：
1 sentinel replicas mymaster
或
1 sentinel slaves master-name
查询 Sentinel 集群中的其他 Sentinel 信息 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 127.0.0.1:26379> sentinel sentinels mymaster 1) 1) "name" 2) "6455f2f74614a71ce0a63398b2e48d6cd1cf0d06" 3) "ip" 4) "127.0.0.1" 5) "port" 6) "26377" 7) "runid" 8) "6455f2f74614a71ce0a63398b2e48d6cd1cf0d06" 9) "flags" 10) "sentinel" 11) "link-pending-commands" 12) "0" 13) "link-refcount" 14) "1" 15) "last-ping-sent" 16) "0" 17) "last-ok-ping-reply" 18) "571" 19) "last-ping-reply" 20) "571" 21) "down-after-milliseconds" 22) "30000" 23) "last-hello-message" 24) "1043" 25) "voted-leader" 26) "?" 27) "voted-leader-epoch" 28) "0"
相关语法：
1 sentinel sentinels master-name
检查可用 Sentinel 的数量 1 2 127.0.0.1:26379> sentinel ckquorum mymaster OK 2 usable Sentinels. Quorum and failover authorization can be reached
有两个可用的 Sentinel，可用完成仲裁和故障转移授权。
相关语法：
1 sentinel ckquorum master-name
强制故障转移 1 2 127.0.0.1:26379> sentinel failover mymaster OK
相关语法：
1 sentinel failover master-name
在线修改配置信息
在 Redis 2.8.4 之前如果需要修改 Sentinel 的配置文件，例如添加或删除一个监视主节点，需要先停止 Sentinel 服务，再找到配置文件修改之后，重新启动 Sentinel 才行，这样就给我们带来了很多的不便，尤其是生产环境的 Sentinel，正常情况下如果是非致命问题我们是不能手动停止服务的，幸运的是 Redis 2.8.4 之后，我们可以不停机在线修改配置文件了，修改命令有以下几个。
增加监视主节点 使用 sentinel monitor mymaster IP Port Quorum 命令来添加监视主节点，如下命令所示：
1 2 127.0.0.1:26379> sentinel monitor mymaster 127.0.0.1 6379 2 OK
OK 表示添加监视主节点成功。
移除主节点的监视 使用 sentinel remove master-name 命令来实现移除主节点的监视，如下命令所示：
1 2 127.0.0.1:26379> sentinel remove mymaster OK
OK 表示操作成功。
修改 quorum 参数 使用 sentinel set master-name quorum n 来修改 quorum 参数，如下命令所示：
1 2 127.0.0.1:26379> sentinel set mymaster quorum 1 OK
quorum 参数用来表示确认主节点下线的 Sentinel 数量，如果 quorum 设置为 1 表示只要有一台 Sentinel 确认主观下线后，这个主节点就客观（真正地）下线了。
小贴士：以上所有对配置文件的修改，都会自动被刷新到物理配置文件 sentinel.conf 中。
代码实战
本文我们通过 Java 代码来实现，通过 Sentinel 连接信息获取相关 Redis 客户端，再进行相关 Redis 操作，这样 Sentinel 就会帮我们做容灾恢复，我们就不用担心操作某一个 Redis 服务器端，因为服务器挂了之后就会导致程序不可用了，具体实现代码如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 import redis.clients.jedis.Jedis;import redis.clients.jedis.JedisSentinelPool;import utils.Config;import java.util.HashSet;import java.util.Set;public class SentinelExample { private static String _MASTER_NAME = "mymaster" ; public static void main (String[] args) { Set<String> set = new HashSet <>(); set.add("127.0.0.1:26379" ); JedisSentinelPool jedisSentinel = new JedisSentinelPool (_MASTER_NAME, set, Config.REDIS_AUTH); Jedis jedis = jedisSentinel.getResource(); String setRes = jedis.set("key" , "Hello, redis." ); System.out.println(setRes); System.out.println(jedis.get("key" )); } }
以上程序执行结果如下：
小结
本文我们讲了 Sentinel 相关的命令操作，主要是用于查询相关主从节点和其他 Sentinel 信息的，还可以执行强制故障转移等，我们还讲了 2.8.4 提供的在线修改 Sentinel 参数的三个方法，方便我们更好的使用 Sentinel，最后用代码实现了通过 Sentinel 获取主节点并进行 Redis 服务器操作的实例，这样就讲完整个 Sentinel 的介绍和应用。
39 实战：Redis 集群模式（上）
Redis Cluster 是 Redis 3.0 版本推出的 Redis 集群方案，它将数据分布在不同的服务区上，以此来降低系统对单主节点的依赖，并且可以大大的提高 Redis 服务的读写性能。
Redis 将所有的数据分为 16384 个 slots（槽），每个节点负责其中的一部分槽位，当有 Redis 客户端连接集群时，会得到一份集群的槽位配置信息，这样它就可以直接把请求命令发送给对应的节点进行处理。
Redis Cluster 是无代理模式去中心化的运行模式，客户端发送的绝大数命令会直接交给相关节点执行，这样大部分情况请求命令无需转发，或仅转发一次的情况下就能完成请求与响应，所以集群单个节点的性能与单机 Redis 服务器的性能是非常接近的，因此在理论情况下，当水平扩展一倍的主节点就相当于请求处理的性能也提高了一倍，所以 Redis Cluster 的性能是非常高的。
Redis Cluster 架构图如下所示：
搭建 Redis Cluster
Redis Cluster 的搭建方式有两种，一种是使用 Redis 源码中提供的 create-cluster 工具快速的搭建 Redis 集群环境，另一种是配置文件的方式手动创建 Redis 集群环境。
快速搭建 Redis Cluster create-cluster 工具在 utils/create-cluster 目录下，如下图所示：
使用命令 ./create-cluster start 就可以急速创建一个 Redis 集群，执行如下：
1 2 3 4 5 6 7 $ ./create-cluster start Starting 30001 Starting 30002 Starting 30003 Starting 30004 Starting 30005 Starting 30006
接下来我们需要把以上创建的 6 个节点通过 create 命令组成一个集群，执行如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 [@iZ2ze0nc5n41zomzyqtksmZ:create-cluster]$ ./create-cluster create # 组建集群 > >> Performing hash slots allocation on 6 nodes... Master[0] -> Slots 0 - 5460 Master[1] -> Slots 5461 - 10922 Master[2] -> Slots 10923 - 16383 Adding replica 127.0.0.1:30005 to 127.0.0.1:30001 Adding replica 127.0.0.1:30006 to 127.0.0.1:30002 Adding replica 127.0.0.1:30004 to 127.0.0.1:30003 > >> Trying to optimize slaves allocation for anti-affinity [WARNING] Some slaves are in the same host as their master M: 445f2a86fe36d397613839d8cc1ae6702c976593 127.0.0.1:30001 slots:[0-5460] (5461 slots) master M: 63bb14023c0bf58926738cbf857ea304bff8eb50 127.0.0.1:30002 slots:[5461-10922] (5462 slots) master M: 864d4dfe32e3e0b81a64cec8b393bbd26a65cbcc 127.0.0.1:30003 slots:[10923-16383] (5461 slots) master S: 64828ab44566fc5ad656e831fd33de87be1387a0 127.0.0.1:30004 replicates 445f2a86fe36d397613839d8cc1ae6702c976593 S: 0b17b00542706343583aa73149ec5ff63419f140 127.0.0.1:30005 replicates 63bb14023c0bf58926738cbf857ea304bff8eb50 S: e35f06ca9b700073472d72001a39ea4dfcb541cd 127.0.0.1:30006 replicates 864d4dfe32e3e0b81a64cec8b393bbd26a65cbcc Can I set the above configuration? (type 'yes' to accept): yes > >> Nodes configuration updated > >> Assign a different config epoch to each node > >> Sending CLUSTER MEET messages to join the cluster Waiting for the cluster to join . > >> Performing Cluster Check (using node 127.0.0.1:30001) M: 445f2a86fe36d397613839d8cc1ae6702c976593 127.0.0.1:30001 slots:[0-5460] (5461 slots) master 1 additional replica(s) M: 864d4dfe32e3e0b81a64cec8b393bbd26a65cbcc 127.0.0.1:30003 slots:[10923-16383] (5461 slots) master 1 additional replica(s) S: e35f06ca9b700073472d72001a39ea4dfcb541cd 127.0.0.1:30006 slots: (0 slots) slave replicates 864d4dfe32e3e0b81a64cec8b393bbd26a65cbcc S: 0b17b00542706343583aa73149ec5ff63419f140 127.0.0.1:30005 slots: (0 slots) slave replicates 63bb14023c0bf58926738cbf857ea304bff8eb50 M: 63bb14023c0bf58926738cbf857ea304bff8eb50 127.0.0.1:30002 slots:[5461-10922] (5462 slots) master 1 additional replica(s) S: 64828ab44566fc5ad656e831fd33de87be1387a0 127.0.0.1:30004 slots: (0 slots) slave replicates 445f2a86fe36d397613839d8cc1ae6702c976593 [OK] All nodes agree about slots configuration. > >> Check for open slots... > >> Check slots coverage... [OK] All 16384 slots covered.
在执行的过程中会询问你是否通过把 30001、30002、30003 作为主节点，把 30004、30005、30006 作为它们的从节点，输入 yes 后会执行完成。
我们可以先使用 redis-cli 连接到集群，命令如下：
在使用 nodes 命令来查看集群的节点信息，命令如下：
1 2 3 4 5 6 7 127.0.0.1:30001> cluster nodes 864d4dfe32e3e0b81a64cec8b393bbd26a65cbcc 127.0.0.1:30003@40003 master - 0 1585125835078 3 connected 10923-16383 e35f06ca9b700073472d72001a39ea4dfcb541cd 127.0.0.1:30006@40006 slave 864d4dfe32e3e0b81a64cec8b393bbd26a65cbcc 0 1585125835078 6 connected 0b17b00542706343583aa73149ec5ff63419f140 127.0.0.1:30005@40005 slave 63bb14023c0bf58926738cbf857ea304bff8eb50 0 1585125835078 5 connected 63bb14023c0bf58926738cbf857ea304bff8eb50 127.0.0.1:30002@40002 master - 0 1585125834175 2 connected 5461-10922 445f2a86fe36d397613839d8cc1ae6702c976593 127.0.0.1:30001@40001 myself,master - 0 1585125835000 1 connected 0-5460 64828ab44566fc5ad656e831fd33de87be1387a0 127.0.0.1:30004@40004 slave 445f2a86fe36d397613839d8cc1ae6702c976593 0 1585125835000 4 connected
可以看出 30001、30002、30003 都为主节点，30001 对应的槽位是 0~5460，30002 对应的槽位是 5461~10922，30003 对应的槽位是 10923~16383，总共有槽位 16384 个（0~16383）。
create-cluster 搭建的方式虽然速度很快，但是该方式搭建的集群主从节点数量固定以及槽位分配模式固定，并且安装在同一台服务器上，所以只能用于测试环境。
我们测试完成之后，可以使用以下命令，关闭并清理集群 ：
1 2 3 4 5 6 7 8 $ ./create-cluster stop Stopping 30001 Stopping 30002 Stopping 30003 Stopping 30004 Stopping 30005 Stopping 30006 $ ./create-cluster clean
手动搭建 Redis Cluster 由于 create-cluster 本身的限制，在实际生产环境中我们需要使用手动添加配置的方式搭建 Redis 集群，为此我们先要把 Redis 安装包复制到 node1 到 node6 文件中，因为我们要安装 6 个节点，3 主 3 从
接下来我们进行配置并启动 Redis 集群。
1. 设置配置文件
我们需要修改每个节点内的 redis.conf 文件，设置 cluster-enabled yes 表示开启集群模式，并且修改各自的端口，我们继续使用 30001 到 30006，通过 port 3000X 设置。
2. 启动各个节点
redis.conf 配置好之后，我们就可以启动所有的节点了，命令如下：
1 2 cd /usr/local/soft/mycluster/node1 ./src/redis-server redis.conf
3. 创建集群并分配槽位
之前我们已经启动了 6 个节点，但这些节点都在各自的集群之内并未互联互通，因此接下来我们需要把这些节点串连成一个集群，并为它们指定对应的槽位，执行命令如下：
1 redis-cli --cluster create 127.0.0.1:30001 127.0.0.1:30002 127.0.0.1:30003 127.0.0.1:30004 127.0.0.1:30005 127.0.0.1:30006 --cluster-replicas 1
其中 create 后面跟多个节点，表示把这些节点作为整个集群的节点，而 cluster-replicas 表示给集群中的主节点指定从节点的数量，1 表示为每个主节点设置一个从节点。
在执行了 create 命令之后，系统会为我们指定节点的角色和槽位分配计划，如下所示：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 > >> Performing hash slots allocation on 6 nodes... Master[0] -> Slots 0 - 5460 Master[1] -> Slots 5461 - 10922 Master[2] -> Slots 10923 - 16383 Adding replica 127.0.0.1:30005 to 127.0.0.1:30001 Adding replica 127.0.0.1:30006 to 127.0.0.1:30002 Adding replica 127.0.0.1:30004 to 127.0.0.1:30003 > >> Trying to optimize slaves allocation for anti-affinity [WARNING] Some slaves are in the same host as their master M: bdd1c913f87eacbdfeabc71befd0d06c913c891c 127.0.0.1:30001 slots:[0-5460] (5461 slots) master M: bdd1c913f87eacbdfeabc71befd0d06c913c891c 127.0.0.1:30002 slots:[5461-10922] (5462 slots) master M: bdd1c913f87eacbdfeabc71befd0d06c913c891c 127.0.0.1:30003 slots:[10923-16383] (5461 slots) master S: bdd1c913f87eacbdfeabc71befd0d06c913c891c 127.0.0.1:30004 replicates bdd1c913f87eacbdfeabc71befd0d06c913c891c S: bdd1c913f87eacbdfeabc71befd0d06c913c891c 127.0.0.1:30005 replicates bdd1c913f87eacbdfeabc71befd0d06c913c891c S: bdd1c913f87eacbdfeabc71befd0d06c913c891c 127.0.0.1:30006 replicates bdd1c913f87eacbdfeabc71befd0d06c913c891c Can I set the above configuration? (type 'yes' to accept):
从以上信息可以看出，Redis 打算把 30001、30002、30003 设置为主节点，并为他们分配的槽位，30001 对应的槽位是 0~5460，30002 对应的槽位是 5461~10922，30003 对应的槽位是 10923~16383，并且把 30005 设置为 30001 的从节点、30006 设置为 30002 的从节点、30004 设置为 30003 的从节点，我们只需要输入 yes 即可确认并执行分配，如下所示：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 Can I set the above configuration? (type 'yes' to accept): yes > >> Nodes configuration updated > >> Assign a different config epoch to each node > >> Sending CLUSTER MEET messages to join the cluster Waiting for the cluster to join .... > >> Performing Cluster Check (using node 127.0.0.1:30001) M: 887397e6fefe8ad19ea7569e99f5eb8a803e3785 127.0.0.1:30001 slots:[0-5460] (5461 slots) master 1 additional replica(s) S: abec9f98f9c01208ba77346959bc35e8e274b6a3 127.0.0.1:30005 slots: (0 slots) slave replicates 887397e6fefe8ad19ea7569e99f5eb8a803e3785 S: 1a324d828430f61be6eaca7eb2a90728dd5049de 127.0.0.1:30004 slots: (0 slots) slave replicates f5958382af41d4e1f5b0217c1413fe19f390b55f S: dc0702625743c48c75ea935c87813c4060547cef 127.0.0.1:30006 slots: (0 slots) slave replicates 3da35c40c43b457a113b539259f17e7ed616d13d M: 3da35c40c43b457a113b539259f17e7ed616d13d 127.0.0.1:30002 slots:[5461-10922] (5462 slots) master 1 additional replica(s) M: f5958382af41d4e1f5b0217c1413fe19f390b55f 127.0.0.1:30003 slots:[10923-16383] (5461 slots) master 1 additional replica(s) [OK] All nodes agree about slots configuration. > >> Check for open slots... > >> Check slots coverage... [OK] All 16384 slots covered.
显示 OK 表示整个集群就已经成功启动了。
接下来，我们使用 redis-cli 连接并测试一下集群的运行状态，代码如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 $ redis-cli -c -p 30001 127.0.0.1:30001> cluster info # 查看集群信息 cluster_state:ok # 状态正常 cluster_slots_assigned:16384 # 槽位数 cluster_slots_ok:16384 # 正常的槽位数 cluster_slots_pfail:0 cluster_slots_fail:0 cluster_known_nodes:6 # 集群的节点数 cluster_size:3 # 集群主节点数 cluster_current_epoch:6 cluster_my_epoch:1 cluster_stats_messages_ping_sent:130 cluster_stats_messages_pong_sent:127 cluster_stats_messages_sent:257 cluster_stats_messages_ping_received:122 cluster_stats_messages_pong_received:130 cluster_stats_messages_meet_received:5 cluster_stats_messages_received:257
相关字段的说明已经标识在上述的代码中了，这里就不再赘述。
动态增删节点
某些情况下，我们需要根据实际的业务情况，对已经在运行的集群进行动态的添加或删除节点，那我们就需要进行以下操作。
增加主节点 添加方式一：cluster meet
使用 cluster meet ip:port 命令就可以把一个节点加入到集群中，执行命令如下：
1 2 3 4 5 6 7 8 9 10 127.0.0.1:30001> cluster meet 127.0.0.1 30007 OK 127.0.0.1:30001> cluster nodes dc0702625743c48c75ea935c87813c4060547cef 127.0.0.1:30006@40006 slave 3da35c40c43b457a113b539259f17e7ed616d13d 0 1585142916000 6 connected df0190853a53d8e078205d0e2fa56046f20362a7 127.0.0.1:30007@40007 master - 0 1585142917740 0 connected f5958382af41d4e1f5b0217c1413fe19f390b55f 127.0.0.1:30003@40003 master - 0 1585142916738 3 connected 10923-16383 3da35c40c43b457a113b539259f17e7ed616d13d 127.0.0.1:30002@40002 master - 0 1585142913000 2 connected 5461-10922 abec9f98f9c01208ba77346959bc35e8e274b6a3 127.0.0.1:30005@40005 slave 887397e6fefe8ad19ea7569e99f5eb8a803e3785 0 1585142917000 5 connected 887397e6fefe8ad19ea7569e99f5eb8a803e3785 127.0.0.1:30001@40001 myself,master - 0 1585142915000 1 connected 0-5460 1a324d828430f61be6eaca7eb2a90728dd5049de 127.0.0.1:30004@40004 slave f5958382af41d4e1f5b0217c1413fe19f390b55f 0 1585142916000 4 connected
可以看出端口为 30007 的节点并加入到集群中，并设置成了主节点。
添加方式二：add-node
使用 redis-cli --cluster add-node 添加节点ip:port 集群某节点ip:port 也可以把一个节点添加到集群中，执行命令如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 $ redis-cli --cluster add-node 127.0.0.1:30008 127.0.0.1:30001 > >> Adding node 127.0.0.1:30008 to cluster 127.0.0.1:30001 > >> Performing Cluster Check (using node 127.0.0.1:30001) M: 887397e6fefe8ad19ea7569e99f5eb8a803e3785 127.0.0.1:30001 slots:[0-5460] (5461 slots) master 1 additional replica(s) S: dc0702625743c48c75ea935c87813c4060547cef 127.0.0.1:30006 slots: (0 slots) slave replicates 3da35c40c43b457a113b539259f17e7ed616d13d M: df0190853a53d8e078205d0e2fa56046f20362a7 127.0.0.1:30007 slots: (0 slots) master M: f5958382af41d4e1f5b0217c1413fe19f390b55f 127.0.0.1:30003 slots:[10923-16383] (5461 slots) master 1 additional replica(s) M: 1d09d26fd755298709efe60278457eaa09cefc26 127.0.0.1:30008 slots: (0 slots) master M: 3da35c40c43b457a113b539259f17e7ed616d13d 127.0.0.1:30002 slots:[5461-10922] (5462 slots) master 1 additional replica(s) S: abec9f98f9c01208ba77346959bc35e8e274b6a3 127.0.0.1:30005 slots: (0 slots) slave replicates 887397e6fefe8ad19ea7569e99f5eb8a803e3785 S: 1a324d828430f61be6eaca7eb2a90728dd5049de 127.0.0.1:30004 slots: (0 slots) slave replicates f5958382af41d4e1f5b0217c1413fe19f390b55f [OK] All nodes agree about slots configuration. > >> Check for open slots... > >> Check slots coverage... [OK] All 16384 slots covered. [ERR] Node 127.0.0.1:30008 is not empty. Either the node already knows other nodes (check with CLUSTER NODES) or contains some key in database 0.
从以上结果可以看出 30008 节点也被设置成了主节点。
添加从节点 使用 cluster replicate nodeId 命令就可以把当前节点设置为目标节点的从节点，执行命令如下：
1 2 3 4 5 6 7 8 9 10 11 127.0.0.1:30008> cluster replicate df0190853a53d8e078205d0e2fa56046f20362a7 OK 127.0.0.1:30008> cluster nodes df0190853a53d8e078205d0e2fa56046f20362a7 127.0.0.1:30007@40007 master - 0 1585147827000 0 connected abec9f98f9c01208ba77346959bc35e8e274b6a3 127.0.0.1:30005@40005 slave 887397e6fefe8ad19ea7569e99f5eb8a803e3785 0 1585147827000 1 connected 1a324d828430f61be6eaca7eb2a90728dd5049de 127.0.0.1:30004@40004 slave f5958382af41d4e1f5b0217c1413fe19f390b55f 0 1585147823000 3 connected 887397e6fefe8ad19ea7569e99f5eb8a803e3785 127.0.0.1:30001@40001 master - 0 1585147826000 1 connected 0-5460 dc0702625743c48c75ea935c87813c4060547cef 127.0.0.1:30006@40006 slave 3da35c40c43b457a113b539259f17e7ed616d13d 0 1585147826930 2 connected f5958382af41d4e1f5b0217c1413fe19f390b55f 127.0.0.1:30003@40003 master - 0 1585147826000 3 connected 10923-16383 1d09d26fd755298709efe60278457eaa09cefc26 127.0.0.1:30008@40008 myself,slave df0190853a53d8e078205d0e2fa56046f20362a7 0 1585147823000 7 connected 3da35c40c43b457a113b539259f17e7ed616d13d 127.0.0.1:30002@40002 master - 0 1585147827933 2 connected 5461-10922
可以看出 30008 已经变为 30007 的从节点了。
删除节点 使用 cluster forget nodeId 命令就可以把一个节点从集群中移除。
此命令和 meet 命令不同的时，删除节点需要把使用节点的 Id 进行删除，可以通过 cluster nodes 命令查看所有节点的 Id 信息，其中每一行的最前面的 40 位字母和数组的组合就是该节点的 Id，如下图所示：
执行命令如下：
1 2 127.0.0.1:30001> cluster forget df0190853a53d8e078205d0e2fa56046f20362a7 OK
此时我们使用 cluster nodes 命令查看集群的所有节点信息：
1 2 3 4 5 6 7 127.0.0.1:30001> cluster nodes dc0702625743c48c75ea935c87813c4060547cef 127.0.0.1:30006@40006 slave 3da35c40c43b457a113b539259f17e7ed616d13d 0 1585143789940 6 connected f5958382af41d4e1f5b0217c1413fe19f390b55f 127.0.0.1:30003@40003 master - 0 1585143791000 3 connected 10923-16383 3da35c40c43b457a113b539259f17e7ed616d13d 127.0.0.1:30002@40002 master - 0 1585143789000 2 connected 5461-10922 abec9f98f9c01208ba77346959bc35e8e274b6a3 127.0.0.1:30005@40005 slave 887397e6fefe8ad19ea7569e99f5eb8a803e3785 0 1585143789000 5 connected 887397e6fefe8ad19ea7569e99f5eb8a803e3785 127.0.0.1:30001@40001 myself,master - 0 1585143786000 1 connected 0-5460 1a324d828430f61be6eaca7eb2a90728dd5049de 127.0.0.1:30004@40004 slave f5958382af41d4e1f5b0217c1413fe19f390b55f 0 1585143791945 4 connected
可以看出之前的端口为 30007 的节点已经被我们成功的移除了。
小结
本文讲了 Redis 集群的两种搭建方式：create-cluster start 和 cluster create，前一种方式虽然速度比较快，但它只能创建数量固定的主从节点，并且所有节点都在同一台服务器上，因此只能用于测试环境。我们还讲了 Redis 集群动态添加主、从节点和删除任意节点的功能。
40 实战：Redis 集群模式（下）
上篇文章我们讲了 Redis 集群的搭建与节点的动态添加和删除，我们这里再来简单的复习一下，其中 30001~30006 是我们最初搭建的集群，而 30007 和 30008 是后面动态添加的主从节点，我们使用 --cluster info 命令来看一下主节点和槽位的分配情况，执行代码如下：
1 2 3 4 5 6 7 $ redis-cli --cluster info 127.0.0.1:30001 127.0.0.1:30001 (887397e6...) -> 0 keys | 5461 slots | 1 slaves. 127.0.0.1:30007 (df019085...) -> 0 keys | 0 slots | 1 slaves. 127.0.0.1:30003 (f5958382...) -> 0 keys | 5461 slots | 1 slaves. 127.0.0.1:30002 (3da35c40...) -> 0 keys | 5462 slots | 1 slaves. [OK] 0 keys in 4 masters. 0.00 keys per slot on average.
可以看出动态添加的主节点 30007 有一个从节点，但并没有分配任何槽位，这显然是不能满足我们的需求的，只添加了节点，但不处理任何数据，所以我们需要重新分片，让数据存储在所有的主节点上，这样才能发挥集群的最大作用。
重新分片
我们可以使用 reshard 命令，对槽位（slots）进行重新分配，执行命令如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 $ redis-cli --cluster reshard 127.0.0.1:30007 > >> Performing Cluster Check (using node 127.0.0.1:30007) M: df0190853a53d8e078205d0e2fa56046f20362a7 127.0.0.1:30007 slots:[0-1332],[5461-6794],[10923-12255] (4000 slots) master 1 additional replica(s) S: dc0702625743c48c75ea935c87813c4060547cef 127.0.0.1:30006 slots: (0 slots) slave replicates 3da35c40c43b457a113b539259f17e7ed616d13d M: 3da35c40c43b457a113b539259f17e7ed616d13d 127.0.0.1:30002 slots:[6795-10922] (4128 slots) master 1 additional replica(s) S: 1a324d828430f61be6eaca7eb2a90728dd5049de 127.0.0.1:30004 slots: (0 slots) slave replicates f5958382af41d4e1f5b0217c1413fe19f390b55f S: 1d09d26fd755298709efe60278457eaa09cefc26 127.0.0.1:30008 slots: (0 slots) slave replicates df0190853a53d8e078205d0e2fa56046f20362a7 S: abec9f98f9c01208ba77346959bc35e8e274b6a3 127.0.0.1:30005 slots: (0 slots) slave replicates 887397e6fefe8ad19ea7569e99f5eb8a803e3785 M: f5958382af41d4e1f5b0217c1413fe19f390b55f 127.0.0.1:30003 slots:[12256-16383] (4128 slots) master 1 additional replica(s) M: 887397e6fefe8ad19ea7569e99f5eb8a803e3785 127.0.0.1:30001 slots:[1333-5460] (4128 slots) master 1 additional replica(s) [OK] All nodes agree about slots configuration. > >> Check for open slots... > >> Check slots coverage... [OK] All 16384 slots covered. How many slots do you want to move (from 1 to 16384)?
在执行的过程中，它会询问你打算移动多少个节点，取值范围是 1 到 16384，我们这里输入 4000，意思是移动 4000 个槽位到某个主节点，输入命令之后，执行效果如下：
1 2 How many slots do you want to move (from 1 to 16384)? 4000 What is the receiving node ID?
接着它会询问你需要把这些槽位分配到哪个节点上，请输入节点 Id，我们把上面 30007 端口的 Id 输入进去之后，敲击回车，执行效果如下：
1 2 3 4 5 6 How many slots do you want to move (from 1 to 16384)? 4000 What is the receiving node ID? df0190853a53d8e078205d0e2fa56046f20362a7 Please enter all the source node IDs. Type 'all' to use all the nodes as source nodes for the hash slots. Type 'done' once you entered all the source nodes IDs. Source node #1:
此时它会询问你要从那个源节点中进行转移，输入 all 命令表示从所有节点中随机抽取，执行效果如下：
1 2 3 4 5 6 7 8 9 10 11 12 # ......忽略其他 Moving slot 2656 from 887397e6fefe8ad19ea7569e99f5eb8a803e3785 Moving slot 2657 from 887397e6fefe8ad19ea7569e99f5eb8a803e3785 Moving slot 2658 from 887397e6fefe8ad19ea7569e99f5eb8a803e3785 Moving slot 2659 from 887397e6fefe8ad19ea7569e99f5eb8a803e3785 Moving slot 2660 from 887397e6fefe8ad19ea7569e99f5eb8a803e3785 Moving slot 2661 from 887397e6fefe8ad19ea7569e99f5eb8a803e3785 Moving slot 2662 from 887397e6fefe8ad19ea7569e99f5eb8a803e3785 Moving slot 2663 from 887397e6fefe8ad19ea7569e99f5eb8a803e3785 Moving slot 2664 from 887397e6fefe8ad19ea7569e99f5eb8a803e3785 Moving slot 2665 from 887397e6fefe8ad19ea7569e99f5eb8a803e3785 Do you want to proceed with the proposed reshard plan (yes/no)?
此时它会把所有要转移的节点信息列举出来，让你确认，你只需要输入 yes 就开始执行转移操作了。
在执行完转移之后，我们使用 cluster slots 命令来查看一下槽位的相关信息，结果如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 $ redis-cli -c -p 30001 127.0.0.1:30001> cluster slots # 查看集群槽位信息 1) 1) (integer) 0 2) (integer) 1332 3) 1) "127.0.0.1" 2) (integer) 30007 3) "df0190853a53d8e078205d0e2fa56046f20362a7" 4) 1) "127.0.0.1" 2) (integer) 30008 3) "1d09d26fd755298709efe60278457eaa09cefc26" 2) 1) (integer) 5461 2) (integer) 6794 3) 1) "127.0.0.1" 2) (integer) 30007 3) "df0190853a53d8e078205d0e2fa56046f20362a7" 4) 1) "127.0.0.1" 2) (integer) 30008 3) "1d09d26fd755298709efe60278457eaa09cefc26" 3) 1) (integer) 10923 2) (integer) 12255 3) 1) "127.0.0.1" 2) (integer) 30007 3) "df0190853a53d8e078205d0e2fa56046f20362a7" 4) 1) "127.0.0.1" 2) (integer) 30008 3) "1d09d26fd755298709efe60278457eaa09cefc26" 4) 1) (integer) 12256 2) (integer) 16383 3) 1) "127.0.0.1" 2) (integer) 30003 3) "f5958382af41d4e1f5b0217c1413fe19f390b55f" 4) 1) "127.0.0.1" 2) (integer) 30004 3) "1a324d828430f61be6eaca7eb2a90728dd5049de" 5) 1) (integer) 6795 2) (integer) 10922 3) 1) "127.0.0.1" 2) (integer) 30002 3) "3da35c40c43b457a113b539259f17e7ed616d13d" 4) 1) "127.0.0.1" 2) (integer) 30006 3) "dc0702625743c48c75ea935c87813c4060547cef" 6) 1) (integer) 1333 2) (integer) 5460 3) 1) "127.0.0.1" 2) (integer) 30001 3) "887397e6fefe8ad19ea7569e99f5eb8a803e3785" 4) 1) "127.0.0.1" 2) (integer) 30005 3) "abec9f98f9c01208ba77346959bc35e8e274b6a3"
从结果可以看出 30007 分别从其他三个主节点中抽取了一部分槽位，作为了自己的槽位。
注意，执行此过程中如果出现 /usr/bin/env: ruby: No such file or directory 错误，表明工具在执行的时候需要依赖 Ruby 环境，可使用命令 yum install ruby 安装 Ruby 环境即可。
槽位定位算法
Redis 集群总共的槽位数是 16384 个，每一个主节点负责维护一部分槽以及槽所映射的键值数据，Redis 集群默认会对要存储的 key 值使用 CRC16 算法进行 hash 得到一个整数值，然后用这个整数值对 16384 进行取模来得到具体槽位，公式为：
slot = CRC16(key) % 16383
负载均衡
在 Redis 集群负载不均衡的情况下，我们可以使用 rebalance 命令重新分配各个节点负责的槽数量，从而使得各个节点的负载压力趋于平衡，从而提高 Redis 集群的整体运行效率。
rebalance 命令如下：
1 $ redis-cli --cluster rebalance 127.0.0.1:30007
需要注意的是，即使输入 rebalance 命令，但它可能不会执行，当它认为没有必要进行分配时会直接退出，如下所示：
1 2 3 4 5 6 7 $ redis-cli --cluster rebalance 127.0.0.1:30007 > >> Performing Cluster Check (using node 127.0.0.1:30007) [OK] All nodes agree about slots configuration. > >> Check for open slots... > >> Check slots coverage... [OK] All 16384 slots covered. *** No rebalancing needed! All nodes are within the 2.00% threshold.
代码实战
前面我们讲了 Redis 集群搭建的相关功能，接下来我们使用 Java 代码来操作一下 Redis 集群，本文依然使用 Jedis 来作为客户端进行相关的操作，核心代码如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 import redis.clients.jedis.HostAndPort;import redis.clients.jedis.JedisCluster;import java.util.HashSet;import java.util.Set;public class ClusterExample { public static void main (String[] args) { Set<HostAndPort> nodes = new HashSet <>(); nodes.add(new HostAndPort ("127.0.0.1" , 30001 )); nodes.add(new HostAndPort ("127.0.0.1" , 30002 )); nodes.add(new HostAndPort ("127.0.0.1" , 30003 )); nodes.add(new HostAndPort ("127.0.0.1" , 30004 )); nodes.add(new HostAndPort ("127.0.0.1" , 30005 )); nodes.add(new HostAndPort ("127.0.0.1" , 30006 )); JedisCluster jedisCluster = new JedisCluster (nodes, 10000 , 10 ); String setResult = jedisCluster.set("lang" , "redis" ); System.out.println("添加：" + setResult); String getResult = jedisCluster.get("lang" ); System.out.println("查询：" + getResult); } }
以上程序的执行结果如下：
此结果表明 Redis 集群操作正常，除了使用的操作对象不同之外，操作的方法名称都是相同的，所以对程序员来说比较友好，你可以根据自己的业务场景去写相应的代码了。
故障
在文章的最后部分，我们来看一下 Redis 集群故障相关的知识点，这样在我们遇到一些故障问题时就不会那么慌张了，并且能为我们处理故障时提供一些帮助。
故障发现 故障发现里面有两个重要的概念：疑似下线（PFAIL-Possibly Fail）和确定下线（Fail）。
集群中的健康监测是通过定期向集群中的其他节点发送 PING 信息来确认的，如果发送 PING 消息的节点在规定时间内，没有收到返回的 PONG 消息，那么对方节点就会被标记为疑似下线。
一个节点发现某个节点疑似下线，它会将这条信息向整个集群广播，其它节点就会收到这个消息，并且通过 PING 的方式监测某节点是否真的下线了。如果一个节点收到某个节点疑似下线的数量超过集群数量的一半以上，就可以标记该节点为确定下线状态，然后向整个集群广播，强迫其它节点也接收该节点已经下线的事实，并立即对该失联节点进行主从切换。
这就是疑似下线和确认下线的概念，这个概念和哨兵模式里面的主观下线和客观下线的概念比较类似。
故障转移 当一个节点被集群标识为确认下线之后就可以执行故障转移了，故障转移的执行流程如下：
从下线的主节点的所有从节点中，选择一个从节点（选择的方法详见下面“新主节点选举原则”部分）；
从节点会执行 SLAVEOF NO ONE 命令，关闭这个从节点的复制功能，并从从节点转变回主节点，原来同步所得的数据集不会被丢弃；
新的主节点会撤销所有对已下线主节点的槽指派，并将这些槽全部指派给自己；
新的主节点向集群广播一条 PONG 消息，这条 PONG 消息是让集群中的其他节点知道此节点已经由从节点变成了主节点，并且这个主节点已经接管了原本由已下线节点负责处理的槽位信息；
新的主节点开始处理相关的命令请求，此故障转移过程完成。
新主节点选举原则 新主节点选举的方法是这样的：
集群的纪元（epoch）是一个自增计数器，初始值为0；
而每个主节点都有一次投票的机会，主节点会把这一票投给第一个要求投票的从节点；
当从节点发现自己正在复制的主节点确认下线之后，就会向集群广播一条消息，要求所有有投票权的主节点给此从节点投票；
如果有投票权的主节点还没有给其他人投票的情况下，它会向第一个要求投票的从节点发送一条消息，表示把这一票投给这个从节点；
当从节点收到投票数量大于集群数量的半数以上时，这个从节点就会当选为新的主节点。
到这里整个新主节点的选择就完成了。
小结
本文从动态新增的主节点通过 reshard 命令重新分配槽位开始，讲了槽位定位的算法以及负载均衡的实现方法，还使用代码的方式演示了如何在程序中操作 Redis 集群，最后讲了 Redis 集群故障发现以及故障转移、新主节点选举的整个流程，希望对你理解 Redis 的集群有帮助。
41 案例：Redis 问题汇总和相关解决方案
本文收集了一些 Redis 使用中经常遇到的一些问题，和与之相对应的解决方案，这些内容不但会出现在实际工作中，也是面试的高频问题，接下来一起来看。
缓存雪崩
缓存雪崩是指在短时间内，有大量缓存同时过期，导致大量的请求直接查询数据库，从而对数据库造成了巨大的压力，严重情况下可能会导致数据库宕机的情况叫做缓存雪崩。
我们先来看下正常情况下和缓存雪崩时程序的执行流程图，正常情况下系统的执行流程如下图所示：
缓存雪崩的执行流程，如下图所示：
以上对比图可以看出缓存雪崩对系统造成的影响，那如何解决缓存雪崩的问题？
缓存雪崩的常用解决方案 有以下几个。
加锁排队 加锁排队可以起到缓冲的作用，防止大量的请求同时操作数据库，但它的缺点是增加了系统的响应时间，降低了系统的吞吐量，牺牲了一部分用户体验。
加锁排队的代码实现，如下所示：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 String cacheKey = "userlist" ;String data = jedis.get(cacheKey);if (StringUtils.isNotBlank(data)) { return data; } else { synchronized (cacheKey) { data = jedis.get(cacheKey); if (!StringUtils.isNotBlank(data)) { data = findUserInfo(); jedis.set(cacheKey, data); } return data; } }
以上为加锁排队的实现示例，读者可根据自己的实际项目情况做相应的修改。
随机化过期时间 为了避免缓存同时过期，可在设置缓存时添加随机时间，这样就可以极大的避免大量的缓存同时失效。
示例代码如下：
1 2 3 4 5 6 int exTime = 10 * 60 ;Random random = new Random ();jedis.setex(cacheKey, exTime+random.nextInt(1000 ) , value);
设置二级缓存 二级缓存指的是除了 Redis 本身的缓存，再设置一层缓存，当 Redis 失效之后，先去查询二级缓存。
例如可以设置一个本地缓存，在 Redis 缓存失效的时候先去查询本地缓存而非查询数据库。
加入二级缓存之后程序执行流程，如下图所示：
缓存穿透
缓存穿透是指查询数据库和缓存都无数据，因为数据库查询无数据，出于容错考虑，不会将结果保存到缓存中，因此每次请求都会去查询数据库，这种情况就叫做缓存穿透。
缓存穿透执行流程如下图所示：
其中红色路径表示缓存穿透的执行路径，可以看出缓存穿透会给数据库造成很大的压力。
缓存穿透的解决方案有以下几个。
使用过滤器 我们可以使用过滤器来减少对数据库的请求，例如使用我们前面章节所学的布隆过滤器，我们这里简单复习一下布隆过滤器，它的原理是将数据库的数据哈希到 bitmap 中，每次查询之前，先使用布隆过滤器过滤掉一定不存在的无效请求，从而避免了无效请求给数据库带来的查询压力。
缓存空结果 另一种方式是我们可以把每次从数据库查询的数据都保存到缓存中，为了提高前台用户的使用体验 (解决长时间内查询不到任何信息的情况)，我们可以将空结果的缓存时间设置得短一些，例如 3~5 分钟。
缓存击穿
缓存击穿指的是某个热点缓存，在某一时刻恰好失效了，然后此时刚好有大量的并发请求，此时这些请求将会给数据库造成巨大的压力，这种情况就叫做缓存击穿。
缓存击穿的执行流程如下图所示：
它的解决方案有以下 2 个。
加锁排队 此处理方式和缓存雪崩加锁排队的方法类似，都是在查询数据库时加锁排队，缓冲操作请求以此来减少服务器的运行压力。
设置永不过期 对于某些热点缓存，我们可以设置永不过期，这样就能保证缓存的稳定性，但需要注意在数据更改之后，要及时更新此热点缓存，不然就会造成查询结果的误差。
缓存预热
首先来说，缓存预热并不是一个问题，而是使用缓存时的一个优化方案，它可以提高前台用户的使用体验。
缓存预热指的是在系统启动的时候，先把查询结果预存到缓存中，以便用户后面查询时可以直接从缓存中读取，以节约用户的等待时间。
缓存预热的执行流程，如下图所示：
缓存预热的实现思路有以下三种：
把需要缓存的方法写在系统初始化的方法中，这样系统在启动的时候就会自动的加载数据并缓存数据；
把需要缓存的方法挂载到某个页面或后端接口上，手动触发缓存预热；
设置定时任务，定时自动进行缓存预热。
小结
本文介绍了缓存雪崩产生的原因是因为短时间内大量缓存同时失效，而导致大量请求直接查询数据库的情况，解决方案是加锁、随机设置过期时间和设置二级缓存等；还介绍了查询数据库无数据时会导致的每次空查询都不走缓存的缓存穿透问题，解决方案是使用布隆过滤器和缓存空结果等；同时还介绍了缓存在某一个高并发时刻突然失效导致的缓存击穿问题，以及解决方案——加锁、设置永不过期等方案，最后还介绍了优化系统性能的手段缓存预热。
补充
面试1：Redis大key的处理方案！
概念
String：值大于 10KB
Hash、List、Set、ZSet：元素个数超过 5,000 个
查找方式
string类型通过命令查找
redis-cli -h 127.0.0.1 -p6379 -a " password”—-bigkeys
RdbTools工具
rdb dump.rdb -c memory --bytes 10240 -f redis.csv
删除方式
直接删除大key会造成阻塞，国为redis是单线程执行，阻塞期间，其他所有清求可能都会超时。超时越来越多，会造成redis连接会耗尽，产生各种异常。
低峰期删除:凌晨，观察qps，选择低的时候，无法彻底解决阻塞。
分批次删除:对于hash，使用hscan扫描法，对于集台采用srandmember每次随机取数据进行删除。对于有序集台可以使zremrangebyrank直接删除，对于列表直接pop即可。
异步删除法:用unlink代替del来删除，这样redis会将这个key放入到一个异步线程中，进行删除，这样不会阻塞主线程。
