4）深度解密 Redis 的列表（List）

楔子

下面来解密 Redis 的列表，整篇文章分为三个部分。

• Redis 列表的相关命令；
  
• Redis 列表的应用场景；
  
• Redis 列表的实现原理；
  

Redis 列表的相关命令

先来看看列表的相关命令，我们首先要学会如何使用它。

lpush key value1 value2 ...：将多个值设置到列表里面，从左边 push

rpush key value1 value2 ...：将多个值设置到列表里面，从右边 push

# 返回插入成功之后，列表的元素个数
# 因为是从左往右 push
# 所以此时列表内的元素是 koishi mashiro
127.0.0.1:6379> lpush girls mashiro koishi
(integer) 2  
# 从右往左 push
127.0.0.1:6379> rpush girls satori
(integer) 3
127.0.0.1:6379>

lrange key start end：遍历列表，索引从 0 开始，最后一个为 -1，且包含两端

127.0.0.1:6379> lrange girls 0 -1
1) "koishi"
2) "mashiro"
3) "satori"
127.0.0.1:6379> lrange girls 0 2
1) "koishi"
2) "mashiro"
3) "satori"
127.0.0.1:6379> lrange girls 0 1
1) "koishi"
2) "mashiro"
# 对不存在的列表使用 lrange，会得到空数组
127.0.0.1:6379> lrange lst 0 -1
(empty array)

lpop key：从列表的左端弹出一个值，列表长度改变

rpop key：从列表的右端弹出一个值，列表长度改变

127.0.0.1:6379> lpop girls
"koishi"
127.0.0.1:6379> rpop girls
"satori"
127.0.0.1:6379> lrange girls 0 -1
1) "mashiro"
127.0.0.1:6379>

lindex key index：获取指定索引位置的元素，列表长度不变

127.0.0.1:6379> lindex girls 0
"mashiro"
127.0.0.1:6379> lrange girls 0 -1
1) "mashiro"
# 对不存在的列表使用 lindex，会得到 nil
127.0.0.1:6379> lindex lst 0 
(nil)  
127.0.0.1:6379>

llen key：获取指定列表的长度

127.0.0.1:6379> llen girls
(integer) 1
# 对不存在的列表使用 llen，会得到 0
127.0.0.1:6379> llen lst
(integer) 0  
127.0.0.1:6379>

lrem key count value：删除 count 个 value，如果 count 为 0，那么将全部删除

127.0.0.1:6379> lpush lst 1 1 1 1
(integer) 4
# 删除 3 个 1
127.0.0.1:6379> lrem lst 3 1
(integer) 3  
127.0.0.1:6379> lrange lst 0 -1
1) "1"
127.0.0.1:6379>

ltrim key start end：从 start 截取到 end，再重新赋值给 key

127.0.0.1:6379> rpush lst 2 3 4 5
(integer) 5
127.0.0.1:6379> lrange lst 0 -1
1) "1"
2) "2"
3) "3"
4) "4"
5) "5"
# 将 5 重新赋值给 lst
127.0.0.1:6379> ltrim lst 4 -1
OK  
127.0.0.1:6379> lrange lst 0 -1
1) "5"
127.0.0.1:6379>

rpoplpush key1 key2：移除 key1 的最后一个元素，并添加到 key2 的开头

127.0.0.1:6379> rpush lst1 1 2 3
(integer) 3
127.0.0.1:6379> rpush lst2 11 22 33
(integer) 3
127.0.0.1:6379> rpoplpush lst1 lst2
"3"
127.0.0.1:6379> lrange lst2 0 -1
1) "3"
2) "11"
3) "22"
4) "33"
127.0.0.1:6379>

lset key index value：将 key 中索引为 index 的元素设置为 value

127.0.0.1:6379> lrange lst2 0 -1
1) "3"
2) "11"
3) "22"
4) "33"
127.0.0.1:6379> lset lst2 1 2333
OK
127.0.0.1:6379> lrange lst2 0 -1
1) "3"
2) "2333"
3) "22"
4) "33"
# 索引越界则报错，显然索引为 10 越界了
127.0.0.1:6379> lset lst2 10 2333
(error) ERR index out of range  
127.0.0.1:6379>

linsert key before/after value1 value2：在首次出现的 value1 的前面或者后面插入一个 value2。如果 value1 不存在，则该语句无效

127.0.0.1:6379> rpush lst3 1 2 2 3
(integer) 4
127.0.0.1:6379> linsert lst3 before 2 666
(integer) 5
127.0.0.1:6379> lrange lst3 0 -1
1) "1"
2) "666"
3) "2"
4) "2"
5) "3"
127.0.0.1:6379> linsert lst3 after 2 2333
(integer) 6
127.0.0.1:6379> lrange lst3 0 -1
1) "1"
2) "666"
3) "2"
4) "2333"
5) "2"
6) "3"
127.0.0.1:6379>

以上就是 Redis 列表的一些基础操作，其实还有两个 API 我们没有说，先卖个关子。

Redis 列表的应用场景

列表的典型使用场景主要有两个。

1）文章列表

对于博客站点来说，当用户和文章都越来越多时，为了加快程序的响应速度，我们可以把用户自己的文章存入到 List 中。因为 List 是有序的结构，所以这样还可以完美地实现分页功能，从而加速程序的响应速度。

当然这里以博客站点为例，还有很多其它与之类似的场景。

2）消息队列

在消息队列方面，虽然 Redis 的专业程度不如 RabbitMQ，网络或者系统出故障时可能会丢数据。但由于它轻量、使用方便，所以如果你的业务场景比较简单，那么完全可以考虑使用 Redis 作为消息队列。

而把 Redis 当作队列来使用，肯定最先想到的就是使用 List 这个数据类型。因为 List 底层的实现是一个链表（一会说），在头部和尾部操作元素，时间复杂度都是 O(1)，这意味着它非常符合消息队列的模型。

生产者使用 LPUSH 发布消息：

# 从左往右 push，所以是 n3 n2 n1
127.0.0.1:6379> LPUSH items n1 n2 n3
(integer) 3
127.0.0.1:6379>

消费者使用 RPOP 拉取消息：

127.0.0.1:6379> RPOP items
"n1"
127.0.0.1:6379> RPOP items
"n2"
127.0.0.1:6379> RPOP items
"n3"

这个模型非常简单，也很容易理解。

但这里有个小问题，当队列中已经没有消息了，消费者在执行 RPOP 时，会返回 NULL。而我们在编写消费者逻辑时，一般是一个「死循环」，这个逻辑需要不断地从队列中拉取消息进行处理，伪代码一般会这么写：

while True:
    msg = client.rpop("items")
    if msg is None:
        continue
    handle(msg)

如果此时队列为空，那消费者依旧会频繁拉取消息，这会造成「CPU 空转」，不仅浪费 CPU 资源，还会对 Redis 造成压力。怎么解决这个问题呢？最容易想到的一个解决办法是，当队列为空时，我们可以「休眠」一会，再去尝试拉取消息。

这就解决了 CPU 空转问题，但又带来另外一个问题：当消费者在休眠等待时，有新消息来了，那消费者处理新消息就会存在「延迟」。如果缩短这个延迟，只能减小休眠的时间，但休眠时间变小，又有可能引发 CPU 空转问题。那如何做，既能及时处理新消息，还能避免 CPU 空转呢？

Redis 是否存在这样一种机制：如果队列为空，消费者在拉取消息时就「阻塞等待」，一旦有新消息过来，就通知消费者立即处理新消息呢？幸运的是，Redis 确实提供了「阻塞式」拉取消息的命令：BRPOP / BLPOP，这里的 B 指的是阻塞（Block）。

现在可以这样来拉取消息了：

while True:
    msg = client.brpop("items")
    if msg is None:
        continue
    handle(msg)

使用 BRPOP 这种阻塞式方式拉取消息时，还支持传入一个「超时时间」，如果设置为 0，则表示不设置超时，直到有新消息才返回，否则会在指定的超时时间后返回 NULL。所以这个方案不错，既兼顾了效率，还避免了 CPU 空转问题，一举两得。

注意：如果设置的超时时间太长，这个连接太久没有活跃过，可能会被 Redis Server 判定为无效连接，之后 Redis Server 会强制把这个客户端踢下线。所以采用这种方案，客户端要有重连机制。

解决了消息处理不及时的问题，我们可以再思考一下，这种队列模型有什么缺点呢？显然缺点非常明显且致命：

• 不支持重复消费：消费者拉取消息后，这条消息就从 List 中删除了，无法被其它消费者再次消费，即不支持多个消费者消费同一批数据；
  
• 消息丢失：消费者拉取到消息后，如果发生异常宕机，那这条消息就丢失了；
  

第一个问题是功能上的，使用 List 做消息队列，它仅仅支持最简单的，一组生产者对应一组消费者，不能满足多组生产者和消费者的业务场景。

第二个问题就比较棘手了，因为从 List 中 POP 一条消息出来后，这条消息就会立即从链表中删除。也就是说，无论消费者是否处理成功，这条消息都没办法再次消费了，这也意味着，如果消费者在处理消息时异常宕机，那这条消息就相当于丢失了。

所以 List 可以用在消息队列上面，但它只能适用于简单的业务场景，或者对数据丢失不那么敏感的业务场景。

比如我前一段时间写了一个客户投诉的后台，运营和客户之间基于 WebSocket + XMPP 通信。客户使用的是 APP，发送的消息需要先通过 XMPP 协议发送给后台的一个组件，组件将消息推送到队列。我编写的后台服务再从队列里面取消息，通过 WebSocket 发送给前端，最终展示在页面上，让运营查看。

而这个队列就是使用 Redis 的 List 实现的，当初我想引入 RabbitMQ，后来经过商量之后其他人觉得没必要，认为投诉消息丢了就丢了，这玩意也不太重要，丢了就让客户重发一遍😂。不过到目前为止，还没有遇见过丢失消息的情况，服务运行的还是很稳定的。

Redis 列表的实现原理

Redis 提供了一个命令，可以让我们查看某个类型使用的底层数据结构。

127.0.0.1:6379> lpush scores 97 98
(integer) 2
# 类型是 List
127.0.0.1:6379> type scores
list
# 使用的底层结构是 quicklist
127.0.0.1:6379> object encoding scores
"quicklist"
127.0.0.1:6379>

我们看到 List 类型的底层数据结构是 quicklist，这是 Redis 在 3.2 版本时引入的数据结构。早期的 List 类型是使用 ziplist（压缩列表）和双向链表实现的，Redis3.2 的时候改为 quicklist，下面就来看一下这几个结构的具体实现。

双向链表

由于 C 语言本身没有链表这个数据结构，所以 Redis 自己设计了一个链表数据结构。

// src/adlist.h
typedef struct listNode {
    // 前继节点
    struct listNode *prev;
    // 后继节点
    struct listNode *next;
    // 节点的值
    void *value;
} listNode;

有了前继节点和后继节点，可以看出节点之间会形成一个双向链表。

有了 ListNode 之后，Redis 在其基础之上又封装了一层，这样操作起来会更加方便。

// src/adlist.h
typedef struct list {
    // 链表头节点
    listNode *head;
    // 链表尾节点
    listNode *tail;
    // 节点值复制函数
    void *(*dup)(void *ptr);
    // 节点值释放函数
    void (*free)(void *ptr);
    // 节点值比较函数
    int (*match)(void *ptr, void *key);
    // 链表中节点的数量
    unsigned long len;
} list;

Redis 封装了一个数据结构叫 list ，该结构提供了链表头节点 head、链表尾节点 tail、链表节点数量 len、以及可以自定义实现的 dup、free、match 函数。

• dup 函数用于复制链表节点所保存的值；
  
• free 函数用于释放链表节点所保存的值；
  
• match 函数用于对比链表节点所保存的值和另一个输入值是否相等；
  
  

举个例子，下图是由 list 和 3 个 ListNode 组成的双向链表。

结构还是比较简单和清晰的，毕竟链表算是最常见的数据结构之一了。那么问题来了，双向链表它的优缺点是什么呢？

优点：

1）listNode 链表节点的结构里带有 prev 和 next 指针，获取某个节点的前置节点或后置节点的时间复杂度只需 O(1)；而且这两个指针都可以指向 NULL，所以对链表的访问以 NULL 为终点，是无环链表。

2）由于 list 内部包含了头节点 head 和尾节点 tail，所以获取链表的头节点和尾节点的时间复杂度只需 O(1)。

3）list 内部的成员变量 len 会维护持有的链表节点个数，所以获取链表中节点数量的时间复杂度为 O(1)。

4）链表节点使用 void* 指针来保存节点值，并且可以通过 list 结构的 dup, free, match 三个属性为节点值设置类型特定函数，所以链表可以用于保存各种不同类型的值（的指针）。

缺点：

1）链表和数组不同，链表的每个节点之间的内存都是不连续的，这意味着链表无法像数组那样很好地利用 CPU 缓存。

2）每一个链表节点除了保存值之外，还包含了 prev 和 next 两个指针，因此会有额外的内存开销。

在 Redis 3.2 之前，List 会使用双向链表作为底层数据结构的实现，但如果 List 对象的数据量比较少，那么会采用压缩列表（ziplist）来实现，它的优势是节省内存空间，并且是内存紧凑型的数据结构。

下面来看一下压缩列表。

压缩列表（ziplist）

压缩列表的特点是节省内存，它是由连续内存块组成的顺序型数据结构，有点类似于数组。

• zlbytes：记录整个压缩列表占用的内存字节数，该字段占 4 个字节；
  
• zltail：记录压缩列表「尾部」节点距离起始地址有多少字节，也就是列表尾部的偏移量；
  
• zllen：记录压缩列表包含的节点数量；
  
• zlend：标记压缩列表的结束点，固定值 0xFF；
  
  

在压缩列表中，如果我们要查找第一个元素和最后一个元素，可以通过表头三个字段的长度直接定位，复杂度是 O(1)。而查找其它元素时，就没有这么高效了，只能逐个查找，此时的复杂度就是 O(N) 级别，因此压缩列表不适合保存过多的元素。

然后压缩列表的每一个节点叫做一个 entry，是一个结构体，其内部字段如下：

压缩列表节点包含三部分内容：

• prevlen：记录了前一个节点的长度。如果前一个节点的长度小于 254 字节，那么 prevlen 字段需要用 1 字节的空间来保存这个长度值；如果前一个节点的长度大于等于 254 字节，那么 prevlen 字段需要用 5 字节的空间来保存这个长度值。
  
• encoding：记录了当前节点实际数据的类型以及长度。如果当前节点的数据是整数，encoding 会使用 1 字节的空间进行编码；如果当前节点的数据是字符串，根据字符串的长度大小，encoding 会使用 1 / 2 / 5 字节的空间进行编码。
  
• data：记录了当前节点的实际数据；
  
  

当我们往压缩列表中插入数据时，压缩列表就会根据数据是字符串还是整数，以及数据的大小，选择不同空间大小的 prevlen 和 encoding。我们看到这种做法类似于 SDS，目的就是将元数据所占的内存降到最低。

然后我们再来仔细观察一下压缩列表的结构，它除了查询元素没那么高效之外，还有没有别的问题呢？

假设一个压缩列表中有多个连续的、长度在 250～253 之间的节点，如下图：

因为这些节点长度值小于 254 字节，所以 prevlen 字段需要用 1 字节的空间来保存这个长度值。这时如果将一个长度大于等于 254 字节的新节点加入到压缩列表的头部，即新节点将成为 entry1 的前继节点，如下图所示：

因为 entry1 节点的 prevlen 字段只有 1 个字节大小，无法保存新节点的长度，此时就需要对压缩列表的空间进行重分配操作。将 entry1 节点的 prevlen 字段从原来的 1 字节大小扩展为 5 字节大小，因此 entry1 节点的大小相比之前会增加 4 字节。

而 entry1 一旦增加，那么 entry1 也会大于等于 254 字节，所以此时就要扩展 entry2 的 prevlen 字段。而一旦扩展 entry2 的 prevlen 字段，那么会有什么结果相信你已经猜到了，就像多米诺骨牌一样，连锁效应一发不可收拾。

空间扩展就意味着重新分配内存，所以一旦出现「连锁更新」，就会导致压缩列表占用的内存空间被多次重新分配，这会直接影响到压缩列表的访问性能。

因此虽然压缩列表紧凑型的内存布局能节省内存开销，但如果保存的元素数量增加了，或是元素变大了，会导致内存重新分配，最糟糕的是会有「连锁更新」的问题。因此压缩列表只用于保存节点数量不多的场景，如果节点数量足够少，即使发生连锁更新也是能接受的。

quicklist

总之从效果上看，压缩列表和双向链表都不尽人意，所以 Redis 在 3.2 版本设计了新的数据结构 quicklist，并将 List 对象的底层数据结构改由 quicklist 实现。而 quicklist 就相当于将双向链表和压缩列表结合起来了。

另外多提一句，压缩列表不仅可以作为 List 的底层结构，还可以作为 Hash 和 ZSet 的底层结构。但我们说压缩列表性能不行，于是在 Redis5.0 又设计出了新的数据结构 listpack，它不仅沿用了压缩列表紧凑型的内存布局，还提升了性能。在 Redis6.? 版本，将 Hash 和 Zset 的底层数据结构实现之一的压缩列表，替换成了 listpack。

• 3.2 版本之前，压缩列表同时作为 List、Hash、ZSet 的底层结构实现之一；
  
• 3.2 版本开始，List 类型引入了 quicklist，替换掉了压缩列表，但压缩列表仍是 Hash、ZSet 的底层结构；
  
• 5.0 版本开始，引入了新的数据结构 listpack；
  
• 6.? 版本开始，将 Hash、ZSet 的底层实现之一的压缩列表替换为 listpack；
  
  

总之 quicklist 和 listpack 的设计目标，就是尽可能地保持压缩列表节省内存的优势，同时解决连锁更新的问题。

下面看一下 quicklist，而 listpack 我们后续再说。

// src/quicklist.h
typedef struct quicklist {
    // quicklist 的链表头
    quicklistNode *head;      
    // quicklist 的链表尾
    quicklistNode *tail; 
    // 所有压缩列表中的总元素个数
    unsigned long count;
    // quicklistNodes 的个数
    unsigned long len;
    // LZF 压缩算法深度
    unsigned long compress : 16;       
    // ...
} quicklist;

接下来看看，quicklistNode 的结构定义：

// src/quicklist.h
typedef struct quicklistNode {
    // 前一个 quicklistNode
    struct quicklistNode *prev;     
    // 后一个 quicklistNode
    struct quicklistNode *next;     
    // quicklistNode 指向的压缩列表
    unsigned char *zl;              
    // 压缩列表的的字节大小
    unsigned int sz;                
    // 压缩列表的元素个数
    unsigned int count : 16;      
    // ...
} quicklistNode;

可以看到，quicklistNode 结构体里包含了前一个节点和后一个节点的指针，这样每个 quicklistNode 形成了一个双向链表。但是链表节点的元素不再是单纯保存元素值，而是保存了一个压缩列表，所以 quicklistNode 结构体里有个指向压缩列表的指针 *zl。

在向 quicklist 添加一个元素的时候，不会像普通的链表那样，直接新建一个链表节点。而是会检查插入位置的压缩列表是否能容纳该元素，如果能容纳就直接保存；如果不能容纳，则新建一个 quicklistNode 结构。

listpack

quicklist 虽然通过控制 quicklistNode 结构里的压缩列表的大小或者元素个数，来减少连锁更新带来的性能影响，但是并没有完全解决连锁更新的问题。因为 quicklistNode 还是使用了压缩列表来保存元素，压缩列表连锁更新的问题，来源于它的结构设计，所以要想彻底解决这个问题，需要设计一个新的数据结构。

于是 Redis 在 5.0 的时候新设计了一个数据结构叫 listpack，目的是替代压缩列表，它最大特点是 listpack 中每个节点不再包含前一个节点的长度了，压缩列表每个节点正因为需要保存前一个节点的长度字段，就会有连锁更新的隐患。

虽然 listpack 是在 5.0 的时候设计的，但在 6.2 之后才将 quicklist 内部的 ziplist 换成 listpack。

listpack 采用了压缩列表的很多优秀的设计，比如还是用一块连续的内存空间来紧凑地保存数据，并且为了节省内存的开销，listpack 节点会采用不同的编码方式保存不同大小的数据。

listpack 头包含两个属性，分别记录了 listpack 总字节数和元素数量，大小分别为 4 字节和 2 字节；然后 listpack 末尾也有个结尾标识，和压缩列表一样，值也为固定的 0xFF。而 listpack entry 就是 listpack 的节点了，每个 listpack 节点结构如下：

• encoding：定义该元素的编码类型，会对不同长度的整数和字符串进行编码；
  
• data：实际存放的数据；
  
• len：encoding + data 的总长度；
  
  

可以看到 listpack 的结构和压缩列表（ziplist）还是很相似的，只是没有记录前一个节点长度的字段了，listpack 只记录当前节点的长度。当我们向 listpack 加入一个新元素的时候，不会影响其它节点的长度字段的变化，从而避免了压缩列表的连锁更新问题。

小结

虽然 Redis 的 List 采用了 quicklist，但我们要理解这其中的变迁。首先是 ziplist，它的设计初衷就是「节省内存」，在存储数据时，把内存利用率发挥到了极致：

• 数字按「整型」编码存储，相比当成字符串存储，内存占用更少；
  
• 如果是字符串，会根据长度的大小，选择不同的编码；
  
• 对于极小的数据，干脆把内容直接放到了「长度」字段中（前几个位表示长度，后几个位存数据）；
  
  

但是缺点也很明显：

• 寻找元素只能挨个遍历，存储的数据过多，查询性能会变低；
  
• 每个元素保存了「上一个」元素的长度（为了方便反向遍历），这会导致当上一个元素内容发生修改，长度超过了原来的编码长度时，下一个元素的内容也要跟着变。而这个过程可能会恶性循环，发生连锁更新，导致不断地重新分配内存；
  
  

想要解决 ziplist 的问题，比较简单直接的方案就是，多个数据项不再用一个 ziplist 来存，而是分散到多个 ziplist 中，每个 ziplist 用指针串起来。这样修改其中一个数据项，即便发生连锁更新，也只会影响这一个 ziplist，其它 ziplist 不受影响，这种方案就是 quicklist。

所以 quicklist 相当于将 ziplist 和双向链表结合在了一起，它的 LPUSH, LPOP, RPUSH, RPOP 的时间复杂度都是 O(1)。

另外还可以设置 List 中每个 ziplist 节点可保存的元素个数 / 总大小，通过 list-max-ziplist-size 配置：

• 参数为正数：表示 ziplist 最多包含几个数据项；
  
• 参数为负数（-1 ~ -5）：表示每个 ziplist 存储最大的字节数，默认 -2，表示 8KB；
  
  

ziplist 超过上述任意一个配置，添加新元素就会新建 ziplist 插入到链表中。另外 List 因为更多是两头操作，那么为了节省内存，还可以把中间的 ziplist「压缩」，具体可看 list-compress-depth 配置项，默认配置不压缩。

虽然引入了 quicklist，但连锁更新的问题并没有得到解决，只是相应的范围变小了而已。要想彻底解决 ziplist 连锁更新的问题，本质上必须要修改 ziplist 的存储结构，也就是不要让每个元素保存「上一个」元素的长度即可。于是 Redis 又设计出了 listpack，对于那些使用了 ziplist 的数据数据结构，将其内部的 ziplist 换成 listpack。

ziplist 和 listpack 的结构非常相似，ziplist 之所以要保存上一个元素的长度（导致连锁更新的原因），主要是为了能够从后往前遍历。但 listpack 每个元素项不再保存上一个元素的长度，而是通过优化元素内字段的顺序，来保证既可以从前往后、也可以从后往前遍历，同时避免了连锁更新的问题。

本文参考自：

• 极客时间蒋德钧：《Redis 源码剖析与实战》
  
• 水滴与银弹：《把Redis当作队列来用，真的合适吗？》
  
• 小林 coding：《图解 Redis》
  
• 课代表 kaito 在极客时间《Redis 源码剖析与实战》评论区的精彩回答
  
• 微信读书：《Redis 设计与实现》