Java 开发 Redis 面试题分享

​1、Redis 为什么这么快？

1. 什么是 Redis？

Redis 是一个使用 C 语言编写的，高性能的 K-V 非关系型数据库（NoSQL）。它支持存储多种数据类型，譬如：string，list，set，sorted set，hash 等。

Redis 的读写性能非常优秀，可java培训以达到十万级的 QPS。在 Web 应用发展的初期阶段，一个网站的访问量本身就不是很高，直接使用关系型数据库就可以应付绝大部分场景。但是随着互联网时代的崛起，人们对于网站访问速度有着越来越高的要求，直接使用关系型数据库的方案在性能上就出现了瓶颈。

目前主流应用架构的解决方案是在客户端与数据存储层之间加入一个缓存层：

假如用户第一次访问数据库中的某些数据，我们除了将数据正常返回给用户，还会将数据存储到缓存中。当用户下一次再访问这些数据，就可以直接从缓存中获取了。从缓存里拿数据比直接查询数据库要快得多，因为操作缓存相当于直接操作内存，通过缓存可以显著地提高性能。不仅如此，使用缓存还可以实现熔断机制，当我们的数据库发生崩溃时，可以直接从缓存中返回，不再让请求走数据库查询，等到数据库修复好以后再恢复调用。

最常被提及的分布式缓存中间件有两种：Redis 与 Memcached。

Memcached 也是一种 K-V 非关系型数据库，它的优点为使用起来方便简单，缺点为不支持数据的持久化存储，不支持主从同步与分片功能。

相比于 Memcached，Redis 支持更多的数据存储类型，支持 RDB，AOF 两种持久化存储机制，支持 Cluster 模式，可以实现主从同步与分片功能。

Redis 在整体上比 Memcached 强大，这也是 Redis 被许多企业作为分布式缓存首选的原因。

2. 非阻塞 I/O 与 I/O 多路复用

当我们开启了 Redis 客户端并与服务端进行连接，客户端与服务端便会通过套接字 Socket 来处理他们之间的读写请求。

譬如，我们在 Redis 客户端输入了一个 get 命令：

127.0.0.1:6379> get name

那么对于 Redis 服务端而言，都发生了哪些事儿呢？

首先，服务端必须要先监听客户端的请求（listen），然后当客户端到来时与其建立连接（accept），接着服务端需要从套接字 Socket 中读取客户端的请求（recv），对请求进行解析（parse），这里解析出请求的类型是 “get”，key 为 “name”，然后再根据 key 获取对应的 value ，最后将值返回给客户端，也就是向 Socket 写入数据（send）。

Socket 默认的读写方法是阻塞的，当 Redis 通过 recv 或 send 向客户端读写数据时，如果数据一直没有到达，那么 Redis 主线程便会一直处于阻塞状态。

Redis 自然不会使用阻塞 I/O，它将 Socket 设java培训机构置为非阻塞模式（Non_Blocking ），在非阻塞 I/O 模式下，读写方法不会产生阻塞，而是能读多少就读多少，能写多少就写多少。当客户端一直不发送数据时，主线程便不会傻傻地等待，而是直接返回，然后去做其他的事情。

除了将 Socket 设置为非阻塞 I/O，Redis 还使用了 I/O 多路复用机制，使得 Redis 在单线程模型下可以高效地处理多个 I/O 流：

Redis 基于 Reactor 模式开发了自己的文件事件处理器（File Event Handler）。它是由套接字，I/O 多路复用程序，文件事件分派器，事件处理器构成的。

I/O 多路复用程序（epoll / kqueue / select...，依据操作系统的不同，会采用不同的函数，Linux 使用的是 epoll，Mac OS 则是 kqueue）会监听多个套接字，每当一个套接字准备执行应答，读写，关闭等操作时，就会产生对应的文件事件，这些事件会存储到一个事件队列中，并由队列向文件事件分派器传送。而文件事件分派器会根据这些事件对应的类型来调用不同的事件处理器执行相应的事件回调函数。

I/O 多路复用机制使得 Redis 无需一直轮询是否有请求发生，这样就避免了资源的浪费。

3. Redis 为什么这么快？

回到我们的主要问题上——Redis 为什么这么快？

总结原因有以下几点：

基于内存操作，所以性能高使用了许多高效的数据结构虽然是单线程模型，但是使用了非阻塞 I/O 与 I/O 多路复用机制，大大提升了 Redis 的读写性能因为 Redis 是单线程模型，所以就避免了不必要的上下文切换与多线程竞争 4. Redis 到底是单线程的还是多线程的？

在上文中，我提到的 Redis 的单线程模型，指的是文件事件队列的生产者-消费者模型是单线程的。使用单线程模型的主要原因是 Redis 的性能瓶颈并不在于 CPU，而在于内存与网络带宽。既然 CPU 不是影响 Redis 速度的主要原因，那么自然就采用易于编写的单线程模型了。

从 Redis 4.0 版本开始，便引入了 Lazy Free 机制。当用户使用 DEL 命令删除比较大的 Key 时，又或者使用 FLUSHDB 或 FLUSHALL 命令删除包含大量 Key 的数据库时，很有可能造成服务器的阻塞。Redis 4.0 添加了 UNLINK 命令，这是 DEL 命令的异步版本，Redis 会使用额外的后台线程来执行删除操作，从而尽可能地避免服务器造成的阻塞：

127.0.0.1:6379> unlink name(integer) 1

此外，Redis 4.0 还可以在 FLUSHDB 和 FLUSHALL 命令后添加 ASYNC 选项，带有这个选项的删除操作也会在后台线程中执行：

127.0.0.1:6379> flushdb asyncOK127.0.0.1:6379> flushall asyncOK

如果说，从 Redis 4.0 版本开始，初步引入了多线程机制，那么 Redis 6.0 版本开始，则是正式地引入了多线程。

Redis 6.0 加入了多线程用来处理网络数据的读写和协议解析以提高网络 IO 性能，不过执行命令仍然是单线程顺序执行的。

Redis 6.0 多线程默认是禁用的，如果需要开启，则需要修改 redis.conf 配置文件：

设置 io-thread-do-reads 配置项为 yes，表示启用多线程

io-thread-do-reads yes

然后通过 io-threads 设置子线程的数量

io-threads 3

官方建议：4 核的机器建议设置为 2 或 3 个线程，8 核的机器建议设置 6 个线程，线程数一定要小于机器核数。

2、Redis 常用的数据类型有哪些？Redis 常用的五种数据类型为：

• String

• Hash

• List

• Set

• Zset

String

String 是 Redis 最基本的数据类型，通常用来做最简单的键值对存储，可以存储的值类型为字符串，整数或浮点数，支持对整数和浮点数执行自增或自减操作：

​

127.0.0.1:6379> set test:count 1OK127.0.0.1:6379> get test:count"1"127.0.0.1:6379> incr test:count(integer) 2127.0.0.1:6379> decr test:count(integer) 1

Hash

Hash 通常用来存储结构化的数据，譬如用于存储一个对象：

127.0.0.1:6379> hset test:user id 1(integer) 1127.0.0.1:6379> hset test:user username zhangsan(integer) 1127.0.0.1:6379> hget test:user id"1"127.0.0.1:6379> hget test:user username"zhangsan"

List

Redis 中的 List 类似于 Java 中的 Deque，是一种可以从两端压入（lpush，rpush）或弹出（lpop，rpop）的数据结构。我们可以使用 List 来存储列表类型的数据，譬如粉丝列表，文章评论列表等等：

127.0.0.1:6379> lpush test:ids 101 102 103(integer) 3127.0.0.1:6379> llen test:ids(integer) 3127.0.0.1:6379> lrange test:ids 0 21) "103"2) "102"3) "101"127.0.0.1:6379> lindex test:ids 0"103"127.0.0.1:6379> rpop test:ids"101"127.0.0.1:6379> lpop test:ids"103"

Set

Set 为无序集合，我们可以使用 spop 命令从集合中弹出一个随机的元素，使用 smembers 来查看集合中所有的元素：

​

127.0.0.1:6379> sadd test:teachers aaa bbb ccc ddd eee(integer) 5127.0.0.1:6379> scard test:teachers(integer) 5127.0.0.1:6379> spop test:teachers"eee"127.0.0.1:6379> smembers test:teachers1) "bbb"2) "ddd"3) "ccc"4) "aaa"

Zset

Zset 即有序集合，它通过分数（score）为集合中的成员进行从小到大的排序，使用的典型场景为排行榜：

​

127.0.0.1:6379> zadd test:students 10 aaa 20 bbb 30 ccc 40 ddd 50 eee(integer) 5127.0.0.1:6379> zcard test:students(integer) 5127.0.0.1:6379> zscore test:students ccc"30"127.0.0.1:6379> zrank test:students ccc(integer) 2127.0.0.1:6379> zrange test:students 0 21) "aaa"2) "bbb"3) "ccc"

除了以上五种常用的基本数据类型，Redis 还有一些功能强大的高级数据类型，譬如：

• HyperLogLog

• Geo

HyperLogLog 采用一种基数算法，用于完成独立总数的统计。譬如，我们有一个需求，要记录网站每天的 UV 数据（Unique Visitor）。当然，我们可以使用 Set，它能满足去重的需求，不过如果该网站的访问量非常巨大，那么使用 Set 来统计便很浪费空间了。我们可以使用 HyperLogLog 这种数据结构，它的优点是无论有多少统计数据，都只会占用 12 KB 的内存空间。不过 HyperLogLog 只是用来做基数统计的，并不会保存元数据，并且它采用的是一种不精确的统计算法，标准误差为 0.81%。即便是这样，HyperLogLog 也可以满足绝大部分的统计需求。

127.0.0.1:6379> pfadd uv user1(integer) 1127.0.0.1:6379> pfadd uv user2(integer) 1127.0.0.1:6379> pfadd uv user3 user4 user5(integer) 1127.0.0.1:6379> pfcount uv(integer) 5

Geo 则是一种用于存储地理位置信息的数据结构，支持将一个或多个经度，纬度，位置名称添加到指定的 Key 中，感兴趣的朋友可以自行了解一下～

【拓展】Redis 中，Zset 的底层数据结构是什么？

Redis 中，有序集合 Zset，其底层实现为 跳表 + 散列表。

为什么 Redis 要使用跳表来实现有序集合而不是红黑树呢？

Redis 中 Zset 主要支持以下几种操作：

• 插入一个数据

• 删除一个数据

• 查找一个数据

• 按照区间查找数据（比如查找值在 [100,356] 之间的数据）

• 迭代输出有序序列

其中，插入、删除、查找以及迭代输出有序序列这几个操作，红黑树也可以完成，并且时间复杂度跟跳表是一样的。但是，按照区间来查找数据这个操作，红黑树的效率没有跳表高。

对于按照区间查找数据，跳表可以做到 的时间复杂度定位区间的起点，然后在原始链表中顺序往后遍历就可以了，这样做非常高效。

当然，Redis 之所以使用跳表来实现有序集合，还有其他原因，比如，跳表比起红黑树的实现简直是容易多了，而简单就意味着可读性好，不容易出错。还有，跳表更加灵活，它可以通过改变上层索引构建策略，有效平衡执行效率和内存消耗。

3、如何从海量的 Key 中查询出某一固定前缀的 Key？

首先，要说明的一点是，我们的 Redis 正在给线上的业务提供服务。

如果在数据规模不大的情况下，我们可以使用 keys pattern 命令来返回当前的 Key 中所有符合 pattern 定义的 Key，譬如要查询所有以 “K” 开头的 Key，我们可以输入这样的命令：

127.0.0.1:6379[10]> keys K*

但是，如果题目中加上“海量数据”这一前提，使用 keys 命令就有一定的隐患了。

在 Redis 拥有数百万甚至是几千万以上的 Key 时，执行 keys 命令的速度会非常慢，keys 命令会将所有的 Key 全部遍历一遍，其复杂度为 。不仅如此，在数据量极大的情况下，该命令会阻塞 Redis I/O 多路复用的主线程一段时间，如果我们的 Redis 正在为线上的业务提供服务，主线程一旦被阻塞，那么在此期间其他的发送向 Redis 服务端的命令都会被阻塞，从而导致 Redis 出现卡顿，引发响应超时的问题。

取而代之的，我们可以使用 scan 命令：

scan cursor [MATCH pattern] [COUNT count]

scan 命令的时间复杂度同样为 ，不过它需要迭代多次才能返回完整的数据，并且每次返回的数据有可能会产生重复。

cursor 为查询游标，游标从 0 开始，也从 0 结束。每次返回的数据中，都有下一次游标该传入的值，我们通过这个值，再去进行下一轮的迭代。scan 命令并不保证每次执行都会返回某个给定数量的元素，每次迭代返回的元素数量都是不确定的。所以，即便是返回了 0 条数据，只要返回的游标值不为 0，就说明需要继续迭代，只有当返回的游标值为 0 时，才代表迭代完毕。

pattern 为匹配的表达式。

count 指定了每次迭代返回元素的最大数量，默认值为 10。

示例如下：

​

127.0.0.1:6379[10]> scan 0 match K* count 101) "20"2) 1) "K1"   2) "K5"   3) "K4"127.0.0.1:6379[10]> scan 20 match K* count 101) "33"2) 1) "K3"   2) "K6"   3) "K8"127.0.0.1:6379[10]> scan 33 match K* count 101) "11"2) 1) "K9"   2) "K2"127.0.0.1:6379[10]> scan 11 match K* count 101) "0"2) 1) "K7"

scan 命令返回的第一条数据便是我们下一次迭代时的游标，当返回 “0” 时，就表示所有匹配 pattern 的 Key 已经全部返回，迭代结束。

scan 命令通过游标分布执行，不会产生线程阻塞，所以非常适合使用于海量数据的生产环境下。

在 Spring Boot 整合 Redis 时，我们也可以在 Java 代码中调用 Redis 的 scan 命令，这里需要注意的是，因为每一次迭代都有可能产生重复的数据，所以我们应该使用 Set 或 Map 这样的数据结构来去重。

示例代码如下：

import org.junit.jupiter.api.Test;import org.springframework.boot.test.context.SpringBootTest;import org.springframework.dao.DataAccessException;import org.springframework.data.redis.connection.RedisConnection;import org.springframework.data.redis.core.Cursor;import org.springframework.data.redis.core.RedisCallback;import org.springframework.data.redis.core.RedisTemplate;import org.springframework.data.redis.core.ScanOptions;import org.springframework.test.context.ContextConfiguration;

import javax.annotation.Resource;import java.util.HashMap;import java.util.Map;

@SpringBootTest@ContextConfiguration(classes = RedisScanCommandApplication.class)public class ScanCommandTest {

@Resourceprivate RedisTemplate redisTemplate;

@Testpublic void testScan() {    String pattern = "K*";    Long limit = 1000L;    Map<String, String> map = scan(redisTemplate, pattern, limit);    System.out.println(map);}
private Map<String, String> scan(RedisTemplate redisTemplate, String pattern, Long limit) {    return (Map<String, String>) redisTemplate.execute(new RedisCallback() {        @Override        public Object doInRedis(RedisConnection connection) throws DataAccessException {            Map<String, String> map = new HashMap<>();            Cursor<byte[]> cursor = connection.scan(new ScanOptions                    .ScanOptionsBuilder()                    .match(pattern)                    .count(limit)                    .build());            while (cursor.hasNext()) {                byte[] bytesKey = cursor.next();                byte[] bytesValue = connection.get(bytesKey);                String key = String.valueOf(redisTemplate.getKeySerializer().deserialize(bytesKey));                String value = String.valueOf(redisTemplate.getValueSerializer().deserialize(bytesValue));                map.put(key, value);            }            return map;        }    });}

}

执行单元测试，程序输出的结果如下：

{K1=K-1, K2=K-2, K3=K-3, K4=K-4, K5=K-5, K6=K-6, K7=K-7, K8=K-8, K9=K-9}

4、如何通过 Redis 实现分布式锁？

实现分布式锁必须要满足以下几点条件：

互斥性安全性不能出现死锁互斥性指的是，在任意时刻，只能有一个客户端获取到锁。

安全性指的是，锁只能被持有该锁的客户端删除，不能由其他的客户端删除。

如果持有锁的客户端因为某些原因宕机而未能释放锁，导致其他客户端再也无法获取到锁便产生了死锁的情况。对于实现分布式锁来说，必须有机制来避免死锁的发生。

在 Redis 中，有一个 SETNX key value 的指令，该指令的含义为：Set if not exists。也就是说，如果 Key 不存在，那么使用 SETNX 便可以创建 Key 并赋值成功；反之，如果 Key 存在，就什么也不做。

127.0.0.1:6379[10]> setnx lock Kim(integer) 1127.0.0.1:6379[10]> setnx lock Tom(integer) 0 // 加锁失败

如上面的代码所示，我们第一次加锁成功，第二次加锁则是失败的。而释放锁也很简单，直接使用 DEL 命令删除这个 Key 即可：

127.0.0.1:6379[10]> del lock(integer) 1

这个方案看似简单完美，不过却存在着一个很大的问题。当客户端 A 拿到了锁以后，发生了某些原因导致宕机而未能释放锁，那么其他的客户端便无法获取到该锁，进而就产生了死锁的情况。

为了解决这一问题，我们需要给锁设定一个过期时间。在 Redis 中，有一个 EXPIRE key seconds 指令，它可以设置 Key 的存活时间，当 Key 过期时，会被自动删除：

127.0.0.1:6379[10]> setnx lock kim(integer) 1127.0.0.1:6379[10]> expire lock 10(integer) 1

这样一来，锁的释放就得到了保证。

不过该方案也是有问题的。SETNX 与 EXPIRE 属于两条指令，不能保证原子性操作。假如，客户端 A 在执行 SETNX 命令后，还没有来得及为锁设置过期时间就发生了宕机，还是会发生死锁的问题。有没有一种方案可以让使这两条命令绑定在一起呢？

答案是：有的。

在 Redis 2.6 版本开始，便扩展了 SET 命令：

SET key value [EX seconds|PX milliseconds] [NX|XX]

该命令中，新增的参数含义如下：

EX seconds 为设置 Key 的过期时间，其单位为秒 PX milliseconds 为设置 Key 的过期时间，其单位为毫秒 NX 的含义为只有当 Key 不存在时，才会对 Key 进行设置 XX 的含义为只有当 Key 存在时，才会对 Key 进行设置这样，我们便可以使用一条命令来实现 SETNX 与 EXPIRE 了 ：

127.0.0.1:6379[10]> set lock kim ex 10 nxOK

在 Spring 整合 Jedis 时，我们可以使用这样的代码完成加锁的逻辑：

private boolean lock(Jedis jedis, String key, String value, int expireSeconds) {String result = jedis.set(key, value, "NX", "EX", expireSeconds);if (result.equals("OK")) {// do sthreturn true;}return false;}

那么如何释放锁呢？

大家可能想到的姿势是：

private void unlock(Jedis jedis, String key, String value) {if (value.equals(jedis.get(key)))jedis.del(key);}

不过这段代码很显然是不能保证原子性的。

举个例子：

假设锁的过期时间设置为 10 秒。客户端 A 在执行 get 锁的命令时，锁过期，与此同时，客户端 B 获取到了锁，而客户端 A 又继续执行了 del 命令，释放了客户端 B 刚刚获取到的锁，这便违背了安全性。

针对于这种情况，我们的解决方法是将获取锁与释放锁这两个命令写成 Lua 脚本交给 Redis 来执行。因为 Redis 执行命令是单线程执行的，所以就保证 Lua 脚本的执行是一个原子的操作。

Spring 整合 Jedis 时，我们可以使用这样的代码完成释放锁的逻辑：

private boolean unlock(Jedis jedis, String key, String value) {String luaScript = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";Object result = jedis.eval(luaScript, Collections.singletonList(key), Collections.singletonList(value));if (result.equals(1L)) {// do sthreturn true;}return false;}

Redisson 实现分布式锁

再来思考一个问题：当我们使用上述方案，对锁设置了过期时间为 10 秒。

假如客户端 A 持有锁，正常的业务执行了 10 秒还未执行完毕，锁因为过期就被释放了，这种情况要怎么办？

Redisson 给出的解决方案是使用看门狗线程。

看门狗线程会定时监测锁的失效时间，如果锁快要过期了，而业务还没有执行完毕就会对锁进行“续期”操作，重新设置锁的过期时间。

Redisson SDK 封装了许多易用的功能，包括：

可重入锁（ReentrantLock）公平锁（FairLock）联锁（MultiLock）读写锁（ReadWriteLock）红锁（RedLock）... ...关于 Redisson 的具体使用，以及大名鼎鼎的红锁 RedLock，这些部分我会在后续的系列文章中进行介绍，这里就先不展开了。

文章来源：憨憨二师兄