redis技术总结

写在前面的话

网上Java的资料零零散散，甚至有一些错误，作者希望能结合自己的实际开发经验和面试经验，对Redis知识体系进行系统梳理。

本文参考主要【引用】中的内容，并结合自己的日常积累，欢迎留言交流指正。

引用

• 注：本文主要参考
• https://mp.weixin.qq.com/s/vXBFscXqDcXS_VaIERplMQ！
  • https://github.com/AobingJava/JavaFamily
  • https://github.com/Snailclimb/JavaGuide

基础篇

谈谈你对缓存的理解

Redis 优缺点

优点

• 读写性能优异， Redis能读的速度是 110000 次/s，写的速度是 81000 次/s。
• 支持数据持久化，支持 AOF 和 RDB 两种持久化方式。
• 支持事务，Redis 的所有操作都是原子性的，同时 Redis 还支持对几个操作合并后的原子性执行。
• 数据结构丰富，除了支持 string 类型的 value 外还支持 hash、set、zset、list 等数据结构。
• 支持主从复制，主机会自动将数据同步到从机，可以进行读写分离，多种集群方案。

缺点

• 数据库 容量受到物理内存的限制，不能用作海量数据的高性能读写，因此 Redis 适合的场景主要局限在较小数据量的高性能操作和运算上。
• Redis 不具备自动容错和恢复功能，主机从机的宕机都会导致前端部分读写请求失败，需要等待机器重启或者手动切换前端的 IP 才能恢复。
• 主机宕机，宕机前有部分数据未能及时同步到从机，切换 IP 后还会引入数据不一致的问题，降低了 系统的可用性。
• Redis 较难支持在线扩容，在集群容量达到上限时在线扩容会变得很复杂。为避免这一问题，运维人员在系统上线时必须确保有足够的空间，这对资源造成了很大的浪费。

Redis 和 Memcached 有啥区别，为啥选Redis？

• Redis 相比 Memcached 来说，拥有更多的数据结构，能支持更丰富的数据操作。如果需要缓存能够支持更复杂的结构和操作， Redis 会是不错的选择。
• Redis 原生支持集群模式，而 Memcached 没有原生的集群模式，需要依靠客户端来实现往集群中分片写入数据。
• 性能对比：由于 Redis 只使用单核，而 Memcached 可以使用多核，所以平均每一个核上 Redis 在存储小数据时比 Memcached 性能更高。而在 100k 以上的数据中，Memcached 性能要高于 Redis，虽然 Redis 最近也在存储大数据的性能上进行优化，但是比起 Memcached，还是稍有逊色。

Redis 为什么这么快？

简单总结：

1. 纯内存操作：读取不需要进行磁盘 I/O，所以比传统数据库要快上不少；(但不要有误区说磁盘就一定慢，例如 Kafka 就是使用磁盘顺序读取但仍然较快)
2. 单线程，无锁竞争：这保证了没有线程的上下文切换，不会因为多线程的一些操作而降低性能；
3. 多路 I/O 复用模型，非阻塞 I/O：采用多路 I/O 复用技术可以让单个线程高效的处理多个网络连接请求（尽量减少网络 IO 的时间消耗）；
4. 高效的数据结构，加上底层做了大量优化：Redis 对于底层的数据结构和内存占用做了大量的优化，例如不同长度的字符串使用不同的结构体表示，HyperLogLog 的密集型存储结构等等..

​ 多个Socket可能会并发产生不同的操作，每个操作对应不同的文件事件，但是 IO 多路复用程序会监听多个 Socket，会将 Socket 产生的事件放入队列中排队，事件分派器每次从队列中取出一个事件，把该事件交给对应的事件处理器进行处理。

Redis 有哪些数据结构

Redis 可以存储 键 和 不同类型数据结构值 之间的映射关系。键的类型只能是字符串，而值除了支持最 基础的五种数据类型 外，还支持一些 高级数据类型：

“

一定要说出一些高级数据结构 (当然你自己也要了解.. 下面会说到的别担心)，这样面试官的眼睛才会亮。

可重点了解HyperLogLog和Pub/Sub

pub/sub有什么缺点？

在消费者下线的情况下，生产的消息会丢失，得使用专业的消息队列如RocketMQ等。

实际项目缓存应用场景

String

• 全局订单号
• 分布式Session共享（否则从一个页面跳转到另一个页面可能Session就过期了）
• 分布式锁

Hash：相当于嵌套Map

• 存储用户信息、商品信息等

注意：

• bigkey问题

List

• 微信推送消息列表
• 异步任务队列（解决redis数据不一致问题）
  • rpush生产消息，lpop消费消息。当lpop没有消息的时候，要适当sleep一会再重试。另外list还有个指令叫blpop，在没有消息的时候，它会阻塞住直到消息到来。

Set

• Ip黑名单
• 微博关注取交集

ZSet

• 接口调用用户TopK

• 秒杀实现

HyperLogLog

• 实现网站所有网页pv(浏览量，用户没点一次记录一次)、uv(同一个用户一天之内的多次访问请求只能计数一次)统计。

注⚠️：

在大量数据的情况下，有一定误差。HyperLogLog 提供了两个指令 PFADD 和 PFCOUNT，字面意思就是一个是增加，另一个是获取计数。PFADD 和 set 集合的 SADD 的用法是一样的，来一个用户 ID，就将用户 ID 塞进去就是，PFCOUNT 和 SCARD的用法是一致的，直接获取计数值。

GeoHash

• 查找附近人

使用缓存会出现什么问题？

一般来说有如下几个问题，回答思路遵照 是什么 → 为什么 → 怎么解决：

1. 缓存雪崩问题；
2. 缓存穿透问题；
3. 缓存和数据库双写一致性问题；
4. 并发写竞争

缓存雪崩问题

对于 “Redis 挂掉了，请求全部走数据库” 这样的情况，我们还可以有如下的思路：

• 事发前：实现 Redis 的高可用(主从架构 + Sentinel 或者 Redis Cluster)，尽量避免 Redis 挂掉这种情况发生。
• 事发中：万一 Redis 真的挂了，我们可以设置本地缓存(ehcache) + 限流(hystrix)，尽量避免我们的数据库被干掉(起码能保证我们的服务还是能正常工作的)
• 事发后：Redis 持久化，重启后自动从磁盘上加载数据，快速恢复缓存数据。

缓存穿透问题

布隆过滤器的实现：使用 Google 开源的 Guava 中自带的布隆过滤器，Guava 中布隆过滤器的实现算是比较权威的，所以实际项目中我们不需要手动实现一个布隆过滤器。

参数做校验，不合法的参数直接代码Return，比如：id 做基础校验，id <=0的直接拦截等。

缓存击穿

缓存击穿是指一个Key非常热点，在不停的扛着大并发，大并发集中对这一个点进行访问，当这个Key在失效的瞬间，持续的大并发就穿破缓存，直接请求数据库，就像在一个完好无损的桶上凿开了一个洞。

解决：

定时任务刷缓存（占用系统资源、10分钟过期、9分钟刷），设置热点数据永远不过期。或者访问数据库时加上互斥锁就能搞定了

热点Key问题

https://segmentfault.com/a/1190000019745366?utm_source=tag-newest

如何解决？

针对于热点Key的解决方案网上的查找出来无非就是两种

• 服务端缓存：即将热点数据缓存至服务端的内存中
• 备份热点Key：即将 热点Key + 随机数，随机分配至Redis其他节点中。这样访问热点key的时候就不会全部命中到一台机器上了。（给hot key加上前缀或者后缀，把一个hotkey 的数量变成 redis 实例个数N的倍数M，从而由访问一个 redis key 变成访问 N * M 个redis key）

其实这两个解决方案前提都是知道了热点Key是什么的情况，那么如何找到热点key呢？

热点检测

1. 凭借经验，进行预估：例如提前知道了某个活动的开启，那么就将此Key作为热点Key
2. 客户端收集：在操作Redis之前对数据进行统计
3. 抓包进行评估：Redis使用TCP协议与客户端进行通信，通信协议采用的是RESP，所以能进行拦截包进行解析
4. 在proxy层，对每一个 redis 请求进行收集上报
5. Redis自带命令查询：Redis4.0.4版本提供了redis-cli –hotkeys就能找出热点Key

缓存与数据库双写一致问题

对于读操作，流程是这样的

上面讲缓存穿透的时候也提到了：如果从数据库查不到数据则不写入缓存。

一般我们对读操作的时候有这么一个固定的套路：

• 如果我们的数据在缓存里边有，那么就直接取缓存的。
• 如果缓存里没有我们想要的数据，我们会先去查询数据库，然后将数据库查出来的数据写到缓存中。
• 最后将数据返回给请求

如果仅仅查询的话，缓存的数据和数据库的数据是没问题的。但是，当我们要更新时候呢？各种情况很可能就造成数据库和缓存的数据不一致了。

• 这里不一致指的是：数据库的数据跟缓存的数据不一致

从理论上说，只要我们设置了键的过期时间，我们就能保证缓存和数据库的数据最终是一致的。因为只要缓存数据过期了，就会被删除。随后读的时候，因为缓存里没有，就可以查数据库的数据，然后将数据库查出来的数据写入到缓存中。

除了设置过期时间，我们还需要做更多的措施来尽量避免数据库与缓存处于不一致的情况发生。

对于更新操作

一般来说，执行更新操作时，我们会有两种选择：

• 先操作数据库，再操作缓存
• 先操作缓存，再操作数据库

首先，要明确的是，无论我们选择哪个，我们都希望这两个操作要么同时成功，要么同时失败。所以，这会演变成一个分布式事务的问题。

所以，如果原子性被破坏了，可能会有以下的情况：

• 操作数据库成功了，操作缓存失败了。
• 操作缓存成功了，操作数据库失败了。

如果第一步已经失败了，我们直接返回Exception出去就好了，第二步根本不会执行。

下面我们具体来分析一下吧。

操作缓存

操作缓存也有两种方案：

• 更新缓存
• 删除缓存

一般我们都是采取删除缓存缓存策略的，原因如下：

1. 高并发环境下，无论是先操作数据库还是后操作数据库而言，如果加上更新缓存，那就更加容易导致数据库与缓存数据不一致问题。(删除缓存直接和简单很多)
2. 如果每次更新了数据库，都要更新缓存【这里指的是频繁更新的场景，这会耗费一定的性能】，倒不如直接删除掉。等再次读取时，缓存里没有，那我到数据库找，在数据库找到再写到缓存里边(体现懒加载)

基于这两点，对于缓存在更新时而言，都是建议执行删除操作！

先更新数据库，再删除缓存

正常的情况是这样的：

• 先操作数据库，成功；
• 再删除缓存，也成功；

如果原子性被破坏了：

• 第一步成功(操作数据库)，第二步失败(删除缓存)，会导致数据库里是新数据，而缓存里是旧数据。
• 如果第一步(操作数据库)就失败了，我们可以直接返回错误(Exception)，不会出现数据不一致。

如果在高并发的场景下，出现数据库与缓存数据不一致的概率特别低，也不是没有：

• 缓存刚好失效
• 线程A查询数据库，得一个旧值
• 线程B将新值写入数据库
• 线程B删除缓存
• 线程A将查到的旧值写入缓存

要达成上述情况，还是说一句概率特别低：

因为这个条件需要发生在读缓存时缓存失效，而且并发着有一个写操作。而实际上数据库的写操作会比读操作慢得多，而且还要锁表，而读操作必需在写操作前进入数据库操作，而又要晚于写操作更新缓存，所有的这些条件都具备的概率基本并不大。

对于这种策略，其实是一种设计模式：Cache Aside Pattern

删除缓存失败的解决思路：

• 将需要删除的key发送到消息队列中
• 自己消费消息，获得需要删除的key
• 不断重试删除操作，直到成功

先删除缓存，再更新数据库

正常情况是这样的：

• 先删除缓存，成功；
• 再更新数据库，也成功；

如果原子性被破坏了：

• 第一步成功(删除缓存)，第二步失败(更新数据库)，数据库和缓存的数据还是一致的。
• 如果第一步(删除缓存)就失败了，我们可以直接返回错误(Exception)，数据库和缓存的数据还是一致的。

看起来是很美好，但是我们在并发场景下分析一下，就知道还是有问题的了：

• 线程A删除了缓存
• 线程B查询，发现缓存已不存在
• 线程B去数据库查询得到旧值
• 线程B将旧值写入缓存
• 线程A将新值写入数据库

所以也会导致数据库和缓存不一致的问题。

并发下解决数据库与缓存不一致的思路：

• 将删除缓存、修改数据库、读取缓存等的操作积压到队列里边，实现串行化。

对比两种策略

我们可以发现，两种策略各自有优缺点：

• 先删除缓存，再更新数据库

• • 在高并发下表现不如意，在原子性被破坏时表现优异

• 先更新数据库，再删除缓存(Cache Aside Pattern设计模式)

• • 在高并发下表现优异，在原子性被破坏时表现不如意

其他保障数据一致的方案与资料

可以用databus或者阿里的canal监听binlog进行更新。

参考资料：

• 缓存更新的套路

• • https://coolshell.cn/articles/17416.html

• 如何保证缓存与数据库双写时的数据一致性？

• • https://github.com/doocs/advanced-java/blob/master/docs/high-concurrency/redis-consistence.md

• 分布式之数据库和缓存双写一致性方案解析

• • https://zhuanlan.zhihu.com/p/48334686

• Cache Aside Pattern

• • https://blog.csdn.net/z50l2o08e2u4aftor9a/article/details/81008933

多系统并发操作Redis带来的问题

嗯嗯这个问题我以前开发的时候遇到过，其实并发过程中确实会有这样的问题，比如下面这样的情况

系统A、B、C三个系统，分别去操作Redis的同一个Key，本来顺序是1，2，3是正常的，但是因为系统A网络突然抖动了一下，B，C在他前面操作了Redis，这样数据不就错了么。

就好比下单，支付，退款三个顺序你变了，你先退款，再下单，再支付，那流程就会失败，那数据不就乱了？你订单还没生成你却支付，退款了？明显走不通了，这在线上是很恐怖的事情。

如何解决呢？

我们可以找个管家帮我们管理好数据的嘛！

1、某个时刻，多个系统实例都去更新某个 key。可以基于 Zookeeper 实现分布式锁。每个系统通过 Zookeeper 获取分布式锁，确保同一时间，只能有一个系统实例在操作某个 Key，别人都不允许读和写。

2、你要写入缓存的数据，都是从 MySQL 里查出来的，都得写入 MySQL 中，写入 MySQL 中的时候必须保存一个时间戳，从 MySQL 查出来的时候，时间戳也查出来。

每次要写之前，先判断一下当前这个 Value 的时间戳是否比缓存里的 Value 的时间戳要新。如果是的话，那么可以写，否则，就不能用旧的数据覆盖新的数据。

持久化篇

什么是持久化？

Redis 的数据 全部存储 在 内存 中，如果 突然宕机，数据就会全部丢失，因此必须有一套机制来保证 Redis 的数据不会因为故障而丢失，这种机制就是 Redis 的 持久化机制，它会将内存中的数据库状态 保存到磁盘 中。

两者可同时共存，先加载.aof。

RDB

特点

• 在指定的时间间隔内将内存中的数据集快照写入磁盘，也就是行话讲的Snapshot快照，它恢复时是将快照文件直接读到内存里。rdb保存的是dump.rdb文件。
• Redis会单独创建（fork）一个子进程来进行持久化，会先将数据写入到一个临时文件中，待持久化过程都结束了，再用这个临时文件替换上次持久化好的文件。

整个过程中，主进程是不进行任何IO操作的，这就确保了极高的性能。如果需要进行大规模数据的恢复，且对于数据恢复的完整性不是非常敏感，那RDB方式要比AOF方式更加的高效。RDB的缺点是最后一次持久化后的数据可能丢失。

配置使用

• 配置文件中默认的快照配置，冷拷贝后重新使用，可以cp dump.rdb dump_new.rdb
• 命令save或者是bgsave。

Save：save时只管保存，其它不管，全部阻塞.

BGSAVE：Redis会在后台异步进行快照操作，快照同时还可以响应客户端请求。

可以通过lastsave命令获取最后一次成功执行快照的时间.

如何恢复

将备份文件 (dump.rdb) 移动到 redis 安装目录并启动服务即可

CONFIG GET dir获取目录 

如何停止

动态所有停止RDB保存规则的方法：redis-cli config set save ""

AOF

特点

• 以日志的形式来记录每个写操作，将Redis执行过的所有写指令记录下来(读操作不记录)，
• 只许追加文件但不可以改写文件，redis重启根据日志文件的内容将写指令从前到后执行一次以完成数据的恢复工作。

配置使用

• 修改默认的appendonly no，改为yes

同步频率

每次修改都同步：appendfsync always 同步操作，每次发生数据变更会被立即记录到磁盘，性能较差但数据完整性比较好

每秒同步一次：appendfsync everysec 异步操作，每秒记录如果一秒内宕机，有数据丢失

不同步：appendfsync no 从不同步

被写坏的AOF文件修复：

redis-check-aof --fix进行修复

恢复：重启redis然后重新加载

RDB与AOF优劣比对：

RDB和AOF并不互斥，它俩可以同时使用。

• RDB的优点：载入时恢复数据快、文件体积小。
• RDB的缺点：会一定程度上丢失数据(因为系统一旦在定时持久化之前出现宕机现象，此前没有来得及写入磁盘的数据都将丢失。)
• AOF的优点：丢失数据少(默认配置只丢失一秒的数据)。
• AOF的缺点：恢复数据相对较慢，文件体积大

如果Redis服务器同时开启了RDB和AOF持久化，服务器会优先使用AOF文件来还原数据(因为AOF更新频率比RDB更新频率要高，还原的数据更完善)。

实际生产应redis宕机的话，应使用rdb进行快速恢复，然后使用aof进行补充完整。

分布式锁

　　分布式锁其实可以理解为：控制分布式系统有序的去对共享资源进行操作，通过互斥来保持一致性。 举个不太恰当的例子：假设共享的资源就是一个房子，里面有各种书，分布式系统就是要进屋看书的人，分布式锁就是保证这个房子只有一个门并且一次只有一个人可以进，而且门只有一把钥匙。然后许多人要去看书，可以，排队，第一个人拿着钥匙把门打开进屋看书并且把门锁上，然后第二个人没有钥匙，那就等着，等第一个出来，然后你在拿着钥匙进去，然后就是以此类推。

使用场景

在识别完成后，如识别成功，需要先查出接口当前使用次数，判断一下，然后给接口使用表进行减一操作，统计表进行加一操作。我们知道这个过程在并发条件下是线程不安全的，为了保证线程的安全性在此加了分布式锁。

单机锁解决单机内存共享变量问题，分布式锁解决多台机器共享变量问题，例如并发扣减库存。

实现原理

• 互斥性
  • 保证同一时间只有一个客户端可以拿到锁，也就是可以对共享资源进行操作
• 安全性
  • 只有加锁的服务才能有解锁权限，也就是不能让a加的锁，bcd都可以解锁，如果都能解锁那分布式锁就没啥意义了
  • 可能出现的情况就是a去查询发现持有锁，就在准备解锁，这时候忽然a持有的锁过期了，然后b去获得锁，因为a锁过期，b拿到锁，这时候a继续执行第二步进行解锁如果不加校验，就将b持有的锁就给删除了
• 避免死锁
• • 出现死锁就会导致后续的任何服务都拿不到锁,不能再对共享资源进行任何操作了
• 保证加锁与解锁操作是原子性操作
• • 这个其实属于是实现分布式锁的问题，假设a用redis实现分布式锁
  • 假设加锁操作，操作步骤分为两步：
  • 1，设置key set（key，value）2，给key设置过期时间
• • 假设现在a刚实现set后，程序崩了就导致了没给key设置过期时间就导致key一直存在就发生了死锁

分布式锁的几种方式

　　实现分布式锁的方式有很多，只要满足上述条件的都可以实现分布式锁，比如数据库，redis，zookeeper，在这里就先讲一下如何使用redis实现分布式锁

使用redis实现分布式锁

• 使用redis命令 set key value NX EX max-lock-time 实现加锁（同时把setnx和expire合成一条指令来用，否则setnx之后执行expire之前进程意外crash，这个锁就永远得不到释放了。）
• 使用redis命令 EVAL 实现解锁

加锁：

Jedis jedis = new Jedis("127.0.0.1", 6379);

 private static final String SUCCESS = "OK";



 /**

  * 加锁操作

  * @param key 锁标识

  * @param value 客户端标识

  * @param timeOut 过期时间

  */

 public Boolean lock(String key,String value,Long timeOut){

     String var1 = jedis.set(key,value,"NX","EX",timeOut);

     if(LOCK_SUCCESS.equals(var1)){

         return true;

     }

     return false;

 }

解读：

• 加锁操作：jedis.set(key,value,”NX”,”EX”,timeOut)【保证加锁的原子操作】

  

---

解锁

Jedis jedis = new Jedis("127.0.0.1", 6379);

 private static final Long UNLOCK_SUCCESS = 1L;


 /**

  * 解锁操作

  * @param key 锁标识

  * @param value 客户端标识

  * @return

  */

 public static Boolean unLock(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 var2 = jedis.eval(luaScript,Collections.singletonList(key), Collections.singletonList(value));

     if (UNLOCK_SUCCESS == var2) {
         return true;
     }

     return false;

 }

解读:

• luaScript 这个字符串是个lua脚本，代表的意思是如果根据key拿到的value跟传入的value相同就执行del，否则就返回0【保证安全性】
• jedis.eval(String,list,list);这个命令就是去执行lua脚本，KEYS的集合就是第二个参数，ARGV的集合就是第三参数【保证解锁的原子操作】

上述就实现了怎么使用redis去正确的实现分布式锁，但是有个小缺陷就是锁过期时间要设置为多少合适，这个其实还是需要去根据业务场景考量一下的

---

重试机制：　　

上面那只是讲了加锁与解锁的操作，试想一下如果在业务中去拿锁如果没有拿到是应该阻塞着一直等待还是直接返回，这个问题其实可以写一个重试机制，根据重试次数和重试时间做一个循环去拿锁，当然这个重试的次数和时间设多少合适，是需要根据自身业务去衡量的

/**

 * 重试机制

 * @param key 锁标识

 * @param value 客户端标识

 * @param timeOut 过期时间

 * @param retry 重试次数

 * @param sleepTime 重试间隔时间

 * @return

 */

public Boolean lockRetry(String key,String value,Long timeOut,Integer retry,Long sleepTime){

    Boolean flag = false;

    try {

        for (int i=0;i<retry;i++){

            flag = lock(key,value,timeOut);

            if(flag){

                break;

            }

            Thread.sleep(sleepTime);

        }

    }catch (Exception e){

        e.printStackTrace();

    }

    return flag;

}

进阶分布式锁

以上实现方式会出现并发竞争，系统异常导致锁未释放，锁过期时间不合理，当前线程锁被其他线程解锁等问题。以下为优化方案：

stringRedisTemplate实现(自定义逻辑)

        /**
     * 模拟秒杀场景，利用Redis实现分布式锁（原生实现）
     */
    @Test
    public void test1() {
        // 设置clientId，避免在高并发场景下自己的锁被他人解了
      // 这里可以加上requestId放在value里面，在finally释放锁之前判断一下是不是当前请求的锁
        String clientId = UUID.randomUUID().toString();
        String lockKey = "lock";

        try {
            // 加锁
            // 设置过期时间为10s，避免系统异常导致锁无法释放（不安全升级）
            Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, clientId, 10, TimeUnit.SECONDS);
            if (!result) {
                return;
            }
            //获取当前库存
            int stock = Integer.parseInt(Objects.requireNonNull(stringRedisTemplate.opsForValue().get("stock")));
            if (stock > 0) {
                int realStock = stock - 1;
                stringRedisTemplate.opsForValue().set("stock", realStock + "");
                System.out.println("扣减成功，剩余库存" + realStock);
            } else {
                System.out.println("库存不足");
            }

        } catch (Exception e) {
            System.out.println(e.getMessage());
        } finally {
            // 释放锁
            if (clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))) {
                stringRedisTemplate.delete(lockKey);
            }
        }
    }

Redisson实现

通过 eval 命令来执行 Lua 脚本，保证原子性

默认30S过期时间

/**
     * 模拟秒杀场景，利用Redisson实现分布式锁
     *
     * SpringBoot集成redisson(单机,集群,哨兵)：参考 https://www.jianshu.com/p/2b19dec72ab0
     *
     * 这里简单实现
     */
    @Test
    public void test2() {

        String lockKey = "lock";
        RLock redissonLock = redisson.getLock(lockKey);

        try {
            // 加锁
            redissonLock.lock();
            //获取当前库存
            int stock = Integer.parseInt(Objects.requireNonNull(stringRedisTemplate.opsForValue().get("stock")));
            if (stock > 0) {
                int realStock = stock - 1;
                stringRedisTemplate.opsForValue().set("stock", realStock + "");
                System.out.println("扣减成功，剩余库存" + realStock);
            } else {
                System.out.println("库存不足");
            }
        } catch (Exception e) {
            System.out.println(e.getMessage());
        } finally {

            // 释放锁
            redissonLock.unlock();
        }
    }

@SpringBootApplication
@MapperScan("com.example.mapper")
public class RedisApplication {

    public static void main(String[] args) {
        SpringApplication.run(RedisApplication.class, args);
    }

    @Bean
    public Redisson redisson() {
        // 单机模式
        Config config = new Config();
        config.useSingleServer().setAddress("redis://172.16.219.215:6379").setDatabase(0);
        return (Redisson) Redisson.create(config);
    }

}

Redisson底层原理

基本同自己stringRedisTemplate实现的逻辑，补充：看门锁在原线程中开启一个子线程，定时检测父线程的锁是否还存在，如果存在则进行锁续命。

加锁和解锁（判断锁存在即删除锁）都通过用lua脚本保证了自己的原子性。

分布式锁性能优化

其实说出来也很简单，相信很多人看过java里的ConcurrentHashMap的源码和底层原理，应该知道里面的核心思路，就是分段加锁！

把数据分成很多个段，每个段是一个单独的锁，所以多个线程过来并发修改数据的时候，可以并发的修改不同段的数据。不至于说，同一时间只能有一个线程独占修改ConcurrentHashMap中的数据。

另外，Java 8中新增了一个LongAdder类，也是针对Java 7以前的AtomicLong进行的优化，解决的是CAS类操作在高并发场景下，使用乐观锁思路，会导致大量线程长时间重复循环。

LongAdder中也是采用了类似的分段CAS操作，失败则自动迁移到下一个分段进行CAS的思路。

其实分布式锁的优化思路也是类似的，之前我们是在另外一个业务场景下落地了这个方案到生产中，不是在库存超卖问题里用的。

但是库存超卖这个业务场景不错，很容易理解，所以我们就用这个场景来说一下。大家看看下面的图：

其实这就是分段加锁。你想，假如你现在iphone有1000个库存，那么你完全可以给拆成20个库存段，要是你愿意，可以在数据库的表里建20个库存字段，比如stock_01，stock_02，类似这样的，也可以在redis之类的地方放20个库存key。

总之，就是把你的1000件库存给他拆开，每个库存段是50件库存，比如stock_01对应50件库存，stock_02对应50件库存。

接着，每秒1000个请求过来了，好！此时其实可以是自己写一个简单的随机算法，每个请求都是随机在20个分段库存里，选择一个进行加锁。

bingo！这样就好了，同时可以有最多20个下单请求一起执行，每个下单请求锁了一个库存分段，然后在业务逻辑里面，就对数据库或者是Redis中的那个分段库存进行操作即可，包括查库存 -> 判断库存是否充足 -> 扣减库存。

这相当于什么呢？相当于一个20毫秒，可以并发处理掉20个下单请求，那么1秒，也就可以依次处理掉20 * 50 = 1000个对iphone的下单请求了。

一旦对某个数据做了分段处理之后，有一个坑大家一定要注意：==就是如果某个下单请求，咔嚓加锁，然后发现这个分段库存里的库存不足了，此时咋办？==

这时你得自动释放锁，然后立马换下一个分段库存，再次尝试加锁后尝试处理。这个过程一定要实现。

Redis 3种过期策略？

3种策略各自的特点

• 定时删除
  • 含义：在设置key的过期时间的同时，为该key创建一个定时器，让定时器在key的过期时间来临时，对key进行删除
  • 优点：保证内存被尽快释放
  • 缺点：
    • 若过期key很多，删除这些key会占用很多的CPU时间，在CPU时间紧张的情况下，CPU不能把所有的时间用来做要紧的事儿，还需要去花时间删除这些key
    • 定时器的创建耗时，若为每一个设置过期时间的key创建一个定时器（将会有大量的定时器产生），性能影响严重
    • 没人用
• 惰性删除
  • 含义：key过期的时候不删除，每次从数据库获取key的时候去检查是否过期，若过期，则删除，返回null。
  • 优点：删除操作只发生在从数据库取出key的时候发生，而且只删除当前key，所以对CPU时间的占用是比较少的，而且此时的删除是已经到了非做不可的地步（如果此时还不删除的话，我们就会获取到了已经过期的key了）
  • 缺点：若大量的key在超出超时时间后，很久一段时间内，都没有被获取过，那么可能发生内存泄露（无用的垃圾占用了大量的内存）
• 定期删除
  • 含义：每隔一段时间执行一次删除(在redis.conf配置文件设置hz，1s刷新的频率)过期key操作
  • 优点：
    • 通过限制删除操作的时长和频率，来减少删除操作对CPU时间的占用–处理”定时删除”的缺点
    • 定期删除过期key–处理”惰性删除”的缺点
  • 缺点
    • 在内存友好方面，不如”定时删除”
    • 在CPU时间友好方面，不如”惰性删除”
  • 难点
    • 合理设置删除操作的执行时长（每次删除执行多长时间）和执行频率（每隔多长时间做一次删除）（这个要根据服务器运行情况来定了）

看完上面三种策略后可以得出以下结论：
定时删除和定期删除为主动删除：Redis会定期主动淘汰一批已过去的key

惰性删除为被动删除：用到的时候才会去检验key是不是已过期，过期就删除

惰性删除为redis服务器内置策略

定期删除可以通过：

• 第一、配置redis.conf 的hz选项，默认为10 （即1秒执行10次，100ms一次，值越大说明刷新频率越快，最Redis性能损耗也越大）
• 第二、配置redis.conf的maxmemory最大值，当已用内存超过maxmemory限定时，就会触发主动清理策略

Redis采用的过期策略

惰性删除+定期删除

• 惰性删除流程
  • 在进行get或setnx等操作时，先检查key是否过期，
  • 若过期，删除key，然后执行相应操作；
  • 若没过期，直接执行相应操作
• 定期删除流程（简单而言，对指定个数个库的每一个库随机删除小于等于指定个数个过期key）
  • 遍历每个数据库（就是redis.conf中配置的”database”数量，默认为16）
  • 检查当前库中的指定个数个key（默认是每个库随机检查20个key，注意相当于该循环执行20次，循环体时下边的描述）
  • 如果当前库中没有一个key设置了过期时间，直接执行下一个库的遍历
  • 随机获取一个设置了过期时间的key，检查该key是否过期，如果过期，删除key
  • 判断定期删除操作是否已经达到指定时长，若已经达到，直接退出定期删除。

==最后就是如果的如果，定期没删，我也没查询，那可咋整？==

Redis 的淘汰策略有哪些？

我们通过配置redis.conf中的maxmemory这个值来开启内存淘汰功能。根据用户设置的逐出策略，选出待逐出的key，直到当前内存小于最大内存值为止。

Redis 有六种淘汰策略

策略	描述
volatile-lru	从已设置过期时间的 KV 集中优先对最近最少使用(less recently used)的数据淘汰
volitile-ttl	从已设置过期时间的 KV 集中优先对剩余时间短(time to live)的数据淘汰
volitile-random	从已设置过期时间的 KV 集中随机选择数据淘汰
allkeys-lru	从所有 KV 集中优先对最近最少使用(less recently used)的数据淘汰
allKeys-random	从所有 KV 集中随机选择数据淘汰
noeviction	不淘汰策略，若超过最大内存，返回错误信息

4.0 版本后增加以下两种

• volatile-lfu：从已设置过期时间的数据集(server.db[i].expires)中挑选最不经常使用的数据淘汰
• allkeys-lfu：当内存不足以容纳新写入数据时，在键空间中，移除最不经常使用的 key。

手写一个LRU算法？

public class LRUCache<K, V> extends LinkedHashMap<K, V> {
        private int maxEntries;

        public LRUCache(int maxEntries){
        // 这块就是设置一个hashmap的初始大小，同时最后一个true指的是让linkedhashmap按照访问顺序来进行排序，最近访问的放在头，最老访问的就在尾
            super(16, 0.75f, true);
            this.maxEntries = maxEntries;
        }

        @Override
        protected boolean removeEldestEntry(Entry<K, V> eldest) {
        // 这个意思就是说当map中的数据量大于指定的缓存个数的时候，就自动删除最老的数据
            return size() > maxEntries;
        }
    }   

Redis常见性能问题和解决方案？

1. Master 最好不要做任何持久化工作，包括内存快照和 AOF 日志文件，特别是不要启用内存快照做持久化。

2. 如果数据比较关键，某个 Slave 开启 AOF 备份数据，策略为每秒同步一次。

3. Redis Sentinal着眼于高可用，在master宕机时会自动将slave提升为master，继续提供服务。Redis Cluster着眼于扩展性，在单个redis内存不足时，使用Cluster进行分片存储。

那他是单线程的，我们现在服务器都是多核的，那不是很浪费？

是的他是单线程的，但是，我们可以通过在单机开多个Redis实例嘛。

---

Redis高可用

Redis的同步机制了解么？

全量同步(新增salve结点)

Redis可以使用主从同步，从从同步。第一次同步时，主节点做一次bgsave，并同时将后续修改操作记录(复制偏移量)到内存buffer(复制积压缓冲区)，待完成后将RDB文件全量同步到复制节点，复制节点接受完成后将RDB镜像加载到内存。加载完成后，再通知主节点将期间修改的操作记录同步到复制节点进行重放就完成了同步过程。后续的增量数据通过AOF日志同步即可，有点类似数据库的binlog。

部分重新同步(断线重连)

部分重同步可以让我们断线后重连只需要同步缺失的数据，部分重同步功能由以下部分组成：

• 主从服务器的复制偏移量
• 主服务器的复制积压缓冲区
• 服务器运行的ID(run ID)

首先我们来解释一下上面的名词：

复制偏移量：执行复制的双方都会分别维护一个复制偏移量

• 主服务器每次传播N个字节，就将自己的复制偏移量加上N
• 从服务器每次收到主服务器的N个字节，就将自己的复制偏移量加上N

通过对比主从复制的偏移量，就很容易知道主从服务器的数据是否处于一致性的状态！

那断线重连以后，从服务器向主服务器发送PSYNC命令，报告现在的偏移量是36，那么主服务器该对从服务器执行完整重同步还是部分重同步呢？？这就交由复制积压缓冲区来决定。

当主服务器进行命令传播时，不仅仅会将写命令发送给所有的从服务器，还会将写命令入队到复制积压缓冲区里面(这个大小可以调的)。如果复制积压缓冲区存在丢失的偏移量的数据，那就执行部分重同步，否则执行完整重同步。

服务器运行的ID(run ID)实际上就是用来比对ID是否相同。如果不相同，则说明从服务器断线之前复制的主服务器和当前连接的主服务器是两台服务器，这就会进行完整重同步。

Redis高可用配置（主从、哨兵、集群）

详见如下链接：

https://github.com/Snailclimb/JavaGuide/blob/master/docs/database/Redis/redis-collection/Redis(9)%E2%80%94%E2%80%94%E9%9B%86%E7%BE%A4%E5%85%A5%E9%97%A8%E5%AE%9E%E8%B7%B5%E6%95%99%E7%A8%8B.md

介绍了Redis主从、哨兵、集群如何实践，及其原理。以下仅从面试热点进行梳理：

哨兵

在 哨兵系统 中，节点分为 数据节点 和 哨兵节点：前者存储数据，后者实现额外的控制功能。在 集群 中，没有数据节点与非数据节点之分：所有的节点都存储数据，也都参与集群状态的维护。为此，集群中的每个节点，都提供了两个 TCP 端口：

• 普通端口： 即我们在前面指定的端口 (7000等)。普通端口主要用于为客户端提供服务 （与单机节点类似）；但在节点间数据迁移时也会使用。
• 集群端口： 端口号是普通端口 + 10000 （10000是固定值，无法改变），如 7000 节点的集群端口为 17000。集群端口只用于节点之间的通信，如搭建集群、增减节点、故障转移等操作时节点间的通信；不要使用客户端连接集群接口。为了保证集群可以正常工作，在配置防火墙时，要同时开启普通端口和集群端口。

Redis 集群

上图 展示了 Redis Cluster 典型的架构图，集群中的每一个 Redis 节点都 互相两两相连，客户端任意 直连 到集群中的 任意一台，就可以对其他 Redis 节点进行 读写 的操作。

基本原理

Redis 集群中内置了 16384 (162^10^)个哈希槽。当客户端连接到 Redis 集群之后，会同时得到一份关于这个 *集群的配置信息，当客户端具体对某一个 key 值进行操作时，会计算出它的一个 Hash 值，然后把结果对 16384 **求余数，这样每个 key 都会对应一个编号在 0-16383 之间的哈希槽，Redis 会根据节点数量大致均等的将哈希槽映射到不同的节点。

集群的主要作用

1. 数据分区： 数据分区 (或称数据分片) 是集群最核心的功能。集群将数据分散到多个节点，一方面 突破了 Redis 单机内存大小的限制，存储容量大大增加；另一方面 每个主节点都可以对外提供读服务和写服务，大大提高了集群的响应能力。Redis 单机内存大小受限问题，在介绍持久化和主从复制时都有提及，例如，如果单机内存太大，bgsave 和 bgrewriteaof 的 fork 操作可能导致主进程阻塞，主从环境下主机切换时可能导致从节点长时间无法提供服务，全量复制阶段主节点的复制缓冲区可能溢出……
2. 高可用： 集群支持主从复制和主节点的 自动故障转移 （与哨兵类似），当任一节点发生故障时，集群仍然可以对外提供服务。

数据分区方案简析

方案一：哈希值 % 节点数

哈希取余分区思路非常简单：计算 key 的 hash 值，然后对节点数量进行取余，从而决定数据映射到哪个节点上。

不过该方案最大的问题是，当新增或删减节点时，节点数量发生变化，系统中所有的数据都需要 重新计算映射关系，引发大规模数据迁移。

方案二：一致性哈希分区

一致性哈希算法将 整个哈希值空间 组织成一个虚拟的圆环，范围是 *[0 , 2^32^-1]*，对于每一个数据，根据 key 计算 hash 值，确数据在环上的位置，然后从此位置沿顺时针行走，找到的第一台服务器就是其应该映射到的服务器：

与哈希取余分区相比，一致性哈希分区将 增减节点的影响限制在相邻节点。以上图为例，如果在 node1 和 node2 之间增加 node5，则只有 node2 中的一部分数据会迁移到 node5；如果去掉 node2，则原 node2 中的数据只会迁移到 node4 中，只有 node4 会受影响。

一致性哈希分区的主要问题在于，当 节点数量较少 时，增加或删减节点，对单个节点的影响可能很大，造成数据的严重不平衡。还是以上图为例，如果去掉 node2，node4 中的数据由总数据的 1/4 左右变为 1/2 左右，与其他节点相比负载过高。

方案三：带有虚拟节点的一致性哈希分区

该方案在 一致性哈希分区的基础上，引入了 虚拟节点 的概念。==Redis 集群使用的便是该方案==，其中的虚拟节点称为 槽（slot）。槽是介于数据和实际节点之间的虚拟概念，每个实际节点包含一定数量的槽，每个槽包含哈希值在一定范围内的数据。

在使用了槽的一致性哈希分区中，槽是数据管理和迁移的基本单位。槽 解耦 了 数据和实际节点 之间的关系，增加或删除节点对系统的影响很小。仍以上图为例，系统中有 4 个实际节点，假设为其分配 16 个槽(0-15)；

• 槽 0-3 位于 node1；4-7 位于 node2；以此类推….

如果此时删除 node2，只需要将槽 4-7 重新分配即可，例如槽 4-5 分配给 node1，槽 6 分配给 node3，槽 7 分配给 node4；可以看出删除 node2 后，数据在其他节点的分布仍然较为均衡。

节点通信机制简析

集群的建立离不开节点之间的通信，例如我们上访在 快速体验 中刚启动六个集群节点之后通过 redis-cli 命令帮助我们搭建起来了集群，实际上背后每个集群之间的两两连接是通过了 CLUSTER MEET 命令发送 MEET 消息完成的，下面我们展开详细说说。

数据结构简析

节点需要专门的数据结构来存储集群的状态。所谓集群的状态，是一个比较大的概念，包括：集群是否处于上线状态、集群中有哪些节点、节点是否可达、节点的主从状态、槽的分布……

节点为了存储集群状态而提供的数据结构中，最关键的是 clusterNode 和 clusterState 结构：前者记录了一个节点的状态，后者记录了集群作为一个整体的状态。

clusterNode 结构

clusterNode 结构保存了 一个节点的当前状态，包括创建时间、节点 id、ip 和端口号等。每个节点都会用一个 clusterNode 结构记录自己的状态，并为集群内所有其他节点都创建一个 clusterNode 结构来记录节点状态。

下面列举了 clusterNode 的部分字段，并说明了字段的含义和作用：

typedef struct clusterNode {
    //节点创建时间
    mstime_t ctime;
    //节点id
    char name[REDIS_CLUSTER_NAMELEN];
    //节点的ip和端口号
    char ip[REDIS_IP_STR_LEN];
    int port;
    //节点标识：整型，每个bit都代表了不同状态，如节点的主从状态、是否在线、是否在握手等
    int flags;
    //配置纪元：故障转移时起作用，类似于哨兵的配置纪元
    uint64_t configEpoch;
    //槽在该节点中的分布：占用16384/8个字节，16384个比特；每个比特对应一个槽：比特值为1，则该比特对应的槽在节点中；比特值为0，则该比特对应的槽不在节点中
    unsigned char slots[16384/8];
    //节点中槽的数量
    int numslots;
    …………
} clusterNode;

除了上述字段，clusterNode 还包含节点连接、主从复制、故障发现和转移需要的信息等。

clusterState 结构

clusterState 结构保存了在当前节点视角下，集群所处的状态。主要字段包括：

typedef struct clusterState {
    //自身节点
    clusterNode *myself;
    //配置纪元
    uint64_t currentEpoch;
    //集群状态：在线还是下线
    int state;
    //集群中至少包含一个槽的节点数量
    int size;
    //哈希表，节点名称->clusterNode节点指针
    dict *nodes;
    //槽分布信息：数组的每个元素都是一个指向clusterNode结构的指针；如果槽还没有分配给任何节点，则为NULL
    clusterNode *slots[16384];
    …………
} clusterState;

除此之外，clusterState 还包括故障转移、槽迁移等需要的信息。

假如Redis里面有1亿个key，其中有10w个key是以某个固定的已知的前缀开头的，如何将它们全部找出来？

使用 keys 指令可以扫出指定模式的 key 列表。但是要注意 keys 指令会导致线程阻塞一段时间，线上服务会停顿，直到指令执行完毕，服务才能恢复。这个时候可以使用 scan 指令，scan 指令可以无阻塞的提取出指定模式的 key 列表，但是会有一定的重复概率，在客户端做一次去重就可以了，但是整体所花费的时间会比直接用 keys 指令长。

字典是如何实现的？Rehash了解吗？

先总体聊一下 Redis 中的字典

字典是 Redis 服务器中出现最为频繁的复合型数据结构。除了 hash 结构的数据会用到字典外，整个 Redis 数据库的所有 key 和 value 也组成了一个 全局字典，还有带过期时间的 key 也是一个字典。(存储在 RedisDb 数据结构中)

说明字典内部结构和 rehash

字典结构内部包含 两个 hashtable，通常情况下只有一个 hashtable 有值，但是在字典扩容缩容时，需要分配新的 hashtable，然后进行 渐进式搬迁 (rehash)，这时候两个 hashtable 分别存储旧的和新的 hashtable，待搬迁结束后，旧的将被删除，新的 hashtable 取而代之。

渐进式哈希(rehashing)的机制来提高字典的缩放效率，避免 rehash 对服务器性能造成影响，渐进式 rehash 的好处在于它采取分而治之的方式， 将 rehash 键值对所需的计算工作均摊到对字典的每个添加、删除、查找和更新操作上， 从而避免了集中式 rehash 而带来的庞大计算量。

即 redis中，每次插入键值对时，都会检查是否需要扩容。如果满足扩容条件，则进行扩容。

扩缩容的条件

正常情况下，当 hash 表中 元素的个数超过了哈希表的长度时，就会开始扩容，扩容的新数组是 原数组大小的 2 倍。不过如果 Redis 正在做 bgsave(持久化命令)，为了减少内存也得过多分离，Redis 尽量不去扩容，但是如果 hash 表非常满了，达到了哈希表长度的 5 倍了，这个时候就会 强制扩容。

当 hash 表因为元素逐渐被删除变得越来越稀疏时，Redis 会对 hash 表进行缩容来减少 hash 表的第一维数组空间占用。所用的条件是 元素个数低于哈希表数组长度的 10%，缩容不会考虑 Redis 是否在做 bgsave。

---

数据结构篇

简述一下 Redis 常用数据结构及实现？

首先在 Redis 内部会使用一个 RedisObject 对象来表示所有的 key 和 value：

其次 Redis 为了 平衡空间和时间效率，针对 value 的具体类型在底层会采用不同的数据结构来实现，下图展示了他们之间的映射关系：


String：

​ 动态字符串，内部结构实现上类似于Java的ArrayList，纯 数字用long

List：

​ Redis 早期版本存储 list 列表数据结构使用的是压缩列表 ziplist 和普通的双向链表 linkedlist，也就是说当元素少时使用 ziplist，当元素多时用 linkedlist。但考虑到链表的附加空间相对较高，prev 和 next 指针就要占去 16 个字节（64 位操作系统占用 8 个字节），另外每个节点的内存都是单独分配，会加剧内存的碎片化，影响内存管理效率。

后来 Redis 新版本（3.2）对列表数据结构进行了改造，使用 quicklist 代替了 ziplist 和 linkedlist。

Hash：

​ HashMap。

Set：

​ Hashtable

ZSet：

​ zset的编码有ziplist和skiplist两种。
​ 底层分别使用ziplist（压缩链表）和skiplist（跳表）实现。

当zset满足以下两个条件的时候，使用ziplist：

1. 保存的元素少于128个
2. 保存的所有元素大小都小于64字节

​ ziplist 编码的有序集合对象使用压缩列表作为底层实现，每个集合元素使用两个紧挨在一起的压缩列表节点来保存，第一个节点保存元素的成员，第二个节点保存元素的分值。并且压缩列表内的集合元素按分值从小到大的顺序进行排列，小的放置在靠近表头的位置，大的放置在靠近表尾的位置。

​ skiplist底层实现使用了两个数据结构，hash+跳表，hash的作用就是关联元素value和权重score，保障元素value的唯一性，可以通过元素value找到相应的score值。跳跃列表的目的在于给元素 score 排序，根据score的范围获取元素列表。

GeoHash：

在使用 Redis 进行 Geo 查询 时，我们要时刻想到它的内部结构实际上只是一个 zset(skiplist)。通过 zset 的 score 排序就可以得到坐标附近的其他元素 (实际情况要复杂一些，不过这样理解足够了)，通过将 score 还原成坐标值就可以得到元素的原始坐标了。

当两个坐标元素的距离不是很远的时候，我们就可以简单利用 勾股定理 就能够得出他们之间的 距离。

HyperLogLog 有了解吗？

• Redis(4)——神奇的HyperLoglog解决统计问题

布隆过滤器有了解吗？

• Redis(5)——亿级数据过滤和布隆过滤器

跳跃表是如何实现的？原理？

“

这是 Redis 中比较重要的一个数据结构，建议阅读 之前写过的文章，里面详细介绍了原理和一些细节：

• Redis(2)——跳跃表

GeoHash 了解吗？

“

建议阅读 之前的系列文章：

• Redis(6)——GeoHash查找附近的人

压缩列表了解吗？

这是 Redis 为了节约内存 而使用的一种数据结构，zset 和 hash 容器对象会在元素个数较少的时候，采用压缩列表（ziplist）进行存储。压缩列表是 一块连续的内存空间，元素之间紧挨着存储，没有任何冗余空隙。

“

因为之前自己也没有学习过，所以找了一篇比较好比较容易理解的文章：

• 图解Redis之数据结构篇——压缩列表 - https://mp.weixin.qq.com/s/nba0FUEAVRs0vi24KUoyQg
• 这一篇稍微底层稍微硬核一点：http://www.web-lovers.com/redis-source-ziplist.html

快速列表 quicklist 了解吗？

Redis 早期版本存储 list 列表数据结构使用的是压缩列表 ziplist 和普通的双向链表 linkedlist，也就是说当元素少时使用 ziplist，当元素多时用 linkedlist。但考虑到链表的附加空间相对较高，prev 和 next 指针就要占去 16 个字节（64 位操作系统占用 8 个字节），另外每个节点的内存都是单独分配，会家具内存的碎片化，影响内存管理效率。

后来 Redis 新版本（3.2）对列表数据结构进行了改造，使用 quicklist 代替了 ziplist 和 linkedlist。

“

同上..建议阅读一下以下的文章：

• Redis列表list 底层原理 - https://zhuanlan.zhihu.com/p/102422311

Stream 结构有了解吗？

Redis Stream 从概念上来说，就像是一个 仅追加内容 的 消息链表，把所有加入的消息都一个一个串起来，每个消息都有一个唯一的 ID 和内容，这很简单，让它复杂的是从 Kafka 借鉴的另一种概念：消费者组(Consumer Group) (思路一致，实现不同)：

上图就展示了一个典型的 Stream 结构。每个 Stream 都有唯一的名称，它就是 Redis 的 key，在我们首次使用 xadd 指令追加消息时自动创建。我们对图中的一些概念做一下解释：

• Consumer Group：消费者组，可以简单看成记录流状态的一种数据结构。消费者既可以选择使用 XREAD 命令进行 独立消费，也可以多个消费者同时加入一个消费者组进行 组内消费。同一个消费者组内的消费者共享所有的 Stream 信息，同一条消息只会有一个消费者消费到，这样就可以应用在分布式的应用场景中来保证消息的唯一性。
• last_delivered_id：用来表示消费者组消费在 Stream 上 消费位置 的游标信息。每个消费者组都有一个 Stream 内 唯一的名称，消费者组不会自动创建，需要使用 XGROUP CREATE 指令来显式创建，并且需要指定从哪一个消息 ID 开始消费，用来初始化 last_delivered_id 这个变量。
• pending_ids：每个消费者内部都有的一个状态变量，用来表示 已经 被客户端 获取，但是 还没有 ack 的消息。记录的目的是为了 保证客户端至少消费了消息一次，而不会在网络传输的中途丢失而没有对消息进行处理。如果客户端没有 ack，那么这个变量里面的消息 ID 就会越来越多，一旦某个消息被 ack，它就会对应开始减少。这个变量也被 Redis 官方称为 PEL (Pending Entries List)。

Stream 消息太多怎么办？

很容易想到，要是消息积累太多，Stream 的链表岂不是很长，内容会不会爆掉就是个问题了。xdel 指令又不会删除消息，它只是给消息做了个标志位。

Redis 自然考虑到了这一点，所以它提供了一个定长 Stream 功能。在 xadd 的指令提供一个定长长度 maxlen，就可以将老的消息干掉，确保最多不超过指定长度，使用起来也很简单：

> XADD mystream MAXLEN 2 * value 1
1526654998691-0
> XADD mystream MAXLEN 2 * value 2
1526654999635-0
> XADD mystream MAXLEN 2 * value 3
1526655000369-0
> XLEN mystream
(integer) 2
> XRANGE mystream - +
1) 1) 1526654999635-0
   2) 1) "value"
      2) "2"
2) 1) 1526655000369-0
   2) 1) "value"
      2) "3"

如果使用 MAXLEN 选项，当 Stream 的达到指定长度后，老的消息会自动被淘汰掉，因此 Stream 的大小是恒定的。目前还没有选项让 Stream 只保留给定数量的条目，因为为了一致地运行，这样的命令必须在很长一段时间内阻塞以淘汰消息。(例如在添加数据的高峰期间，你不得不长暂停来淘汰旧消息和添加新的消息)

另外使用 MAXLEN 选项的花销是很大的，Stream 为了节省内存空间，采用了一种特殊的结构表示，而这种结构的调整是需要额外的花销的。所以我们可以使用一种带有 ~ 的特殊命令：

XADD mystream MAXLEN ~ 1000 * ... entry fields here ...

它会基于当前的结构合理地对节点执行裁剪，来保证至少会有 1000 条数据，可能是 1010 也可能是 1030。

PEL 是如何避免消息丢失的？

在客户端消费者读取 Stream 消息时，Redis 服务器将消息回复给客户端的过程中，客户端突然断开了连接，消息就丢失了。但是 PEL 里已经保存了发出去的消息 ID，待客户端重新连上之后，可以再次收到 PEL 中的消息 ID 列表。不过此时 xreadgroup 的起始消息 ID 不能为参数 > ，而必须是任意有效的消息 ID，一般将参数设为 0-0，表示读取所有的 PEL 消息以及自 last_delivered_id 之后的新消息。

和 Kafka 对比起来呢？

Redis 基于内存存储，这意味着它会比基于磁盘的 Kafka 快上一些，也意味着使用 Redis 我们 不能长时间存储大量数据。不过如果您想以 最小延迟 实时处理消息的话，您可以考虑 Redis，但是如果 消息很大并且应该重用数据 的话，则应该首先考虑使用 Kafka。

另外从某些角度来说，Redis Stream 也更适用于小型、廉价的应用程序，因为 Kafka 相对来说更难配置一些。

“

推荐阅读 之前的系列文章，里面 也对 Pub/ Sub 做了详细的描述：

• Redis(8)——发布/订阅与Stream