redis源码分析

持续更新…

一、C语言基础

本节参考资料：

1、C Primer Plus（第6版）中文版

2、stdint.h

3、编译器特性 attribute_((packed))

redis基于C语言编写，因此在阅读之前，先对C语言的相关语法进行简单回顾。

1 变量

1.1 整型

整型一般使用int定义，它占用一个机器字长，现在一般的个人计算机使用32位（4字节）来储存一个整型的数据。

除此之外，还有其它的附属关键字来定义整型：

  • short或short int，它占用16位（2字节）存储一个整型数据。
  • long或long int，它占用32位（4字节）或64位（8字节），具体占用多少取决于什么样的操作系统。
  • long long或long long int，它占用64位（8字节）存储整型数据。
  • 在上述关键字的基础上，使用unsigned可以定义无符号整型，如unsigned int、unsigned short等，它的占用和对应的关键字占用相同，但表示的范围不同（非负）。

1.2 字符型

字符型char实际上也是一个整型，它实际存储的是整数而不是字符。根据不同的字符编码，数字可以代表不同的字符。char占用8位（1字节）存储。

1.3 浮点型

C语言中的浮点类型有float（4字节），double（8字节），long double（16字节）三种形式。

1.4 指针类型

指针存储变量的地址，因此指针本身也是需要占用空间的，根据不同的操作系统架构，它占用32位（4字节）或者64位（8字节）。

下面的代码使用sizeof查看当前系统的不同类型大小是多少，其中%zd用于转换sizeof的返回类型：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
int a;
double b;
int * p1=&a;
double * p2=&b;
int main() {
    printf("char:%zd\n", sizeof(char));
    printf("int:%zd\n", sizeof(int));
    printf("short:%zd\n", sizeof(short));
    printf("long:%zd\n", sizeof(long));
    printf("long long:%zd\n", sizeof(long long));
    printf("float:%zd\n", sizeof(float));
    printf("double:%zd\n", sizeof(double));
    printf("long double:%zd\n", sizeof(long double));
    printf("int pointer:%zd\n", sizeof(p1));
    printf("double pointer:%zd\n", sizeof(p2));

    return 0;
}

在windows64位系统中，输出如下：

image-20220303085643366

注意到long和int同样占用4字节，这是为什么？

在C语言中，由于各自的操作系统不同，因此，根据不同的数据模型（LP32 ILP32 LP64 LLP64 ILP64）所存储的变量在内存中占用的字节也不尽相同。如下表所示：

Data Type ILP32 LP32 ILP64 LP64 LLP64
宏定义 _ _ _ LP64 LLP64
平台 Win32 API / Unix 和 Unix 类的系统 （Linux，Mac OS X） Win16 API Unix 和 Unix 类的系统 （Linux，Mac OS X) Win64 API
char 8 8 8 8 8
short 16 16 16 16 16
int 32 32 64 32 32
long 32 32 64 64 32
long long 64 64 64 64 64
pointer 32 32 64 64 64

一般情况下，windows64位上MinGW使用的是LLP64，long占用32位，而64位Unix、linux使用的是LP64模型，如果在linux下运行该程序，long占用64位。

正是由于此，同样的代码跨不同平台所编译出的可执行文件，变量的大小是不同的。为了保证代码在不同操作系统之间的可移植性，C语言为现有的数据类型创建了更多类型名。

1.5 可移植类型stdint.h

stdint.h定义了一些固定宽度的整数类型别名，主要有下面三类。

  • 宽度完全确定的整数intN_t，比如int32_t。
  • 宽度不小于某个大小的整数int_leastN_t，比如int_least8_t。
  • 宽度不小于某个大小、并且处理速度尽可能快的整数int_fastN_t，比如int_fast64_t。

使用该头文件的好处是保证了编译出的文件变量大小在不同平台是相同的，使用sizeof查看：

1
2
3
4
5
6
7
8
int main() {
    printf("int8_t:%zd\n", sizeof(int8_t));
    printf("int16_t:%zd\n", sizeof(int16_t));
    printf("int32_t:%zd\n", sizeof(int32_t));
    printf("int64_t:%zd\n", sizeof(int64_t));

    return 0;
}

输出结果如下：

1
2
3
4
int8_t:1
int16_t:2
int32_t:4
int64_t:8

2 运算符

C的基本运算符与其他语言的运算符基本相似，比如算术运算符（+、-、*、/、%等），关系运算符（>、<、==、!=等），逻辑运算符（&&、||、!），赋值运算符（=、+=、-=、*=等）。但是平时很少会使用位运算符，这里着重介绍一下：

位运算符作用于二进制位，并逐位执行操作。&（逻辑与）、|（逻辑或）和^（逻辑异或）的真值表如下所示：

p q p & q p | q p ^ q
0 0 0 0 0
0 1 0 1 1
1 1 1 1 0
1 0 0 1 1

除此之外，还有~（逻辑非），>>（向左移位），<<（向右移位）运算符。

在介绍位运算符之前，首先要清楚C语言中整数按照补码存储。对于正数，补码和原码相同；对于负数，补码等于原码的除符号位之外其它位取反加一。因此，位运算符操作的是补码，而不是原码。

移位运算符之左移：

1
2
3
4
5
6
7
8
9
10
11
12
int main() {
    int a=1; //正数，  补码=原码： 00000000 00000000 00000000 00000001
    printf("%d\n",a);
    a<<=2;  //正数左移，补码=原码：  00000000 00000000 00000000 00000100
    printf("%d\n",a);
    int b=-1; // 负数 原码： 10000000 00000000 00000000 00000001
              // 负数 补码： 11111111 11111111 11111111 11111111
    printf("%d\n",b);
    b<<=2;    // 向左移位 补码： 11111111 11111111 11111111 11111100
             // 对应原码 原码： 10000000 00000000 00000000 00000100
    printf("%d\n",b);
}

输出结果为：

1
2
3
4
1
4
-1
-4

移位运算符之右移

1
2
3
4
5
6
7
8
9
10
11
12
13
int main() {
    int a=1; //正数，  补码=原码： 00000000 00000000 00000000 00000001
    printf("%d\n",a);
    a>>=2;  //正数右移，补码=原码： 00000000 00000000 00000000 00000000
    printf("%d\n",a);
    int b=-3; // 负数 原码： 10000000 00000000 00000000 00000011
              // 负数 补码： 11111111 11111111 11111111 11111101
    printf("%d\n",b);
    b>>=2;    // 向右移位 补码： 11111111 11111111 11111111 11111111
              // 对应原码 原码： 10000000 00000000 00000000 00000001
    printf("%d\n",b);
    return 0;
}

输出结果为

1
2
3
4
1
0
-3
-1

可以看到，对于有符号数，移位运算符进行的是算数移位（符号位保持不变）而不是逻辑移位。

取反运算符

1
2
3
4
5
6
7
int main() {
    int a=2; //补码=原码： 00000000 00000000 00000000 00000010
    printf("%d\n",a);
    a=~a;   //对补码取反： 11111111 11111111 11111111 11111101
            //对应的原码： 10000000 00000000 00000000 00000011
    printf("%d\n",a);
}

输出结果为：

1
2
2
-3

可以看到，对于取反运算，连同符号位一起取反。

逻辑与、逻辑或、逻辑异或

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
int main() {
    int a=2; //补码=原码： 00000000 00000000 00000000 00000010
    int b=-1;//原码： 10000000 00000000 00000000 00000001
            // 补码： 11111111 11111111 11111111 11111111
    a&=b;   //对应补码运算： 00000000 00000000 00000000 00000010
            //对应的原码：   00000000 00000000 00000000 00000010
    printf("%d\n",a);
    a=2;
    a|=b;  //对应补码运算： 11111111 11111111 11111111 11111111
           //对应的原码：   10000000 00000000 00000000 00000001
    printf("%d\n",a);
    a=2;
    a^=b;  //对应补码运算： 11111111 11111111 11111111 11111101
           //对应的原码：   10000000 00000000 00000000 00000011
    printf("%d\n",a);

}

3 switch语句

由于一些高级语言可能没有switch语句，这里复习一下C语言的switch语句。它的语法如下：

1
2
3
4
5
6
7
8
switch (整型表达式) {
    case 整型常量1:
		语句..
    case 整型常量2:
        语句..
    default:
        语句..
}

switch语句必须遵循下面的规则：

  • switch语句中的表达式是一个常量表达式，必须是一个整型或枚举类型。
  • 在一个 switch 中可以有任意数量的 case 语句。每个 case 后跟一个要比较的值和一个冒号。
  • 当被测试的变量等于 case 中的常量时，case 后跟的语句将被执行，直到遇到 break 语句为止。
  • 当遇到break语句时，switch终止，控制流将跳转到 switch 语句后的下一行。
  • 不是每一个 case 都需要包含break。如果 case 语句不包含break，控制流将会继续后续的 case，直到遇到 break 为止。
  • 一个switch语句可以有一个可选的default case，出现在switch的结尾。default case可用于在上面所有case都不为真时执行一个任务。default case中的break语句不是必需的。

4 预处理器

C语言的预处理器在执行程序之前把符号缩写替换成其表达的内容，简单说就是进行文本的替换工作。预处理指令从#开始执行，直到遇到第一个换行符\n。

4.1 define

#define通常用于定义常量：

1
#define 宏名称 替换体

比如：

1
2
3
4
5
6
7
8
#define TWO 2
#define FOUR TWO*TWO

int main() {
    int a=TWO;
    int b=FOUR;

}

会被替换成：

1
2
3
4
5
6
7
8
#define TWO 2
#define FOUR TWO*TWO

int main() {
    int a=2;
    int b=2*2; // 预处理器不会进行实际运算，只进行替换；表达式求值在编译期完成

}

还可以使用带参数的宏：

1
2
3
4
5
6
#define SQUARE(X) X*X

int main() {
    int a= SQUARE(3);
    printf("%d",a);
}

4.2 include

#include会将后面的文件内容包含到当前文件中，替换体有两种形式：

1
2
#include <stdio.h>
#include "test.h"

尖括号形式表示预处理器去标准系统目录中查找文件，双引号形式表示在当前目录下查找文件，找不到再去系统目录中查找。

4.3 undef

#undef用于取消已经定义的#define，并且，即使原来没有定义，使用#undef也是有效的。

4.4 条件编译

预处理器 描述
#ifdef 如果宏已经定义，则返回真
#ifndef 如果宏没有定义，则返回真
#if 如果给定条件为真，则编译下面代码
#else 如果给定条件不满足if，则编译下面代码
#elif 如果前面的 #if 给定条件不为真，当前条件为真，则编译下面代码
#endif 结束一个 #if……#else 条件编译块

4.5 内联函数

一般来说，调用一个函数流程为：当前调用命令的地址被保存下来，程序流跳转到所调用的函数并执行该函数，最后跳转回之前所保存的命令地址。

对于需要经常调用的小函数来说，这大大降低了程序运行效率。所以，C99 新增了内联函数（inline function）。

关键字inline告诉编译器，任何地方只要调用内联函数，就直接把该函数的机器码插入到调用它的地方。这样程序执行更有效率，就好像将内联函数中的语句直接插入到了源代码文件中需要调用该函数的地方一样。

注意：内联函数的定义和调用必须在同一个文件中。如果多个文件都想使用同一个内联函数，可以把内联函数定义在头文件中，再其它文件#include包含即可。头文件一般不包含可执行代码，而内联函数是个例外。

4.6 预处理器黏合剂##

##运算符称为预处理器黏合剂，它的作用非常简单，举个例子：

1
2
3
4
5
6
#define NAME(N) (x##N)
int main() {
    int NAME(1)=5;
    printf("%d",x1);
    return 0;
}

这里的x##N会被替换为xN，因此NAME(1)变量就为x1。

5 结构体

下面是一个结构体：

1
2
3
4
5
struct A{
    int a;
    char b;
    int* ptr;
}test;

其中A为结构体标记，括号内可以定义任意的数据类型，并声明了名为test的A类型结构体。

要初始化结构体变量，可以这样：

1
struct A foo={.a=4,.b=2};

配合typedef关键字，可以在初始化结构体变量时，省略struct关键字：

1
2
3
4
5
6
7
typedef struct A{
    int a;
    char b;
}AA;
int main() {
    AA foo={.a=4,.b=2}; // 使用AA来初始化结构体变量
}

在redis源码中，可能会出现__attribute__((packed))，这里有必要说明一下。看如下的两个结构体：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
typedef struct A{
    int a;
    char b;
}AA;
typedef struct __attribute__((packed))B{
    int a;
    char b;
}BB;
int main() {
    AA foo;
    BB bar;
    printf("%zd\n", sizeof(foo));
    printf("%zd\n", sizeof(bar));
}

输出结果：

1
2
8
5

按照常理分析，A和B结构体占用字节应该是相同的，即int4字节+char1字节=5字节，但是在编译过程中，为了提高系统性能，CPU处理内存时，会用数据对齐这种方式。这种方式下，数据在内存中都以固定size保存。而为了进行对齐，有时候就需要在数据中插入一些无意义的字节。比如编译器是以4个bytes为单位对齐的，当你声明一个char的数据，后面就会补齐3个无意义的字节。

而要改变编译器的对齐方式，就要利用到__attribute__关键字，它是用于设置函数属性（Function Attribute）、变量属性（Variable Attribute）、类型属性（Type Attribute）。也可以修饰结构体（struct）或共用体（union）。

写法为__attribute__ ((attribute-list))，后面的attribute-list大概有6个参数值可以设置：aligned, packed, transparent_union, unused, deprecated和 may_alias。

其中packed属性的主要目的是让编译器更紧凑地使用内存：__attribute__((packed))，它的作用就是告诉编译器：取消结构体在编译过程中的优化对齐，按尽可能小的size对齐——也就是按1字节为单位对齐。

__attribute__((packed))和__attribute__((packed, aligned(1)))是等价的。（aligned(x)就是告诉编译器，以x个字节为单位进行对齐，x只能是1，或2的幂）。

redis也使用了这种语法来优化数据结构中的内存占用。

6 伸缩型数组成员

在C99中，新增了一个特性：伸缩型数组成员。用此特性声明的结构，其最后一个数组成员会具有一些特性：

  • 该成员数组不会立刻存在（程序运行时，不会立即为其分配内存空间）
  • 程序员可以利用其创建数目为指定个数的数组

正确声明一个伸缩型数组成员需要满足以下几条规则：

  • 伸缩型数组成员必须是结构体的最后一个成员
  • 结构体中，必须有且至少有一个成员
  • 声明方法类似于普通函数，只不过它的方括号中不用指定数组元素个数，一般设置为空或0
1
2
3
4
5
6
7
struct A{
    int a;
    int foo[];
}a;
int main() {
    printf("%zd\n", sizeof(a));
}

在声明中不能用数组foo[]做任何事，因为系统并没有给该数组分配存储空间。事实上，这和在C99中引入该特性的初衷有关。它并不是让你声明一个struct A类型的结构体变量，而是希望你声明一个指向struct A类型的指针，然后调用malloc函数来为这个伸缩型数组成员动态分配足够的空间，以存储struct A类型结构体中其他内容和伸缩性数组成员所需的额外存储空间，如：

1
2
3
4
5
6
7
8
9
10
struct A{
    int a;
    int foo[];
}a;
int main() {
    struct A* ptr= malloc(sizeof(struct A)+5*sizeof(int)); // 结构体中动态分配一个长度为5的foo数组
    ptr->foo[4]=3;
    printf("%d\n", ptr->foo[4]);
    free(ptr);
}

redis使用了这种特性来优化内存空间的使用。

7 联合

联合union，也称共用体，是一种数据类型，它可以在同一内存空间中存储不同的数据类型（不是同时储存），redis的字典实现中就使用到了该数据类型。它的定义和结构体很相似：

1
2
3
4
5
6
7
union 共用体名{
    成员列表
};
// 如果后续不再使用，也可以匿名定义
union {
    成员列表
}foo;

举个例子：

1
2
3
4
5
union A{
    int aa;
    char bb;
    long long cc;
}foo;

union的内存占用以成员列表中最大的数据类型为准，与结构体不同的是，union的所有成员占用同一段内存，修改一个成员会影响其余所有成员的值。

好了，了解了以上知识，就可以尝试阅读redis源码了。在源码中出现C语言的其它更多内容，会在后面进行补充。

三、数据结构

本节参考资料：

1、Redis 设计与实现（第二版）

2、Redis学习（3）——链表

1 简单动态字符串

1.1 数据结构

源码位于./src/sds.c以及./src/sds.h。sds的意思是简单动态字符串（Simple Dynamic String）。redis定义了如下五种不同长度的SDS类型（SDS_TYPE_5没有被使用过，实际使用的应该是4种）：

1
2
3
4
5
#define SDS_TYPE_5  0
#define SDS_TYPE_8  1
#define SDS_TYPE_16 2
#define SDS_TYPE_32 3
#define SDS_TYPE_64 4

它们的结构体声明如下：

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
/* Note: sdshdr5 is never used, we just access the flags byte directly.
 * However is here to document the layout of type 5 SDS strings. */
struct __attribute__ ((__packed__)) sdshdr5 {
    unsigned char flags; /* 3 lsb of type, and 5 msb of string length */
    char buf[];
};
struct __attribute__ ((__packed__)) sdshdr8 {
    uint8_t len; /* used */
    uint8_t alloc; /* excluding the header and null terminator */
    unsigned char flags; /* 3 lsb of type, 5 unused bits */
    char buf[];
};
struct __attribute__ ((__packed__)) sdshdr16 {
    uint16_t len; /* used */
    uint16_t alloc; /* excluding the header and null terminator */
    unsigned char flags; /* 3 lsb of type, 5 unused bits */
    char buf[];
};
struct __attribute__ ((__packed__)) sdshdr32 {
    uint32_t len; /* used */
    uint32_t alloc; /* excluding the header and null terminator */
    unsigned char flags; /* 3 lsb of type, 5 unused bits */
    char buf[];
};
struct __attribute__ ((__packed__)) sdshdr64 {
    uint64_t len; /* used */
    uint64_t alloc; /* excluding the header and null terminator */
    unsigned char flags; /* 3 lsb of type, 5 unused bits */
    char buf[];
};

可以看到，这些结构体声明除了大小不同之外，基本相似（sdshdr5不使用，不算在内），都拥有如下字段：

  • len表示当前字符串的长度，通过len可以计算出已经实际使用的空间
  • alloc表示已申请的空间（总长度），减去len就是未使用的空间，初始时和len相同
  • flags用于表示sds类型， 3 lsb of type, 5 unused bits中的lsb指的是最低有效位，由于只有五种类型，所以只需低3位即可表示，高5位没有使用
  • buf应用了伸缩型数组成员的特性，在使用时动态分配内存，保存的就是真实的字符串数据，末尾还有一个\0

下图展示了sds在内存中的数据存储：

image-20220304051022899

在取消优化对齐之后，整个sds结构在内存中连续存储，其中len、alloc、flags统称头部header，buf存储真正的字符串。这里的sds s指针在源码中是这样的：

1
typedef char *sds;

在图中可以看到，这个字符指针直接指向buf，用这个指针来操作整个结构体，这样做的好处是可以兼容C语言标准库中的string.h。那如何获取前面的成员呢？由于内存是连续分配的，C语言支持指针自减操作，所以将s减一就可以获取到flags从而判断出sds类型。

看下面的代码：

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
struct __attribute__ ((__packed__)) sdshdr8 {
    uint8_t len; /* used */
    uint8_t alloc; /* excluding the header and null terminator */
    unsigned char flags; /* 3 lsb of type, 5 unused bits */
    char buf[];
};
int main() {
    struct sdshdr8*ptr=malloc(sizeof(struct sdshdr8)+3*sizeof(char));
    ptr->len=3;
    ptr->alloc=3;
    ptr->flags=1;
    ptr->buf[0]='a';
    ptr->buf[1]='e';
    ptr->buf[2]='f';
    sds p=ptr->buf;
    printf("%d\n", *(p-3));
    printf("%d\n", *(p-2));
    printf("%d\n", *(p-1));
    printf("%c\n", *p);
    printf("%c\n", *(p+1));
    printf("%c\n", *(p+2));
}
// 输出：
3
3
1
a
e
f

通过对指向buf的指针减1，就可以获得flags，在C语言中，除了使用*解引用之外，还可以直接使用中括号取索引的形式：

1
2
3
printf("%d\n", *(p-1));
// 等价于
printf("%d\n", p[-1]);

1.2 sdslen

首先，先看一个简单的函数sdslen，它返回字符串的长度。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
#define SDS_TYPE_MASK 7
#define SDS_HDR(T,s) ((struct sdshdr##T *)((s)-(sizeof(struct sdshdr##T))))
#define SDS_TYPE_5_LEN(f) ((f)>>SDS_TYPE_BITS)

static inline size_t sdslen(const sds s) {
    // 获取flags
    unsigned char flags = s[-1];
    // 和掩码进行按位与操作，来判断当前的类型
    switch(flags&SDS_TYPE_MASK) {
        case SDS_TYPE_5:
            return SDS_TYPE_5_LEN(flags);
        case SDS_TYPE_8:
            return SDS_HDR(8,s)->len;
        case SDS_TYPE_16:
            return SDS_HDR(16,s)->len;
        case SDS_TYPE_32:
            return SDS_HDR(32,s)->len;
        case SDS_TYPE_64:
            return SDS_HDR(64,s)->len;
    }
    return 0;
}

首先，调用sdslen传入一个指向buf的指针，通过负索引获取flags成员，使用按位与运算进行判断，使用SDS_HDR传入对应参数，最后返回len长度。

比如传入的指针对应的结构体类型是sdshdr8，那么flags应该为1，二进制表示为00000001，将它和SDS_TYPE_MASK按位与：

1
2
3
4
00000001
00000111
--------
00000001

为什么不直接判断而是使用掩码？redis源码中大量使用了位运算符，我的猜想是（不一定对）：作者是为了提升效率，因为相比于其他运算符而言，直接操作bit位速度更快，想必作者也一定经过了大量的测试而采用这种优化方法。

总之，经过这个switch语句，最终就会"调用"SDS_HDR，以sdshdr8为例

1
2
3
4
return SDS_HDR(8,s)->len;// “调用”SDS_HDR(8,s)
#define SDS_HDR(T,s) ((struct sdshdr##T *)((s)-(sizeof(struct sdshdr##T))))
// 替换为：
// SDS_HDR(8,s) ((struct sdshdr8 *)((s)-(sizeof(struct sdshdr8))))

这里的s-sizeof(struct sdshdr8)相当于s-3，即经过计算将指向buf的指针减去3，得到的就是指向整个结构体指针。

因此return SDS_HDR(8,s)->len相当于return 结构体指针->len，整个操作没有遍历字符数组，时间复杂度为O(1)级别。

在原生的C语言中，字符串不记录自身的长度。如果想要获取长度，必须遍历整个字符串，直至遇到\0为止，时间复杂度为O(n)，并且，由于只要遇到\0就结束，所以C语言字符串只能保存文本数据，不能保存二进制数据，在sds中，由于增加了len成员，解决了这个问题。

1.3 sdsavail

再来看看sdsavail函数，它与sdslen类似，返回的是剩余可用的长度。

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
#define SDS_HDR_VAR(T,s) struct sdshdr##T *sh = (void*)((s)-(sizeof(struct sdshdr##T)));
static inline size_t sdsavail(const sds s) {
    unsigned char flags = s[-1];
    switch(flags&SDS_TYPE_MASK) {
        case SDS_TYPE_5: {
            return 0;
        }
        case SDS_TYPE_8: {
            SDS_HDR_VAR(8,s);
            return sh->alloc - sh->len;
        }
        case SDS_TYPE_16: {
            SDS_HDR_VAR(16,s);
            return sh->alloc - sh->len;
        }
        case SDS_TYPE_32: {
            SDS_HDR_VAR(32,s);
            return sh->alloc - sh->len;
        }
        case SDS_TYPE_64: {
            SDS_HDR_VAR(64,s);
            return sh->alloc - sh->len;
        }
    }
    return 0;
}

这里主要看另一个宏SDS_HDR_VAR，以sdshdr8为例：

1
2
3
4
SDS_HDR_VAR(8,s); // “调用”SDS_HDR_VAR
#define SDS_HDR_VAR(T,s) struct sdshdr##T *sh = (void*)((s)-(sizeof(struct sdshdr##T)));
// 替换为：
// SDS_HDR_VAR(8,s) struct sdshdr8 *sh = (void*)((s)-(sizeof(struct sdshdr8)));

得到一个名为sh的结构体指针（由于要进行运算，使用名字可以更加简洁），最后计算alloc-len返回，因此它的时间复杂度也为O(1)级别。

通过这两个函数的源码，就可以大概了解sds的套路，其实sds还封装了另外4个基础函数，这些函数都被设置为内联的static函数，定义在sds.h头文件中：

  • 设置长度sdssetlen
  • 增加长度sdsinclen
  • 获取总占用空间sdsalloc
  • 设置总占用空间sdssetalloc

它们的时间复杂都为O(1)级别。

1.4 sdsnewlen

sdsnewlen用于创建一个新sds字符串，它定义如下：

1
2
3
sds sdsnewlen(const void *init, size_t initlen) {
    return _sdsnewlen(init, initlen, 0);
}

它调用了_sdsnewlen，下划线是一种编程规范，在c++或java中可以理解为private，在python中类似于双下划线__开头的函数，表示不希望外界直接调用该函数。它的源码如下：

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
sds _sdsnewlen(const void *init, size_t initlen, int trymalloc) {
    void *sh;
    sds s;
    // 使用sdsReqType函数传入初始字符串长度，根据长度获得sds类型
    char type = sdsReqType(initlen);
    // 对于空字符串，不使用sdshdr5而是使用sdshdr8
    if (type == SDS_TYPE_5 && initlen == 0) type = SDS_TYPE_8;
    // 调用sdsHdrSize获取该结构体类型的头部长度
    int hdrlen = sdsHdrSize(type);
    
    unsigned char *fp; //flags指针
    size_t usable;
	// 断言，判断是否长度溢出，+1是因为末尾要保存'\0'字符
    assert(initlen + hdrlen + 1 > initlen); /* Catch size_t overflow */
    // 如果trymalloc不为0，通过s_trymalloc_usable分配内存，否则调用s_malloc_usable
    sh = trymalloc?
        s_trymalloc_usable(hdrlen+initlen+1, &usable) :
        s_malloc_usable(hdrlen+initlen+1, &usable);
    // 如果分配失败sh为NULL,则返回NULL
    if (sh == NULL) return NULL;
    if (init==SDS_NOINIT)
        init = NULL;
    else if (!init)
        // 将这段内存的所有位都设置为0
        memset(sh, 0, hdrlen+initlen+1);
    // sh是结构体指针，加上头部长度得到指向buf的指针s
    s = (char*)sh+hdrlen;
    // fp是指向flag的指针
    fp = ((unsigned char*)s)-1;
    usable = usable-hdrlen-1;
    if (usable > sdsTypeMaxSize(type))
        // 如果可用空间大于该类型的最大空间，设置为最大空间
        usable = sdsTypeMaxSize(type);
    switch(type) {
        case SDS_TYPE_5: {
            *fp = type | (initlen << SDS_TYPE_BITS);
            break;
        }
        case SDS_TYPE_8: {
            SDS_HDR_VAR(8,s);
            // 初始化设置
            sh->len = initlen;
            sh->alloc = usable;
            *fp = type;
            break;
        }
        case SDS_TYPE_16: {
            SDS_HDR_VAR(16,s);
            sh->len = initlen;
            sh->alloc = usable;
            *fp = type;
            break;
        }
        case SDS_TYPE_32: {
            SDS_HDR_VAR(32,s);
            sh->len = initlen;
            sh->alloc = usable;
            *fp = type;
            break;
        }
        case SDS_TYPE_64: {
            SDS_HDR_VAR(64,s);
            sh->len = initlen;
            sh->alloc = usable;
            *fp = type;
            break;
        }
    }
    if (initlen && init)
        // 从init指针开始，将字符内存拷贝到buf中
        memcpy(s, init, initlen);
    // 在结尾添加'\0'结束符
    s[initlen] = '\0';
    return s;
}

1.5 sdsMakeRoomFor

sdsMakeRoomFor函数用于对buf进行扩容，它传入一个sds指针和需要增加的长度，扩容后返回s。函数定义如下：

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
sds sdsMakeRoomFor(sds s, size_t addlen) {
    void *sh, *newsh;
    // 调用sdsavail函数，获取可用空间
    size_t avail = sdsavail(s);
    // 当前长度，最终长度，请求长度
    size_t len, newlen, reqlen;
    // 新类型，旧类型（仍然通过flag和掩码按位与获得）
    char type, oldtype = s[-1] & SDS_TYPE_MASK;
    // 头部长度
    int hdrlen;
    size_t usable;

    /* Return ASAP if there is enough space left. */
    // 如果可用空间大于需要增加的空间，尽快返回。
    if (avail >= addlen) return s;
	// 获得当前长度
    len = sdslen(s);
    // sh为当前结构体指针
    sh = (char*)s-sdsHdrSize(oldtype);
    // 请求长度=最终长度=当前长度+申请长度
    reqlen = newlen = (len+addlen);
    // 溢出判断
    assert(newlen > len);   /* Catch size_t overflow */
    // SDS_MAX_PREALLOC是最大的预分配长度，默认为1024*1024B=1MB
    if (newlen < SDS_MAX_PREALLOC)
        // 如果小于最大预分配长度，以指数型增长
        newlen *= 2;
    else
        // 如果大于等于最大预分配长度，以线性增长，每次增长一个最大预分配长度（1MB）
        newlen += SDS_MAX_PREALLOC;
	// 根据新的长度，决定使用哪种数据结构存储
    type = sdsReqType(newlen);

    // 不使用SDS_TYPE_5，而是使用SDS_TYPE_8
    if (type == SDS_TYPE_5) type = SDS_TYPE_8;
	// 根据新的类型，获得头部长度
    hdrlen = sdsHdrSize(type);
    // 溢出判断
    assert(hdrlen + newlen + 1 > reqlen);  /* Catch size_t overflow */
    // 如果旧类型与新类型相同
    if (oldtype==type) {
        // 调用s_realloc_usable分配内存，得到结构体指针newsh
        newsh = s_realloc_usable(sh, hdrlen+newlen+1, &usable);
        if (newsh == NULL) return NULL;
        // 计算得到buf指针，重新赋值给s
        s = (char*)newsh+hdrlen;
    } else { // 如果旧类型与新类型不同
        
        /* Since the header size changes, need to move the string forward,
         * and can't use realloc */
        // 调用s_malloc_usable分配内存
        newsh = s_malloc_usable(hdrlen+newlen+1, &usable);
        if (newsh == NULL) return NULL;
        // 将原来的buf数组内容拷贝到新的类型中，len+1是因为结尾有'\0'
        memcpy((char*)newsh+hdrlen, s, len+1);
        // 释放旧内存
        s_free(sh);
        // 计算得到buf指针，重新赋值给s
        s = (char*)newsh+hdrlen;
        // 设置type成员的类型为新类型
        s[-1] = type;
        // 设置长度len
        sdssetlen(s, len);
    }
    // 计算可用空间
    usable = usable-hdrlen-1;
    if (usable > sdsTypeMaxSize(type))
        // 如果可用空间大于该类型的最大空间，设置为最大空间
        usable = sdsTypeMaxSize(type);
    // 设置alloc成员的值
    sdssetalloc(s, usable);
    return s;
}

通过源码，可以得出以下信息：

  • 如果待分配长度小于最大预分配长度（1MB），以指数型增长，比如当前长度len=10，申请addlen=10，那么newlen=20*2=40B；如果待分配长度大于等于最大预分配长度，则以线性增长，比如len=1047552，addlen=2000，由于newlen=1049552>1048576因此，newlen=1049552+1048576=2098128。
  • 如果新增长度之后类型不改变，直接realloc，如果类型改变，则需要手动malloc，并释放掉之前申请的内存。

这种内存申请策略很像TCP中的慢开始和拥塞避免算法，即指数增长到阈值之后，采用线性增长。

1.6 sdstrim

sdstrim函数用于删除一部分字符，它采用惰性空间释放策略：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
sds sdstrim(sds s, const char *cset) {
    char *start, *end, *sp, *ep;
    size_t len;
	// 起始指针
    sp = start = s;
    // 结束指针
    ep = end = s+sdslen(s)-1;
    // 只要当前字符在cset查找表中，sp指针一直向后找，直到没找到为止
    while(sp <= end && strchr(cset, *sp)) sp++;
    // 同理，从后向前查找
    while(ep > sp && strchr(cset, *ep)) ep--;
    // sp大于ep则全部查找成功，len=0，否则设置剩余长度，+1是因为还有'\0'字符
    len = (sp > ep) ? 0 : ((ep-sp)+1);
    // sp移动过，则至少查找成功，使用memmove进行内存拷贝，相比于memcpy，前者可以处理内存重叠的情况
    if (s != sp) memmove(s, sp, len);
    // 结尾设置为'\0'
    s[len] = '\0';
    // 更新长度len
    sdssetlen(s,len);
    return s;
}

可以看出，所谓惰性空间释放，就是在函数中不会使用realloc重新分配内存，而是只进行了拷贝。由于数据结构中保存了len，因此只要更新len就代表更新了长度，并不会造成问题。这种策略的优点就是在频繁更改字符串长度的情况下，大大减少了内存重新分配的次数。

类似的策略在其它函数中也频繁出现，如：

1
2
3
4
void sdsclear(sds s) {
    sdssetlen(s, 0);
    s[0] = '\0';
}

sdsclear函数调用sdssetlen，第二个参数传0，仅仅将len设置为0代表空字符串，但并不释放内存空间。

当然，redis也提供了真正释放内存的函数sdsRemoveFreeSpace，它的实现和sdsMakeRoomFor基本相同。

2 列表

在redis3.2之前的版本中，列表的底层实现为双向链表和压缩链表，从redis3.2之后的版本，列表采用快速列表的底层实现，它是一种综合了双向链表和压缩链表的数据结构。接下来会对这三种数据结构分别介绍。

2.1 双向链表

源码位于./src/adlist.c以及./src/adlist.h。adlist的意思是A generic doubly linked list，即一个通用的双向链表。数据结构定义如下：

1
2
3
4
5
typedef struct listNode {
    struct listNode *prev;
    struct listNode *next;
    void *value;
} listNode;

node节点定义非常简单，前指针和后指针，以及数据指针，为了更加方便操作，redis还定义了另一个结构体：

1
2
3
4
5
6
7
8
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;

其中，head和tail分别指向整个链表的头节点和尾节点，还定义了dup、free、match三个可自定义的函数。，分别用来复制、释放、比较节点，另外，为了获取长度的复杂度为O(1)，定义了len表示列表长度。如下图：

img

对于链表来说，它的特点是内存可以不连续分配，使用双向链表，保证了在寻找前驱节点时的时间复杂度也为O(1)，但是正因为链表节点间内存不连续，也就失去了顺序性的特点，并且根据局部性原理，链表对于缓存的利用不如数组。

还有比较关键的一点是，这种数据结构在数据规模很小的时候，由于每个节点都需要保存前驱和后继指针，即多占用了16个字节的空间用于存储指针，这也是redis使用压缩链表的原因。

2.2 压缩链表

这是一种非常节省内存的顺序存储结构，一个压缩列表可以包含任意多个节点（entry）， 每个节点可以存储字符串或整数值，并且支持在两端以O(1)的时间复杂度进行push和pop操作。ziplist的总体结构如下:

1
<zlbytes> <zltail> <zllen> <entry> <entry> ... <entry> <zlend>

其中：

属性 类型 长度 用途
zlbytes uint32_t 4 字节 记录整个压缩列表占用的内存字节数：在对压缩列表进行内存重分配， 或者计算 zlend 的位置时使用。
zltail uint32_t 4 字节 记录压缩列表表尾节点距离压缩列表的起始地址有多少字节： 通过这个偏移量，程序无须遍历整个压缩列表就可以确定表尾节点的地址，从而进行pop操作。
zllen uint16_t 2 字节 记录了压缩列表包含的节点数量： 当这个属性的值小于 UINT16_MAX （65535）时， 这个属性的值就是压缩列表包含节点的数量； 当这个值等于 UINT16_MAX 时， 节点的真实数量需要遍历整个压缩列表才能计算得出。
entryX 列表节点 不定 压缩列表包含的各个节点，节点的长度由节点保存的内容决定。
zlend uint8_t 1 字节 特殊值 0xFF （十进制 255 ），用于标记压缩列表的末端。

每个压缩列表节点由 prevlen 、 encoding 、 entry-data 三个部分组成。其中，prevlen 记录了该节点前一个节点的长度，并且prevlen的长度取决于前一个节点长度：

  • 如果前一个节点长度小于254字节，它将只消耗一个1字节， 前一节点的长度就保存在这1字节里面。
  • 如果前一个节点长度大于等于254字节，它将消耗一个5字节，其中第一个字节会被设置为 0xFE（254）， 而之后的四个字节则用于保存前一节点的长度。

所以实际上每个节点是这样组成的：

1
2
前节点长度小于254：<prevlen from 0 to 253> <encoding> <entry>
前节点长度大于等于254：0xFE <4 bytes unsigned little endian prevlen> <encoding> <entry>

使用该字段，就可以实现从表尾向表头遍历。只要我们拥有了一个指向某个节点起始地址的指针，那么通过这个指针以及prevlen， 程序就可以一直向前一个节点回溯， 最终到达压缩列表的表头节点。

接下来是encoding，它记录了节点的数据类型以及长度：

  • 如果是字符串类型，前两个bit用来用来存储它的类型
  • 如果是数字，前两个bit全部置为1

下表展示了如何使用1字节来表示节点的类型（bbbbbb、xxxxx、aaaaa等代表实际的二进制数据）：

编码 编码长度 entry-data 属性保存的值
00bbbbbb 1 字节 长度小于等于 63 字节的字节数组。
01bbbbbb xxxxxxxx 2 字节 长度小于等于 16383 字节的字节数组。
10______ aaaaaaaa bbbbbbbb cccccccc dddddddd 5 字节 长度小于等于 4294967295 的字节数组。
11000000 1 字节 int16_t 类型的整数。
11010000 1 字节 int32_t 类型的整数。
11100000 1 字节 int64_t 类型的整数。
11110000 1 字节 24 位有符号整数。
11111110 1 字节 8 位有符号整数。
1111xxxx（xxxx在0001-1101之间） 1 字节 无
11111111（表示尾节点，实际就是zlend） 1 字节 无

entry-data则用于保存节点的值， 节点值可以是一个字节数组或者整数， 值的类型和长度由节点的 encoding 属性决定。

下面是一个包含两个节点（字符串2和5）的ziplist。它由15个字节组成：

1
2
3
[0f 00 00 00] [0c 00 00 00] [02 00] [00 f3] [02 f6] [ff]
      |             |          |       |       |     |
    zlbytes        zltail    zllen   "2"     "5"   end
  1. 前4个字节zlbytes表示总长度为15字节
  2. 之后4个字节表示最后一个节点的偏移量为12字节
  3. 接下来2字节记录了压缩列表包含的节点数量为2个。
  4. 继续向后，由于这是第一个entry节点所以prevlen为00，f3的二进制为11111011代表encoding，这是1111xxxx的编码形式，由于xxxx在0001-1101（即1-13之间），又由于数字是从0开始，将其减一就可以用这四位保存了一个介于 0 和12 之间的值，此时没有entry-data字段。所以用3-1得到真实存储的“2”。
  5. 继续向后，这也是entry节点，02表示前面的节点长度为2字节，f6即11110110，和前面同理，由于1111xxxx中的xxxx可以保存介于 0 和12 之间的值，所以实际保存的值为6-1=5。
  6. 最后，ff用于标记压缩列表的末端。

可以看到这种存储方式充分地利用了内存空间（甚至连xxxx都要利用上来保存0-12数字），根据数据大小和类型进行不同的编码和内存分配也是redis能够如此高效的原因。

那么，为什么redis在3.2以后转而使用quicklist，而不是使用双向链表+压缩链表呢？如果说双向链表的缺点是指针消耗了过多内存空间，压缩链表+双向链表的实现有什么缺点？

看下面的情况：

1
<zlbytes> <zltail> <zllen> <entry1> <entry2> ... <entryN> <zlend>

这是一个压缩链表，假设entry1、entry2一直到entryN的长度都在250-253字节之间，根据规则，prevlen只占用1字节，现在，假设再插入一个NewEntry，这个节点长度等于254：

1
<zlbytes> <zltail> <zllen> <NewEntry> <entry1> <entry2> ... <entryN> <zlend>

此时，后续节点为了维护前面所说的规则，要对自己的prevlen进行更新，entry1发现前面的节点等于254，prevlen改为占用5字节，此时entry1本身占用就增加了4字节，即长度在254-257字节之间，事情变得更糟糕了，后续的entry2也会如此，像多米诺骨牌一样，最终entry1、entry2一直到entryN的prevlen都被更新一遍。接下来通过源码来验证这个过程，在这个例子中，我们在整个entry前插入了一个节点，redis为这种操作提供了ziplistPush函数：

1
2
3
4
5
6
7
8
9
10
#define ZIPLIST_HEAD 0
#define ZIPLIST_TAIL 1
// 这个函数的作用是：根据where获取到要插入的位置（从前插入还是从后插入）的指针p，然后将长度为slen的字符串s插入到zl所指的压缩链表中
unsigned char *ziplistPush(unsigned char *zl, unsigned char *s, unsigned int slen, int where) {
    unsigned char *p;
    // where为0代表从头插入，为1代表从尾插入
    p = (where == ZIPLIST_HEAD) ? ZIPLIST_ENTRY_HEAD(zl) : ZIPLIST_ENTRY_END(zl);
    // 调用__ziplistInsert进行插入
    return __ziplistInsert(zl,p,s,slen);
}

这会在内部调用封装的“私有”函数__ziplistInsert：

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
// 这个函数被封装为私有，它通过p指针指的位置，将长度为slen的字符串s插入到zl所指的压缩列表中，然后返回插入完成后的新的ziplist
unsigned char *__ziplistInsert(unsigned char *zl, unsigned char *p, unsigned char *s, unsigned int slen) {
    // curlen为当前压缩链表的长度，reqlen将来用于保存节点的长度
    size_t curlen = intrev32ifbe(ZIPLIST_BYTES(zl)), reqlen, newlen;
    unsigned int prevlensize, prevlen = 0;
    size_t offset;
    int nextdiff = 0;
    unsigned char encoding = 0;
    // 初始化一个值，用来避免出现警告
    long long value = 123456789; /* initialized to avoid warning. Using a value
                                    that is easy to see if for some reason
                                    we use it uninitialized. */
    zlentry tail;

    /* Find out prevlen for the entry that is inserted. */
    // 获取前面的节点长度，以此为新节点设立prevlen
    if (p[0] != ZIP_END) {
        // 如果p不指向结尾，那么ziplist不为空，通过ZIP_DECODE_PREVLEN宏获取prevlen
        ZIP_DECODE_PREVLEN(p, prevlensize, prevlen);
    } else {
        // p指向结尾，获取到结尾指针ptail
        unsigned char *ptail = ZIPLIST_ENTRY_TAIL(zl);
        if (ptail[0] != ZIP_END) {
            // 如果ptail不为空，说明ziplist不为空，ptail指向的是最后一个节点
            // 此时要在结尾插入，ptail指向的节点为前置节点，通过它的长度设置prevlen
            prevlen = zipRawEntryLengthSafe(zl, curlen, ptail);
        }
    }

    /* See if the entry can be encoded */
    // 查看是否能将entry编码，尝试将string转换为整数
    if (zipTryEncoding(s,slen,&value,&encoding)) {
        // 可以转换为整数，encoding已经被设置为最适合的编码类型，并且value被设置为转换后的值
        /* 'encoding' is set to the appropriate integer encoding */        
        // zipIntSize函数通过encoding返回合适的长度
        reqlen = zipIntSize(encoding);
    } else {
        // 不可以转换，此时encoding仍为0
        /* 'encoding' is untouched, however zipStoreEntryEncoding will use the
         * string length to figure out how to encode it. */
        // reqlen暂时设置为slen
        reqlen = slen;
    }
    /* We need space for both the length of the previous entry and
     * the length of the payload. */
    // 首先根据前面节点获取prevlen到底为1还是5，将这个值加到reqlen上
    reqlen += zipStorePrevEntryLength(NULL,prevlen);
    // 再加上当前节点本身的值所需的大小，在这个函数中同时处理了字符串的encoding
    reqlen += zipStoreEntryEncoding(NULL,encoding,slen);

    /* When the insert position is not equal to the tail, we need to
     * make sure that the next entry can hold this entry's length in
     * its prevlen field. */
    // 当要插入的位置不是尾部时，就需要检查后面的节点的prevlen是否需要扩容
    int forcelarge = 0;
    // 要插入的位置不是尾部时，通过zipPrevLenByteDiff函数，传入p指针和reqlen，进行如下计算：
    // 通过p指向的节点计算现在用来编码p的字节数：A，通过reqlen计算所需要的字节数：B
    // 用B-A得到结果，如果是负数，说明所需节点可以用更少的空间存储，如果为0则不需要变化，如果为正数说明所需节点需要更多空间
    // 将这个结果赋值给nextdiff
    nextdiff = (p[0] != ZIP_END) ? zipPrevLenByteDiff(p,reqlen) : 0;
    if (nextdiff == -4 && reqlen < 4) {
        // 如果新节点长度小于4，把nextdiff设为0，标记forcelarge设置为1
        nextdiff = 0;
        forcelarge = 1;
    }

    /* Store offset because a realloc may change the address of zl. */
    //存储偏移量，因为realloc可能会改变zl的地址
    offset = p-zl;
    // newlen=ziplist原来的长度+新节点的长度+后面一个节点需要扩容的长度
    newlen = curlen+reqlen+nextdiff;
    // 调整ziplist的大小
    zl = ziplistResize(zl,newlen);
    // 根据偏移量，还原p的位置
    p = zl+offset;

    /* Apply memory move when necessary and update tail offset. */
    // 如果插入的节点不在尾部，要对内存进行调整
    if (p[0] != ZIP_END) {
        /* Subtract one because of the ZIP_END bytes */
        // 从p-nextdiff位置开始，整体后移，为新节点腾出位置
        memmove(p+reqlen,p-nextdiff,curlen-offset-1+nextdiff);

        /* Encode this entry's raw length in the next entry. */
        // 为后面的节点设置encoding
        if (forcelarge)
            // forcelarge=1
            zipStorePrevEntryLengthLarge(p+reqlen,reqlen);
        else
            // forcelarge=0
            zipStorePrevEntryLength(p+reqlen,reqlen);

        /* Update offset for tail */
        // 更新ziplist尾部偏移量
        ZIPLIST_TAIL_OFFSET(zl) =
            intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+reqlen);

        /* When the tail contains more than one entry, we need to take
         * "nextdiff" in account as well. Otherwise, a change in the
         * size of prevlen doesn't have an effect on the *tail* offset. */
        // 安全更新整个结构体的信息
        assert(zipEntrySafe(zl, newlen, p+reqlen, &tail, 1));
        if (p[reqlen+tail.headersize+tail.len] != ZIP_END) {
            // 如果要插入的节点后面不止一个节点，更新ziplist尾部偏移量的时候要加上nextdiff
            // 不然的话对于prevlen的更新对于表尾的节点没有效果
            ZIPLIST_TAIL_OFFSET(zl) =
                intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+nextdiff);
        }
    } else {
        /* This element will be the new tail. */
        // 如果要插入的节点就是尾节点，更新尾节点offset
        ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(p-zl);
    }

    /* When nextdiff != 0, the raw length of the next entry has changed, so
     * we need to cascade the update throughout the ziplist */
    // 当nextdiff不为0时，说明我们更新了后续节点的长度，因此还需要级联更新再后续的节点
    if (nextdiff != 0) {
        offset = p-zl;
        zl = __ziplistCascadeUpdate(zl,p+reqlen);
        p = zl+offset;
    }

    /* Write the entry */
    // 把prevlen写入新节点
    p += zipStorePrevEntryLength(p,prevlen);
    // 将更新encoding
    p += zipStoreEntryEncoding(p,encoding,slen);
    if (ZIP_IS_STR(encoding)) {
        // 如果是字符串
        memcpy(p,s,slen);
    } else {
        // 如果是数字
        zipSaveInteger(p,value,encoding);
    }
    // 更新ziplist的节点数量
    ZIPLIST_INCR_LENGTH(zl,1);
    return zl;
}

其它的细节暂时忽略，主要研究级联更新__ziplistCascadeUpdate函数：

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
/* We use this function to receive information about a ziplist entry.
 * Note that this is not how the data is actually encoded, is just what we
 * get filled by a function in order to operate more easily. */
// 我们使用这个结构体，以便更好地操作ziplist节点，因此这个结构体并不是节点的实际编码形式
typedef struct zlentry {
    unsigned int prevrawlensize; /* 前一个节点使用的字节数*/
    unsigned int prevrawlen;     /* 前一个节点的长度*/
    unsigned int lensize;        /* 当前节点使用的字节数 */
    unsigned int len;            /* 当前节点的长度 */
    unsigned int headersize;     /* 当前节点header字节数，它等于prevrawlensize + lensize */
    unsigned char encoding;      /* 节点的编码类型 */
    unsigned char *p;            /* 指向当前节点的指针 */
} zlentry;

unsigned char *__ziplistCascadeUpdate(unsigned char *zl, unsigned char *p) {
    zlentry cur;
    size_t prevlen, prevlensize, prevoffset; /* Informat of the last changed entry. */
    size_t firstentrylen; /* Used to handle insert at head. */
    size_t rawlen, curlen = intrev32ifbe(ZIPLIST_BYTES(zl));
    size_t extra = 0, cnt = 0, offset;
    size_t delta = 4; /* Extra bytes needed to update a entry's prevlen (5-1). */
    unsigned char *tail = zl + intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl));

    /* Empty ziplist */
    if (p[0] == ZIP_END) return zl;

    zipEntry(p, &cur); /* no need for "safe" variant since the input pointer was validated by the function that returned it. */
    firstentrylen = prevlen = cur.headersize + cur.len;
    prevlensize = zipStorePrevEntryLength(NULL, prevlen);
    prevoffset = p - zl;
    p += prevlen;

    /* Iterate ziplist to find out how many extra bytes do we need to update it. */
    // 迭代ziplist确定多少字节需要更新
    while (p[0] != ZIP_END) {
        assert(zipEntrySafe(zl, curlen, p, &cur, 0));

        /* Abort when "prevlen" has not changed. */
        // 当后面的不需要更新时，结束循环
        if (cur.prevrawlen == prevlen) break;

        /* Abort when entry's "prevlensize" is big enough. */
        if (cur.prevrawlensize >= prevlensize) {
            if (cur.prevrawlensize == prevlensize) {
                //cur节点长度等于prevlensize
                zipStorePrevEntryLength(p, prevlen);
            } else {
                // cur节点长度小于prevlensize，这种情况意味着缩容，但是这里并不缩容
                // 因为可能会发生抖动现象
                /* This would result in shrinking, which we want to avoid.
                 * So, set "prevlen" in the available bytes. */
                zipStorePrevEntryLengthLarge(p, prevlen);
            }
            break;
        }

        /* cur.prevrawlen means cur is the former head entry. */
        assert(cur.prevrawlen == 0 || cur.prevrawlen + delta == prevlen);

        /* Update prev entry's info and advance the cursor. */
        // 更新前一个节点的信息并移动指针
        rawlen = cur.headersize + cur.len;
        prevlen = rawlen + delta; 
        prevlensize = zipStorePrevEntryLength(NULL, prevlen);
        prevoffset = p - zl;
        p += rawlen;
        extra += delta;
        cnt++;
    }

    /* Extra bytes is zero all update has been done(or no need to update). */
    // 如果extra为0说明已经更新完成（或者不需要更新）
    if (extra == 0) return zl;

    /* Update tail offset after loop. */
    // 更新尾节点偏移
    if (tail == zl + prevoffset) {
        /* When the the last entry we need to update is also the tail, update tail offset
         * unless this is the only entry that was updated (so the tail offset didn't change). */
        // 如果我们需要更新的最后一个节点也是尾节点，就需要更新偏移
        if (extra - delta != 0) {
            ZIPLIST_TAIL_OFFSET(zl) =
                intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+extra-delta);
        }
    } else {
        /* Update the tail offset in cases where the last entry we updated is not the tail. */
        // 不是尾节点时，更新偏移
        ZIPLIST_TAIL_OFFSET(zl) =
            intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+extra);
    }

    /* Now "p" points at the first unchanged byte in original ziplist,
     * move data after that to new ziplist. */
    offset = p - zl;
    zl = ziplistResize(zl, curlen + extra);
    p = zl + offset;
    memmove(p + extra, p, curlen - offset - 1);
    p += extra;

    /* Iterate all entries that need to be updated tail to head. */
    // 从尾到头遍历所有需要被更新的节点
    while (cnt) {
        zipEntry(zl + prevoffset, &cur); /* no need for "safe" variant since we already iterated on all these entries above. */
        rawlen = cur.headersize + cur.len;
        /* Move entry to tail and reset prevlen. */
        memmove(p - (rawlen - cur.prevrawlensize), 
                zl + prevoffset + cur.prevrawlensize, 
                rawlen - cur.prevrawlensize);
        p -= (rawlen + delta);
        if (cur.prevrawlen == 0) {
            /* "cur" is the previous head entry, update its prevlen with firstentrylen. */
            zipStorePrevEntryLength(p, firstentrylen);
        } else {
            /* An entry's prevlen can only increment 4 bytes. */
            zipStorePrevEntryLength(p, cur.prevrawlen+delta);
        }
        /* Foward to previous entry. */
        prevoffset -= cur.prevrawlen;
        cnt--;
    }
    return zl;
}

在最坏情况下，我们需要遍历每一个节点并更新，所以最坏的时间复杂度为O(n^2)。当然，这种情况很难发生，但即使不是最坏情况，也有可能造成多次级联更新，在更新时，需要重新分配内存空间并进行拷贝等操作，影响ziplist的性能。这就是redis在新版本引入quicklist的原因之一。

经过如上分析，可以发现ziplist适用于节点数量较小的情形。由于ziplist在内存中的存储是连续的，当数据规模增大时，可能没有足够的连续内存空间可供分配，并且一旦发生级联更新，十分影响效率。

话虽如此，ziplist这种数据结构对于内存的极致分配和使用是十分值得借鉴的。

2.3 quicklist

在redis3.2版本，引入了该数据结构，它实际上就是双向链表和压缩链表的组合。由于ziplist不宜太大，所以在quicklist中限制了这一点，quicklist本质上还是一个链表，链表中的每个元素都是一个ziplist。

在看源码前，要补充一个C语言的语法——bit field（位域）。假设有一个结构体car，它有两个成员，start=1表示正在行驶，start=0表示停止，broken=1表示故障，broken=0表示正常：

1
2
3
4
struct car{
    int start;
    int broken;
}car;

很明显，它的大小为8字节。但是实际上，我们只用0和1来代表一些事物的状态，使用8字节来存储是“大材小用”，因此C语言提供了更加高效利用内存的语法：

1
2
3
4
5
6
7
8
9
10
11
struct car{
    int start:1;
    int stop:1;
}car;

int main() {

    struct car a;
    printf("%zd", sizeof(a));
    return 0;
}

在上述代码中，冒号:用于把字节中的二进制位划分为几个不同的区域，后面的数字用于定义每个区域的位数。我们将start和stop的划分在同一个字节的不同位域，因此打印结果为4字节。不过位域的类型只能为int或unsigned int。

在上面的例子中，还是有一些字节未被利用。更进一步，我们可以使用更小的类型：

1
2
3
4
5
struct car{
    uint8_t start:1;
    uint8_t stop:1;
    uint8_t other:6;
}car;

其中使用了1字节的整型，它的二进制位被划分为3个位域：

1
2
3
[xxxxxx]-[x]-[x]
   |      |   | 
 other  stop start

在使用上，位域与结构成员的使用相同，都可以使用指针操作，不过使用指针时，由于指针的步长是根据类型增加的，所以无法使用像sds的负指针技术。

好了，现在可以看quicklist的节点的定义了：

1
2
3
4
5
6
7
8
9
10
11
12
13
// quicklistNode是一个32字节的结构，使用了bit field
typedef struct quicklistNode {
    struct quicklistNode *prev;
    struct quicklistNode *next;
    unsigned char *zl;
    unsigned int sz;             /* ziplist size in bytes */
    unsigned int count : 16;     /* count of items in ziplist */
    unsigned int encoding : 2;   /* RAW==1 or LZF==2 */
    unsigned int container : 2;  /* NONE==1 or ZIPLIST==2 */
    unsigned int recompress : 1; /* was this node previous compressed? */
    unsigned int attempted_compress : 1; /* node can't compress; too small */
    unsigned int extra : 10; /* more bits to steal for future usage */
} quicklistNode;

其中：

  • prev和next分别为双向链表的两个指针，各占8字节
  • zl是指向ziplist的指针，占8字节
  • sz是ziplist的大小，4字节
  • 剩下的4字节，被划分为如下几个位域：
    • count表示ziplist里的元素数量，16bit
    • encoding，编码形式，原生存储（01）还是使用LZF压缩存储（10），2bit
    • container，表示节点是否直接存数据（01）还是使用ziplist存数据（10），本意应该是节点作为一个数据容器，不过目前只使用2，2bit
    • recompress用于标记这个节点之前是否被压缩过，1bit
    • attempted_compress，节点是否太小无法压缩，这个成员在测试中使用，1bit
    • extra，预留位，今后使用

在这里提到了LZF压缩，说明redis会对ziplist进行压缩处理，看下面的结构体：

1
2
3
4
typedef struct quicklistLZF {
    unsigned int sz; /* LZF size in bytes*/
    char compressed[];
} quicklistLZF;

quicklistLZF是一个4+n字节的结构体，是压缩后的ziplist，sz是压缩后的长度，compressed是压缩后的数据。quicklist节点的zl在压缩之前指向ziplist，在ziplist压缩后，zl便指向quicklistLZF。

之后是quicklist结构体：

1
2
3
4
5
6
7
8
9
10
11
// 在64位机器上，它是一个40字节的结构体
typedef struct quicklist {
    quicklistNode *head;
    quicklistNode *tail;
    unsigned long count;        /* total count of all entries in all ziplists */
    unsigned long len;          /* number of quicklistNodes */
    int fill : QL_FILL_BITS;              /* fill factor for individual nodes */
    unsigned int compress : QL_COMP_BITS; /* depth of end nodes not to compress;0=off */
    unsigned int bookmark_count: QL_BM_BITS;
    quicklistBookmark bookmarks[];
} quicklist;

其中：

  • head和tail分别指向整个列表的头节点和尾节点，各8字节
  • count是所有entries元素的总数量，8字节
  • len是Node节点的数量，8字节
  • fill，填充因子，14bit或16bit，取决于32位还是64位机器
  • compress压缩节点的深度，14bit或16bit，取决于32位还是64位机器
  • bookmark_count书签的个数，4bit
  • bookmarks书签，伸缩性成员，初始不占内存

这里又多了一些概念，比如填充因子、压缩深度、书签等，其实暂时不需要了解这些，先把大致结构理清即可。

整个quicklist结构示意图如下：

img

这种数据结构控制了ziplist的大小，可以尽量避免级联更新，但由于还是使用ziplist，所以并没有完全解决级联更新的问题。

2.4 quicklistPushHead

和双向链表一样，quicklist可以在头部和尾部进行插入，下面是头部插入的源码，它将大小为sz的value插入到quicklist指向的快速列表中：

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
int quicklistPushHead(quicklist *quicklist, void *value, size_t sz) {
    quicklistNode *orig_head = quicklist->head;
    assert(sz < UINT32_MAX); /* TODO: add support for quicklist nodes that are sds encoded (not zipped) */
    // 判断是否能够直接插入到现有的ziplist中
    if (likely(_quicklistNodeAllowInsert(quicklist->head, quicklist->fill, sz))) {
        // 如果大小没有超过限制，直接ziplistPush插入
        quicklist->head->zl = ziplistPush(quicklist->head->zl, value, sz, ZIPLIST_HEAD);
        // 更新size
        quicklistNodeUpdateSz(quicklist->head);
    } else {
        // 大小超过限制，需要新增一个节点
        quicklistNode *node = quicklistCreateNode();
        // 把当前value值插入到新节点
        node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);
        // 更新size
        quicklistNodeUpdateSz(node);
        // 把新节点插入到head前
        _quicklistInsertNodeBefore(quicklist, quicklist->head, node);
    }
    // 更新计数
    quicklist->count++;
    quicklist->head->count++;
    // 如果没有创建新节点，就返回0，否则返回1
    return (orig_head != quicklist->head);
}

likely和unlikely是编译层面的优化，它不会改变判断的结果。加likely的意思是后面的值为真的可能性更大一些，那么执行if的机会大，而unlikely表示值为假的可能性大一些，执行else机会大一些。加上这种修饰，编译成二进制代码时likely使得if后面的执行语句紧跟着前面的程序，unlikely使得else后面的语句紧跟着前面的程序，这样就会被cache预读取，增加程序的执行速度。

2.5 _quicklistNodeAllowInsert

这个函数用于检查node节点还能否容纳sz大小的值：

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
REDIS_STATIC int _quicklistNodeAllowInsert(const quicklistNode *node,
                                           const int fill, const size_t sz) {
    if (unlikely(!node))
        // node为空，返回0
        return 0;

    int ziplist_overhead;
    /* size of previous offset */
    // 判断ziplist的prevlen长度
    if (sz < 254)
        ziplist_overhead = 1;
    else
        ziplist_overhead = 5;

    /* size of forward offset */
    // 判断encoding的长度
    if (sz < 64)
        // 长度小于64，encoding为1字节
        ziplist_overhead += 1;
    else if (likely(sz < 16384))
        // 长度小于16384，encoding为2字节
        ziplist_overhead += 2;
    else
        // 否则encoding为5字节
        ziplist_overhead += 5;

    /* new_sz overestimates if 'sz' encodes to an integer type */
    // encoding在数字类型上有一些提前量（原本可能只需要1字节，这里直接给5字节）
    // new_sz=原本的大小+头部大小+数据大小
    unsigned int new_sz = node->sz + sz + ziplist_overhead;
    if (likely(_quicklistNodeSizeMeetsOptimizationRequirement(new_sz, fill)))
        // 按照填充因子检测大小是否满足要求
        return 1;
    /* when we return 1 above we know that the limit is a size limit (which is
     * safe, see comments next to optimization_level and SIZE_SAFETY_LIMIT) */
    // 当上面返回1，说明ziplist的限制是根据占用内存大小进行限制的
    // 返回0，则说明超限了，或者ziplist是根据元素个数限制的
    // 此时也会检查内存是否超出SIZE_SAFETY_LIMIT（默认为8192）
    else if (!sizeMeetsSafetyLimit(new_sz))
        // 内存超出默认限制
        return 0;
    else if ((int)node->count < fill)
        // 内存没有超出默认的限制，那么按照个数限制进行检查
        // 此时节点中的数据个数小于填充因子的限制，返回1
        return 1;
    else
        // 否则返回0
        return 0;
}

这里要介绍填充因子fill了，这个字段用于限制ziplist的内存大小，如果fill为负数：

  • 设置为-1，代表每个节点的ziplist大小不能超过4k字节
  • 设置为-2，代表每个节点的ziplist大小不能超过8k字节
  • 设置为-3，代表每个节点的ziplist大小不能超过16k字节
  • 设置为-4，代表每个节点的ziplist大小不能超过32k字节
  • 设置为-5，代表每个节点的ziplist大小不能超过64k字节

在源码中是这样定义的：

1
2
3
4
/* Optimization levels for size-based filling.
 * Note that the largest possible limit is 16k, so even if each record takes
 * just one byte, it still won't overflow the 16 bit count field. */
static const size_t optimization_level[] = {4096, 8192, 16384, 32768, 65536};

如果fill为正数，则表示按照ziplist中元素个数来进行限制，比如设置为5，则ziplist中不能超过5个元素。

通过这个值，就可以限制ziplist的大小，在一定程度上减少了级联更新带来的效率降低问题。

2.6 ziplistPush

头部/尾部插入节点，一旦经过检查，允许插入，就会调用该函数进行插入：

1
2
3
4
5
unsigned char *ziplistPush(unsigned char *zl, unsigned char *s, unsigned int slen, int where) {
    unsigned char *p;
    p = (where == ZIPLIST_HEAD) ? ZIPLIST_ENTRY_HEAD(zl) : ZIPLIST_ENTRY_END(zl);
    return __ziplistInsert(zl,p,s,slen);
}

它将判断where的值，获取到zl的entry指针，将长度为slen的元素s插入到ziplist，这里的__ziplistInsert就是调用了ziplist.c的函数，具体逻辑已经在在2.2节中介绍过了。

2.7 quicklistPop

与插入对应的是删除操作：

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
int quicklistPop(quicklist *quicklist, int where, unsigned char **data,
                 unsigned int *sz, long long *slong) {
    unsigned char *vstr;
    unsigned int vlen;
    long long vlong;
    if (quicklist->count == 0)
        // 如果quicklist没有元素，直接返回0
        return 0;
    // 调用该函数进行操作，_quicklistSaver是一个函数指针
    int ret = quicklistPopCustom(quicklist, where, &vstr, &vlen, &vlong,
                                 _quicklistSaver);
    if (data)
        // 待删除节点是字符串值，用data保存
        *data = vstr;
    if (slong)
        // 待删除节点是整数值，用slong保存
        *slong = vlong;
    if (sz)
        *sz = vlen;
    return ret;
}

// 这个函数用于将sz大小的数据data进行深拷贝，目的是防止出现两次free报错
REDIS_STATIC void *_quicklistSaver(unsigned char *data, unsigned int sz) {
    unsigned char *vstr;
    if (data) {
        vstr = zmalloc(sz);
        memcpy(vstr, data, sz);
        return vstr;
    }
    return NULL;
}

2.8 quicklistPopCustom

删除逻辑调用的是该函数：

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
// 从quicklist中删除
// 如果节点是数字，则把值保存在sval中，如果是字符串，通过saver函数深拷贝后保存到data
// 如果没有删除则返回0，如果删除成功则返回1
int quicklistPopCustom(quicklist *quicklist, int where, unsigned char **data,
                       unsigned int *sz, long long *sval,
                       void *(*saver)(unsigned char *data, unsigned int sz)) {
    unsigned char *p;
    unsigned char *vstr;
    unsigned int vlen;
    long long vlong;
    // 删除头部pos为1，删除尾部为-1
    int pos = (where == QUICKLIST_HEAD) ? 0 : -1;

    if (quicklist->count == 0)
        // 如果没有元素，直接返回0
        return 0;

    if (data)
        *data = NULL;
    if (sz)
        *sz = 0;
    if (sval)
        *sval = -123456789;

    quicklistNode *node;
    if (where == QUICKLIST_HEAD && quicklist->head) {
        // 获取头部node节点
        node = quicklist->head;
    } else if (where == QUICKLIST_TAIL && quicklist->tail) {
        // 否则获取尾部node节点
        node = quicklist->tail;
    } else {
        return 0;
    }
    // 获取ziplist中的节点p
    p = ziplistIndex(node->zl, pos);
    // 获取p节点值，如果p指向ziplist末尾（zlend）返回0，否则返回1
    if (ziplistGet(p, &vstr, &vlen, &vlong)) {
        if (vstr) {
            // 如果是字符串
            if (data)
                // 通过saver函数深拷贝后保存到data
                *data = saver(vstr, vlen);
            if (sz)
                // 保存长度
                *sz = vlen;
        } else {
            // 如果是整数
            if (data)
                // data为NULL
                *data = NULL;
            if (sval)
                // 用sval保存值
                *sval = vlong;
        }
        // 最后删除该元素
        quicklistDelIndex(quicklist, node, &p);
        return 1;
    }
    // 如果p指向ziplist末尾，没有值，返回0
    return 0;
}

2.9 quicklistDelIndex

这个函数就是用于在node中删除ziplist的entry节点，底层调用的是ziplist.c中的相关函数，删除也有可能造成级联更新。

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
// 从quicklist的节点node中删除一个p指针指向的entry
// 这个函数必须删除的是非压缩节点中的entry
// 如果整个node都因此被删除，返回1，如果删除后还保留node，返回0
REDIS_STATIC int quicklistDelIndex(quicklist *quicklist, quicklistNode *node,
                                   unsigned char **p) {
    // 标记整个节点是否因此删除
    int gone = 0;
    
    // 调用ziplist对应函数删除ziplist中节点p指向的entry
    node->zl = ziplistDelete(node->zl, p);
    // 更新元素个数
    node->count--;
    if (node->count == 0) {
        // 如果节点中元素个数为0.则删除该节点
        gone = 1;
        __quicklistDelNode(quicklist, node);
    } else {
        // 否则更新节点
        quicklistNodeUpdateSz(node);
    }
    // 更新quicklist的元素个数
    quicklist->count--;
    /* If we deleted the node, the original node is no longer valid */
    // 删掉node节点返回1，否则返回0
    return gone ? 1 : 0;
}

2.10 _quicklistInsert

尽管平时更多的是在两端进行操作，但不可避免地会有随机插入的需求，为了方便从任意位置插入数据，和压缩链表的entry相同，quicklist中也有类似的数据结构。

1
2
3
4
5
6
7
8
9
typedef struct quicklistEntry {
    const quicklist *quicklist;
    quicklistNode *node;
    unsigned char *zi;
    unsigned char *value;
    long long longval;
    unsigned int sz;
    int offset;
} quicklistEntry;

其中：

  • quicklist指向这个entry所在的quicklist
  • node指向这个entry节点所在的node节点
  • zi指向entry节点在node中的位置
  • value用来保存字符串数据
  • longval用来保存整数，如果entry存储的是整数，那么会从压缩链表中解压出这个整数保存在其中
  • sz保存数据长度
  • offset表示当前entry在当前节点ziplist中的偏移，比如offset=3代表ziplist的第3个元素

对于插入，可以插入到entry前面，也就是_quicklistInsertNodeBefore：

1
2
3
4
void quicklistInsertBefore(quicklist *quicklist, quicklistEntry *entry,
                           void *value, const size_t sz) {
    _quicklistInsert(quicklist, entry, value, sz, 0);
}

与之对应的还有quicklistInsertAfter：

1
2
3
4
void quicklistInsertAfter(quicklist *quicklist, quicklistEntry *entry,
                          void *value, const size_t sz) {
    _quicklistInsert(quicklist, entry, value, sz, 1);
}

它们的区别只是插入位置的不同，都调用_quicklistInsert，函数比较长，但逻辑清晰：

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
// 在现有的entry前/后插入一个新的entry
// 如果after为1，则插入到后面，如果为0插入到前面
REDIS_STATIC void _quicklistInsert(quicklist *quicklist, quicklistEntry *entry,
                                   void *value, const size_t sz, int after) {
    //full用来标记插入后是否会超出填充因子fill的限制
    //at_tail用于标记要插入的位置是否在对应的ziplist的末尾
    //at_head用于标记要插入的位置是否在对应的ziplist的开头
    //full_next用于标记下一个node中的ziplist能否插入
    //full_next用于标记上一个node中的ziplist能否插入
    int full = 0, at_tail = 0, at_head = 0, full_next = 0, full_prev = 0;
    // 获得填充因子
    int fill = quicklist->fill;
    // 获得节点
    quicklistNode *node = entry->node;
    quicklistNode *new_node = NULL;
    assert(sz < UINT32_MAX); /* TODO: add support for quicklist nodes that are sds encoded (not zipped) */

    if (!node) {
        // 如果node为空，新建node，把元素插入到该node中并更新计数
        /* we have no reference node, so let's create only node in the list */
        D("No node given!");
        new_node = quicklistCreateNode();
        new_node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);
        __quicklistInsertNode(quicklist, NULL, new_node, after);
        new_node->count++;
        quicklist->count++;
        return;
    }

    /* Populate accounting flags for easier boolean checks later */
    if (!_quicklistNodeAllowInsert(node, fill, sz)) {
        D("Current node is full with count %d with requested fill %lu",
          node->count, fill);
        // 检查是否会超出限制，如果超出就将full标志置1，方便后续判断
        full = 1;
    }

    if (after && (entry->offset == node->count)) {
        // after为1（在后面插入），并且要插入到当前节点的尾部
        D("At Tail of current ziplist");
        // 将at_tail置1
        at_tail = 1;
        // 检查下一个节点能否插入
        if (!_quicklistNodeAllowInsert(node->next, fill, sz)) {
            // 如果不能插入，说明下一个节点也满了
            D("Next node is full too.");
            // 将full_next置1
            full_next = 1;
        }
    }

    if (!after && (entry->offset == 0)) {
        // 如果after为0（在前面插入），并且插入到当前节点的头部
        D("At Head");
        at_head = 1;
        // 检查上一个节点
        if (!_quicklistNodeAllowInsert(node->prev, fill, sz)) {
            D("Prev node is full too.");
            // 如果上一节点不能插入，将full_prev置1
            full_prev = 1;
        }
    }

    /* Now determine where and how to insert the new element */
    // 现在标志位已经更新，开始判断到底应该如何插入新元素
    if (!full && after) {
        // 如果当前节点可以插入，并且after为1，则直接插入到后面
        D("Not full, inserting after current position.");
        // 可能需要解压node
        quicklistDecompressNodeForUse(node);
        // 获得ziplist下一个元素（如果没有下一个元素就返回NULL）
        unsigned char *next = ziplistNext(node->zl, entry->zi);
        if (next == NULL) {
            // 如果没有下一个元素，就直接将新元素插入到ziplist尾部，底层调用的是__ziplistInsert（2.2节介绍过）
            node->zl = ziplistPush(node->zl, value, sz, ZIPLIST_TAIL);
        } else {
            // 否则将next整体后移，把新元素插入到next所指位置，底层调用的是__ziplistInsert（2.2节介绍过）
            node->zl = ziplistInsert(node->zl, next, value, sz);
        }
        // 更新相关信息
        node->count++;
        quicklistNodeUpdateSz(node);
        // 如果之前经过解压，这里重新压缩
        quicklistRecompressOnly(quicklist, node);
    } else if (!full && !after) {
        // 如果当前节点可以插入，并且after为0，则直接插入到前面，逻辑和上面相同
        D("Not full, inserting before current position.");
        quicklistDecompressNodeForUse(node);
        node->zl = ziplistInsert(node->zl, entry->zi, value, sz);
        node->count++;
        quicklistNodeUpdateSz(node);
        quicklistRecompressOnly(quicklist, node);
    } else if (full && at_tail && node->next && !full_next && after) {
        // 如果当前节点不可以插入、并且要插入到当前节点的尾部、下一个节点存在、下一个节点可以插入
        // 那么就把元素插入到下一个节点的最前面
        /* If we are: at tail, next has free space, and inserting after:
         *   - insert entry at head of next node. */
        D("Full and tail, but next isn't full; inserting next node head");
        new_node = node->next;
        quicklistDecompressNodeForUse(new_node);
        new_node->zl = ziplistPush(new_node->zl, value, sz, ZIPLIST_HEAD);
        new_node->count++;
        quicklistNodeUpdateSz(new_node);
        quicklistRecompressOnly(quicklist, new_node);
    } else if (full && at_head && node->prev && !full_prev && !after) {
        // 如果当前节点不可以插入、并且要插入到当前节点的头部、上一个节点存在、上一个节点可以插入
        // 那么就把元素插入到上一个节点的最后面
        /* If we are: at head, previous has free space, and inserting before:
         *   - insert entry at tail of previous node. */
        D("Full and head, but prev isn't full, inserting prev node tail");
        new_node = node->prev;
        quicklistDecompressNodeForUse(new_node);
        new_node->zl = ziplistPush(new_node->zl, value, sz, ZIPLIST_TAIL);
        new_node->count++;
        quicklistNodeUpdateSz(new_node);
        quicklistRecompressOnly(quicklist, new_node);
    } else if (full && ((at_tail && node->next && full_next && after) ||
                        (at_head && node->prev && full_prev && !after))) {
        // 如果当前节点不可以插入、要插入到当前节点的头部/尾部，上一个节点或下一个节点存在，但是都不可以插入
        // 则创建新节点，把元素插入到新节点中
        /* If we are: full, and our prev/next is full, then:
         *   - create new node and attach to quicklist */
        D("\tprovisioning new node...");
        new_node = quicklistCreateNode();
        new_node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);
        new_node->count++;
        quicklistNodeUpdateSz(new_node);
        // 根据after，把新节点插入到现节点的前/后面
        __quicklistInsertNode(quicklist, node, new_node, after);
    } else if (full) {
        // 否则，要插入的位置不是头部也不是尾部，而是ziplist的中间位置
        // 此时需要拆分节点
        /* else, node is full we need to split it. */
        /* covers both after and !after cases */
        D("\tsplitting node...");
        quicklistDecompressNodeForUse(node);
        // 根据after把节点拆分
        new_node = _quicklistSplitNode(node, entry->offset, after);
        // 把元素插入到新节点，根据after插入到头/尾部
        new_node->zl = ziplistPush(new_node->zl, value, sz,
                                   after ? ZIPLIST_HEAD : ZIPLIST_TAIL);
        // 更新相关信息
        new_node->count++;
        quicklistNodeUpdateSz(new_node);
        __quicklistInsertNode(quicklist, node, new_node, after);
        // 最后，尝试合并节点
        _quicklistMergeNodes(quicklist, node);
    }

    quicklist->count++;
}

其中，比较难理解的是拆分和合并操作，下面分别介绍。

2.11 _quicklistSplitNode

将一个节点拆分为两部分：

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
// 把node拆分为两部分，由offset和after控制，假设整个节点为：[0,...,OFFSET,...END]
// after控制拆分后返回哪部分：
//      如果为1，则返回部分为[OFFSET+1, END]
//      如果为0，则返回部分为[0, OFFSET-1]
REDIS_STATIC quicklistNode *_quicklistSplitNode(quicklistNode *node, int offset,
                                                int after) {
    size_t zl_sz = node->sz;
    // 创建新节点，分配内存并拷贝ziplist
    quicklistNode *new_node = quicklistCreateNode();
    new_node->zl = zmalloc(zl_sz);

    /* Copy original ziplist so we can split it */
    memcpy(new_node->zl, node->zl, zl_sz);

    /* Ranges to be trimmed: -1 here means "continue deleting until the list ends" */
    // 划分删除范围
    int orig_start = after ? offset + 1 : 0;
    int orig_extent = after ? -1 : offset;
    int new_start = after ? 0 : offset;
    int new_extent = after ? offset + 1 : -1;

    D("After %d (%d); ranges: [%d, %d], [%d, %d]", after, offset, orig_start,
      orig_extent, new_start, new_extent);
    // 原节点（input node）示意图：[0,...,offset,...END]
    // 原节点修剪
    // 如果after为1，删除后剩余：[0,offset]
    // 如果after为0，删除后剩余：[offset, END]
    node->zl = ziplistDeleteRange(node->zl, orig_start, orig_extent); 
    node->count = ziplistLen(node->zl);
    quicklistNodeUpdateSz(node);
    // 新节点（return node）示意图：[0,...,offset,...END]
    // 新节点修剪
    // 如果after为1，删除后剩余：[offset+1,END]
    // 如果after为0，删除后剩余：[0,offset-1]
    new_node->zl = ziplistDeleteRange(new_node->zl, new_start, new_extent);
    new_node->count = ziplistLen(new_node->zl);
    quicklistNodeUpdateSz(new_node);

    D("After split lengths: orig (%d), new (%d)", node->count, new_node->count);
    // 返回新节点
    return new_node;
}

2.12 _quicklistMergeNodes

合并节点：

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
// 尝试合并center节点附近的两个ziplist
// 尝试如下合并：
//   - (center->prev->prev, center->prev)
//   - (center->next, center->next->next)
//   - (center->prev, center)
//   - (center, center->next)
REDIS_STATIC void _quicklistMergeNodes(quicklist *quicklist,
                                       quicklistNode *center) {
    // 获取相关参数，创建指针
    int fill = quicklist->fill;
    quicklistNode *prev, *prev_prev, *next, *next_next, *target;
    prev = prev_prev = next = next_next = target = NULL;
    // 初始化指针，方便后续操作
    if (center->prev) {
        // 如果center前面有节点
        prev = center->prev;
        if (center->prev->prev)
            // 如果center前面的前面有节点
            prev_prev = center->prev->prev;
    }

    if (center->next) {
        // center后面后节点
        next = center->next;
        if (center->next->next)
            // center后面的后面有节点
            next_next = center->next->next;
    }

    /* Try to merge prev_prev and prev */
    // 尝试将center前节点与前前节点合并，检查合并后是否超过填充因子的限制
    // _quicklistNodeAllowMerge函数的整个逻辑与前面介绍的插入检查_quicklistNodeAllowInsert函数基本相同
    if (_quicklistNodeAllowMerge(prev, prev_prev, fill)) {
        // 合并
        _quicklistZiplistMerge(quicklist, prev_prev, prev);
        prev_prev = prev = NULL; /* they could have moved, invalidate them. */
    }

    /* Try to merge next and next_next */
    // 尝试将center后节点与后后节点合并，检查合并后是否超过填充因子的限制
    if (_quicklistNodeAllowMerge(next, next_next, fill)) {
        _quicklistZiplistMerge(quicklist, next, next_next);
        next = next_next = NULL; /* they could have moved, invalidate them. */
    }

    /* Try to merge center node and previous node */
    // 尝试将center节点与前节点合并，检查合并后是否超过填充因子的限制
    if (_quicklistNodeAllowMerge(center, center->prev, fill)) {
        target = _quicklistZiplistMerge(quicklist, center->prev, center);
        center = NULL; /* center could have been deleted, invalidate it. */
    } else {
        /* else, we didn't merge here, but target needs to be valid below. */
        target = center;
    }

    /* Use result of center merge (or original) to merge with next node. */
    // 尝试将center节点与后节点合并，检查合并后是否超过填充因子的限制
    if (_quicklistNodeAllowMerge(target, target->next, fill)) {
        _quicklistZiplistMerge(quicklist, target, target->next);
    }
}

以上是quicklist相关函数的介绍，redis为其还提供了更多的函数，比如quicklistDup用于拷贝quicklist，quicklistRotate反转quicklist等等，这里就不展开了。

3 字典

3.1 数据结构

字典也就是哈希表是一种基于键值存储的数据结构，源码位于./src/dict.c以及./src/dict.h，它的数据结构如下：

1
2
3
4
5
6
7
typedef struct dict {
    dictType *type;
    void *privdata;
    dictht ht[2];
    long rehashidx; /* rehashing not in progress if rehashidx == -1 */
    int16_t pauserehash; /* If >0 rehashing is paused (<0 indicates coding error) */
} dict;

其中：

  • type是一个指向 dictType 结构体的指针，这个结构体存储了一些基本操作的函数指针
  • privdata则保存了传给上述函数的一些可选参数
  • ht是一个dictht结构体数组，这就是存储用的哈希表，一个用于存放键值，另一个用于rehash
  • rehashidx为-1时，不进行rehash操作
  • pauserehash暂停rehash

这个结构体包含了许多子结构，首先是dictType：

1
2
3
4
5
6
7
8
9
typedef struct dictType {
    uint64_t (*hashFunction)(const void *key); //计算哈希值的函数 
    void *(*keyDup)(void *privdata, const void *key);//复制键的函数
    void *(*valDup)(void *privdata, const void *obj);//复制值的函数
    int (*keyCompare)(void *privdata, const void *key1, const void *key2);//对比键的函数
    void (*keyDestructor)(void *privdata, void *key); //销毁键的函数
    void (*valDestructor)(void *privdata, void *obj); //销毁值的函数
    int (*expandAllowed)(size_t moreMem, double usedRatio);//检查是否允许分配内存的函数
} dictType;

最核心的是dictht结构体：

1
2
3
4
5
6
typedef struct dictht {
    dictEntry **table;
    unsigned long size;
    unsigned long sizemask;
    unsigned long used;
} dictht;

其中：

  • table是一个二级指针，它是指向dictEntry结构体数组指针的指针
  • size为哈希表的大小
  • sizemask为哈希表大小掩码，用于计算索引值
  • used表示哈希表中已使用的节点数量

下面来看dictEntry结构体：

1
2
3
4
5
6
7
8
9
10
typedef struct dictEntry {
    void *key;
    union {
        void *val;
        uint64_t u64;
        int64_t s64;
        double d;
    } v;
    struct dictEntry *next;
} dictEntry;

其中：

  • key是键
  • v是一个联合，使用此结构可以节省内存空间
    • val用来保存二进制数据
    • u64用来保存64位无符号整型
    • s64用来保存64位有符号整型
    • d用来保存双精度浮点数
  • next为指向下一个dictEntry的指针，说明redis采用拉链法处理冲突，为了保证插入时间复杂度为O(1)，冲突时采用头插法

下图是字典的数据结构示意图：

image-20220306153851718

3.2 创建字典

dictCreate用于创建字典：

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
dict *dictCreate(dictType *type,void *privDataPtr){
    dict *d = zmalloc(sizeof(*d));
	// _dictInit初始化字典
    _dictInit(d,type,privDataPtr);
    return d;
}

/* Initialize the hash table */
int _dictInit(dict *d, dictType *type,void *privDataPtr){
    // 重置ht
    _dictReset(&d->ht[0]);
    _dictReset(&d->ht[1]);
    // 设置初始值
    d->type = type;
    d->privdata = privDataPtr;
    d->rehashidx = -1;
    d->pauserehash = 0;
    // DICT_OK为0表示成功，DICT_ERR为1表示错误
    return DICT_OK;
}
// 重置字典
static void _dictReset(dictht *ht)
{
    ht->table = NULL;
    ht->size = 0;
    ht->sizemask = 0;
    ht->used = 0;
}

3.3 插入元素

和其它哈希表的实现一样，在redis的哈希表中插入元素肯定要经过哈希函数计算，然后再根据计算出的索引值，将包含新键值对的哈希表节点放到哈希表数组的指定索引上面。redis使用了众多哈希算法（默认使用siphash算法），这部分不是本文的研究范畴，所以后续源码中出现的哈希函数不会深究。重点看一下redis中插入操作的实现：

1
2
3
4
5
6
7
8
9
10
11
12
/* Add an element to the target hash table */
int dictAdd(dict *d, void *key, void *val)
{
    // 添加一个dictEntry，并为其设置key，不设置value
    dictEntry *entry = dictAddRaw(d,key,NULL);

    if (!entry) return DICT_ERR;// 如果entry为空返回错误-1
    // 为entry设置value，通过宏定义设置，如果dictType中有对应的函数，会调用这个函数设置value
    // 如果没有对应的函数，直接进行赋值设置
    dictSetVal(d, entry, val);
    return DICT_OK;
}

调用dictAddRaw：

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
// 该函数对指定key进行哈希计算后，向d中添加一个dictEntry
// 如果插入成功，返回对应的dictEntry指针
// 如果key对应的dictEntry已经存在，且传入了existing，就会将这个dictEntry赋值给existing
dictEntry *dictAddRaw(dict *d, void *key, dictEntry **existing)
{
    long index;
    dictEntry *entry;
    dictht *ht;
    // 如果d中rehashidx不等于-1则进行rehash相关操作
    if (dictIsRehashing(d)) _dictRehashStep(d);

    /* Get the index of the new element, or -1 if
     * the element already exists. */
    // 首先使用dictHashKey获取hash，然后调用_dictKeyIndex获取这个key对应的索引
    if ((index = _dictKeyIndex(d, key, dictHashKey(d,key), existing)) == -1)
        // 如果这个key已经存在，index=-1，返回NULL
        return NULL;

    /* Allocate the memory and store the new entry.
     * Insert the element in top, with the assumption that in a database
     * system it is more likely that recently added entries are accessed
     * more frequently. */
    // 如果未重复，为entry新分配内存，以及更新计数等
    // 如果处于rehash状态，则新的键值会被插入到第二个哈希表中，否则插入到第一个哈希表中
    ht = dictIsRehashing(d) ? &d->ht[1] : &d->ht[0];
    entry = zmalloc(sizeof(*entry));
    entry->next = ht->table[index];
    ht->table[index] = entry;
    ht->used++;

    /* Set the hash entry fields. */
    // 设置key，但不设置value
    dictSetKey(d, entry, key);
    return entry;
}

其中，_dictKeyIndex函数是基本的底层函数，它的作用是根据key和hash，返回这个key应该放在哪个索引index下：

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
static long _dictKeyIndex(dict *d, const void *key, uint64_t hash, dictEntry **existing)
{
    unsigned long idx, table;
    dictEntry *he;
    if (existing) *existing = NULL;

    /* Expand the hash table if needed */
    // 扩大哈希表
    if (_dictExpandIfNeeded(d) == DICT_ERR)
        return -1;
    // 遍历两个哈希表
    for (table = 0; table <= 1; table++) {
        // 将hash与sizemask按位与获得index
        idx = hash & d->ht[table].sizemask;
        /* Search if this slot does not already contain the given key */
        he = d->ht[table].table[idx];
        // he为对应的dictEntry指针
        while(he) {
            // 使用值判断，或根据dictType中对应的函数判断两个key是否相等
            if (key==he->key || dictCompareKeys(d, key, he->key)) {
                // 如果相等，说明存在该key，如果传入了existing就把这个指针赋值给existing
                if (existing) *existing = he;
                // 返回-1代表已经存在
                return -1;
            }
            // 拉链法继续向后探测
            he = he->next;
        }
        // 如果没有进行rehash，在第一个表中操作，否则在第二个表中操作
        if (!dictIsRehashing(d)) break;
    }
    // key不存在，返回对应的索引
    return idx;
}

除了dictAdd之外，还有一个dictAddOrFind函数，它是前者的简化版：

1
2
3
4
5
dictEntry *dictAddOrFind(dict *d, void *key) {
    dictEntry *entry, *existing;
    entry = dictAddRaw(d,key,&existing);
    return entry ? entry : existing;
}

3.4 更新元素

更新replace是一个插入或者覆盖的操作，对应的函数为dictReplace：

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
// 插入或覆盖操作
// 添加一个元素，如果键已经存在，则覆盖旧的，返回0；否则插入新的，返回1
int dictReplace(dict *d, void *key, void *val)
{
    dictEntry *entry, *existing, auxentry;

    /* Try to add the element. If the key
     * does not exists dictAdd will succeed. */
    // 调用dictAddRaw尝试获得entry
    entry = dictAddRaw(d,key,&existing);
    if (entry) {
        // entry不为NULL说明是新元素，设置值并返回1
        dictSetVal(d, entry, val);
        return 1;
    }

    /* Set the new value and free the old one. Note that it is important
     * to do that in this order, as the value may just be exactly the same
     * as the previous one. In this context, think to reference counting,
     * you want to increment (set), and then decrement (free), and not the
     * reverse. */
    // 否则说明key已经存在，根据key更新该值，并释放掉旧值
    // 顺序很重要，由于新的key-value可能与旧的相同，在这种情况下考虑引用计数
    // 必须先增加，然后再释放掉旧的
    auxentry = *existing;
    dictSetVal(d, existing, val);
    dictFreeVal(d, &auxentry);
    return 0;
}

3.5 删除元素

dictGenericDelete是最基本的删除函数，可以由dictDelete调用，传入的第三个参数nofree=0表示直接将找到的entry释放掉：

1
2
3
4
5
/* Remove an element, returning DICT_OK on success or DICT_ERR if the
 * element was not found. */
int dictDelete(dict *ht, const void *key) {
    return dictGenericDelete(ht,key,0) ? DICT_OK : DICT_ERR;
}

而dictUnlink，调用时nofree=1表示如果找到了entry不直接删除，而是将其返回，进行操作后可以使用dictFreeUnlinkedEntry释放它：

1
2
3
dictEntry *dictUnlink(dict *ht, const void *key) {
    return dictGenericDelete(ht,key,1);
}

这样做的好处是只需要一次查找：

1
2
3
4
5
6
7
8
9
// 当我们想对要删除的元素做一些操作再真正释放时，以前需要：
entry = dictFind(...);
// 进行一些操作...
dictDelete(dictionary,entry);

// 有了这个函数，现在可以：
entry = dictUnlink(dictionary,entry);
// 进行一些操作...
dictFreeUnlinkedEntry(entry); // <- 释放的时候不需要再次查找表了

下面来看看基本删除函数：

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
// 在d中查找key对应的元素并删除，nofree为1表示在这个函数中不进行释放，为0表示直接进行释放
static dictEntry *dictGenericDelete(dict *d, const void *key, int nofree) {
    uint64_t h, idx;
    dictEntry *he, *prevHe;
    int table;
    // 如果两个哈希表都为空，尽早返回NULL
    if (d->ht[0].used == 0 && d->ht[1].used == 0) return NULL;
    // 是否进行rehash
    if (dictIsRehashing(d)) _dictRehashStep(d);
    // 获取key对应的hash
    h = dictHashKey(d, key);
    // 遍历两个table
    for (table = 0; table <= 1; table++) {
        // 获取对应的索引idx
        idx = h & d->ht[table].sizemask;
        // 获取entry
        he = d->ht[table].table[idx];
        prevHe = NULL;
        while(he) {
            if (key==he->key || dictCompareKeys(d, key, he->key)) {
                /* Unlink the element from the list */
                // 将该元素从整个表中脱离
                if (prevHe)
                    // from：prev->current->next，to：prev->next
                    prevHe->next = he->next;
                else
                    // 本身就是头结点，直接删除
                    d->ht[table].table[idx] = he->next;
                if (!nofree) {
                    // nofree为0，直接在这里释放
                    dictFreeKey(d, he);
                    dictFreeVal(d, he);
                    zfree(he);
                }
                // 更新计数
                d->ht[table].used--;
                return he;
            }
            // 拉链法寻找next节点
            prevHe = he;
            he = he->next;
        }
        // 没有rehash则结束循环，在第一张表操作
        if (!dictIsRehashing(d)) break;
    }
    // 没有找到，返回NULL
    return NULL; /* not found */
}

3.6 rehash

在前面的源码中，出现过rehash操作，它是什么？随着操作的不断执行，哈希表保存的键值对会逐渐地增多或者减少，为了让哈希表的负载因子（load factor）维持在一个合理的范围之内，当哈希表保存的键值对数量太多或者太少时，程序需要对哈希表的大小进行相应的扩展或者收缩。这个扩展和收缩哈希表的过程叫做rehash（重新散列/重哈希）。在3.1数据结构中，dict中拥有两个哈希表ht[2]，它们两个交替使用。

rehash的工作流程如下：

  • 首先根据ht[0]中的元素，以及需要进行操作判断此时是需要扩容还是缩容，为ht[1]分配空间
  • 将ht[0]中的元素重新计算键的哈希值和索引值，迁移到ht[1]对应的位置上
  • 迁移完成后，ht[0] 变为空表，释放ht[0]，将 ht[1] 设置为 ht[0] ， 然后在 ht[1] 新创建一个空白哈希表， 为下一次 rehash 做准备

具体的细节，我们在源码中查看，首先是什么时候rehash。在3.3插入元素中，介绍了_dictKeyIndex函数，其中在一开始就会调用_dictExpandIfNeeded函数进行判断是否需要进行rehash：

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
// 定义的两个默认值，一个控制是否允许resize，另一个为负载因子的比率，二者有一个满足就会进行rehash
static int dict_can_resize = 1;
static unsigned int dict_force_resize_ratio = 5;

/* Expand the hash table if needed */
static int _dictExpandIfNeeded(dict *d)
{
    /* Incremental rehashing already in progress. Return. */
    // 如果rehash正在进行，返回0
    if (dictIsRehashing(d)) return DICT_OK;

    /* If the hash table is empty expand it to the initial size. */
    // 如果哈希表为空，将其扩容到初始大小（DICT_HT_INITIAL_SIZE默认为4）
    if (d->ht[0].size == 0) return dictExpand(d, DICT_HT_INITIAL_SIZE);

    /* If we reached the 1:1 ratio, and we are allowed to resize the hash
     * table (global setting) or we should avoid it but the ratio between
     * elements/buckets is over the "safe" threshold, we resize doubling
     * the number of buckets. */
    // 首先明确一个概念叫做负载因子：负载因子 = 哈希表已保存节点数量 / 哈希表大小
    // 如果满足：
    // - 哈希表当前元素数量超过表大小size，并且：
    //      dict_can_resize参数为1（默认为1）或者 负载因子大于5（当前元素数量是哈希表大小的5倍以上）
    // 则执行扩容
    if (d->ht[0].used >= d->ht[0].size &&
        (dict_can_resize ||
         d->ht[0].used/d->ht[0].size > dict_force_resize_ratio) &&
        dictTypeExpandAllowed(d))
    {
        return dictExpand(d, d->ht[0].used + 1);
    }
    return DICT_OK;
}

通过源码，可以得出rehash触发的要素：

  • ht[0]为空
  • ht[0]元素数量超过ht[0]表的大小且允许扩容
  • ht[0]元素数量超过ht[0]表的大小，不允许扩容，但是负载因子已经超过5

dict_can_resize这个参数默认为1，那么什么时候会变为0？在./src/server.c中，控制了该参数：

1
2
3
4
5
6
7
8
9
10
11
12
/* This function is called once a background process of some kind terminates,
 * as we want to avoid resizing the hash tables when there is a child in order
 * to play well with copy-on-write (otherwise when a resize happens lots of
 * memory pages are copied). The goal of this function is to update the ability
 * for dict.c to resize the hash tables accordingly to the fact we have an
 * active fork child running. */
void updateDictResizePolicy(void) {
    if (!hasActiveChildProcess())
        dictEnableResize();
    else
        dictDisableResize();
}

意思是：如果此时存在正在进行rdb或者aof持久化时（通过pid判断），该参数会被设置为0，即尽量不允许进行rehash（尽量是因为如果负载因子超过5，仍然会进行）。这样设计的原因是为了避免在持久化时过多的内存复制。

当判断需要进行扩容时，由dictExpand进行rehash：

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
int dictExpand(dict *d, unsigned long size) {
    return _dictExpand(d, size, NULL);
}

// 扩大或者创建哈希表，当第三个参数malloc_failed不为NULL时，会避免返回错误
// 如果成功返回0，否则返回1
int _dictExpand(dict *d, unsigned long size, int* malloc_failed)
{
    if (malloc_failed) *malloc_failed = 0;

    /* the size is invalid if it is smaller than the number of
     * elements already inside the hash table */
    // 如果正在进行rehash，或者传来的size小于元素数量，返回错误
    if (dictIsRehashing(d) || d->ht[0].used > size)
        return DICT_ERR;

    dictht n; /* 声明新的哈希表 */
    // 通过_dictNextPower函数计算新大小
    unsigned long realsize = _dictNextPower(size);

    /* Detect overflows */
    // 溢出检查
    if (realsize < size || realsize * sizeof(dictEntry*) < realsize)
        return DICT_ERR;

    /* Rehashing to the same table size is not useful. */
    // 新大小不能与原来的大小相同
    if (realsize == d->ht[0].size) return DICT_ERR;

    /* Allocate the new hash table and initialize all pointers to NULL */
    // 分配内存，并把所有指针设置为NULL
    n.size = realsize;
    n.sizemask = realsize-1;
    if (malloc_failed) {
        n.table = ztrycalloc(realsize*sizeof(dictEntry*));
        *malloc_failed = n.table == NULL;
        if (*malloc_failed)
            return DICT_ERR;
    } else
        n.table = zcalloc(realsize*sizeof(dictEntry*));

    n.used = 0;

    /* Is this the first initialization? If so it's not really a rehashing
     * we just set the first hash table so that it can accept keys. */
    // 如果是第一次初始化，这并不是真正的rehash
    if (d->ht[0].table == NULL) {
        // 把新表赋值给ht[0]即可
        d->ht[0] = n;
        return DICT_OK;
    }

    /* Prepare a second hash table for incremental rehashing */
    // 把新表赋值给ht[1]
    d->ht[1] = n;
    // rehashidx设置为0，表示从现在开始进行rehash
    d->rehashidx = 0;
    return DICT_OK;
}

具体要扩容多大，由其中的_dictNextPower决定：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
/* Our hash table capability is a power of two */
static unsigned long _dictNextPower(unsigned long size)
{
    unsigned long i = DICT_HT_INITIAL_SIZE;
	// 如果大小已经超过了最大值，返回LONG_MAX+1
    if (size >= LONG_MAX) return LONG_MAX + 1LU;
    // 否则以2的指数增长，直到其首次比size大时返回
    while(1) {
        if (i >= size)
            return i;
        i *= 2;
    }
    // 这也解释了为什么在执行完该函数为什么会判断溢出：
    // 因为有可能一开始size没有超过LONG_MAX，但经过指数增长后，超过LONG_MAX
}

至此，已经完成了对ht[1]的扩容操作，下面就是迁移工作了。

3.7 渐进式rehash

考虑这样的场景，假设当前字典的哈希表结构已经很大，此时进行扩容、计算哈希、迁移等操作时，会变得很慢。为了解决这个问题，redis采用了渐进式rehash策略，即将迁移工作分多次完成。当进行扩容后，仅仅只是为ht[1]分配空间，但不一次性把ht[0]迁移过去，而是在每次进行执行添加、删除、查找或者更新操作时，迁移指定索引上的那些键值对，重复这个过程，慢慢地ht[0]中的数据就迁移到ht[1]上了。具体怎么迁移，口说无凭，通过源码来看。在3.3插入元素小节，介绍的dictAddRaw中，就有这个过程

1
2
3
4
5
6
7
dictEntry *dictAddRaw(dict *d, void *key, dictEntry **existing)
{
	....
    // 如果d中rehashidx不等于-1则进行rehash相关操作
    if (dictIsRehashing(d)) _dictRehashStep(d);
	....
}

由于在扩容后，已经将rehashidx设置为0，表示从现在开始进行rehash，所以会调用_dictRehashStep：

1
2
3
4
static void _dictRehashStep(dict *d) {
    // 第二个参数n=1表示进行一次rehash
    if (d->pauserehash == 0) dictRehash(d,1);
}

因此基础rehash函数如下：

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
// 它接收两个参数，对字典d进行n次rehash操作
int dictRehash(dict *d, int n) {
    // 10*n个空桶访问次数
    int empty_visits = n*10; /* Max number of empty buckets to visit. */
    // 如果rehashidx仍小于0，说明没有进行rehash
    if (!dictIsRehashing(d)) return 0;
    // 循环n次，且每次循环时ht[0]中必须还有键值对
    while(n-- && d->ht[0].used != 0) {
        dictEntry *de, *nextde;

        /* Note that rehashidx can't overflow as we are sure there are more
         * elements because ht[0].used != 0 */
        // rehashidx不能超过size，否则会溢出
        assert(d->ht[0].size > (unsigned long)d->rehashidx);
        while(d->ht[0].table[d->rehashidx] == NULL) {
            // 尝试访问rehashidx位置的桶，如果为空则访问下一个
            d->rehashidx++;
            // 并将空桶访问次数减一，如果累计访问了10*n个空桶，则直接返回1
            // 这是为了防止时间过长导致阻塞
            if (--empty_visits == 0) return 1;
        }
        // 找到不为NULL的桶，de即dictEntry
        de = d->ht[0].table[d->rehashidx];
        /* Move all the keys in this bucket from the old to the new hash HT */
        // 将这个桶中所有的键值对（由于拉链法所以可能不止一个键值对）迁移到ht[1]
        while(de) {
            uint64_t h;

            nextde = de->next;
            /* Get the index in the new hash table */
            // 重新计算在ht[1]中的hash
            h = dictHashKey(d, de->key) & d->ht[1].sizemask;
            // 此时ht[1]中可能有键值，所以要进行头插
            de->next = d->ht[1].table[h];
            d->ht[1].table[h] = de;
            // 更新计数
            d->ht[0].used--;
            d->ht[1].used++;
            de = nextde;
        }
        // 迁移后把ht[0]对应位置设置为NULL
        d->ht[0].table[d->rehashidx] = NULL;
        // 更新索引
        d->rehashidx++;
    }

    /* Check if we already rehashed the whole table... */
    // 检查是否已经把ht[0]全部迁移到ht[1]
    if (d->ht[0].used == 0) {
        // 把ht[0]释放
        zfree(d->ht[0].table);
        // 把ht[1]作为ht[0]
        d->ht[0] = d->ht[1];
        // 重置ht[1]
        _dictReset(&d->ht[1]);
        // 将rehashidx设置为-1，表示rehash结束
        d->rehashidx = -1;
        return 0;
    }

    /* More to rehash... */
    // 否则，说明还未迁移完毕，返回1
    return 1;
}

你以为结束了吗，其实还没有，这种方式还存在问题。由于在渐进式rehash执行期间，字典会同时使用 ht[0] 和 ht[1] 两个哈希表， 字典的删除、查找、更新等操作会在两个哈希表上进行（见前面源码），如果长时间如此，空间消耗比较大。想象一个场景，如果字典刚刚开始执行rehash步骤，此时redis长时间没有接收到相关操作，那么字典会持续占用过多内存，因此，redis还有另一种机制：定时rehash。源码在./src/server.c中：

1
2
3
4
5
6
7
8
9
10
11
void databasesCron(void) {
	....
        /* Rehash */
    if (server.activerehashing) {
        for (j = 0; j < dbs_per_call; j++) {
            int work_done = incrementallyRehash(rehash_db);
	....
            }
        }
    }
}

在incrementallyRehash中：

1
2
3
4
5
6
7
8
9
10
11
12
13
int incrementallyRehash(int dbid) {
    /* Keys dictionary */
    if (dictIsRehashing(server.db[dbid].dict)) {
        dictRehashMilliseconds(server.db[dbid].dict,1);
        return 1; /* already used our millisecond for this loop... */
    }
    /* Expires */
    if (dictIsRehashing(server.db[dbid].expires)) {
        dictRehashMilliseconds(server.db[dbid].expires,1);
        return 1; /* already used our millisecond for this loop... */
    }
    return 0;
}

最终调用dictRehashMilliseconds：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
/* Rehash in ms+"delta" milliseconds. The value of "delta" is larger 
 * than 0, and is smaller than 1 in most cases. The exact upper bound 
 * depends on the running time of dictRehash(d,100).*/
int dictRehashMilliseconds(dict *d, int ms) {
    if (d->pauserehash > 0) return 0;

    long long start = timeInMilliseconds();
    int rehashes = 0;
	// 每次执行100步rehash操作
    while(dictRehash(d,100)) {
        rehashes += 100;
        if (timeInMilliseconds()-start > ms) break;
    }
    return rehashes;
}

总结一下，渐进式rehash分为两种情况：

  • 当进行增删改查操作时，每次以步长为1进行rehash

  • 当处于空闲状态时，每隔1ms会进行步长为10的rehash

这样保证了渐进式rehash操作可以尽快完成。

至此，字典的部分就结束了，而有关迭代器的相关介绍，留到后面再介绍。

4 集合

redis中，集合的底层数据结构一个是哈希表（见第3节），另一个是整数集合intset。当一个 Set 对象只包含整数值元素，并且集合中存储的元素个数小于512时，就会使用intset这个数据结构作为底层实现。

4.1 intset

源码位于./src/intset.h和./src/intset.c数据结构的定义如下：

1
2
3
4
5
typedef struct intset {
    uint32_t encoding;
    uint32_t length;
    int8_t contents[];
} intset;

其中：

  • encoding为编码方式，有三种类型，分别可以保存int16_t，int32_t，int64_t的整数
  • length为集合中包含的元素数量
  • contents为伸缩型数组成员，用于保存整数。虽然 intset 结构将 contents 属性声明为 int8_t 类型的数组， 但实际上 contents 数组并不保存任何 int8_t 类型的值 —— contents 数组的真正类型取决于 encoding 属性的值：
    • 如果 encoding 属性的值为 INTSET_ENC_INT16 ， 那么 contents 就是一个 int16_t 类型的数组， 数组里的每个项都是一个 int16_t 类型的整数值 （最小值为 -32,768 ，最大值为 32,767 ）。
    • 如果 encoding 属性的值为 INTSET_ENC_INT32 ， 那么 contents 就是一个 int32_t 类型的数组， 数组里的每个项都是一个 int32_t 类型的整数值 （最小值为 -2,147,483,648 ，最大值为 2,147,483,647 ）。
    • 如果 encoding 属性的值为 INTSET_ENC_INT64 ， 那么 contents 就是一个 int64_t 类型的数组， 数组里的每个项都是一个 int64_t 类型的整数值 （最小值为 -9,223,372,036,854,775,808 ，最大值为 9,223,372,036,854,775,807 ）。

4.2 intsetNew

intsetNew用于新建一个空的整数集合：

1
2
3
4
5
6
7
8
/* Create an empty intset. */
intset *intsetNew(void) {
    intset *is = zmalloc(sizeof(intset));
    // 默认采用INTSET_ENC_INT16，也就是int16_t的编码类型
    is->encoding = intrev32ifbe(INTSET_ENC_INT16);
    is->length = 0;
    return is;
}

4.3 插入元素

前面提到，contents会根据encoding的类型来保存整数，当插入一个整数，这个整数的大小超过了最初的int16_t，需要对其进行“升级”操作，升级整数集合并添加新元素共分为三步进行：

  1. 根据新元素的类型，扩展整数集合底层数组的空间大小， 并为新元素分配空间
  2. 将底层数组现有的所有元素都转换成与新元素相同的类型，并将类型转换后的元素放置到正确的位上，而且在放置元素的过程中，需要继续维持底层数组的有序性质不变
  3. 将新元素添加到底层数组里面

通过源码来理解这个过程：

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
/* Insert an integer in the intset */
intset *intsetAdd(intset *is, int64_t value, uint8_t *success) {
    // 检查要插入的value，看看使用哪种encoding类型可以存的下
    uint8_t valenc = _intsetValueEncoding(value);
    uint32_t pos;
    // 如果传入success，先设为1
    if (success) *success = 1;

    /* Upgrade encoding if necessary. If we need to upgrade, we know that
     * this value should be either appended (if > 0) or prepended (if < 0),
     * because it lies outside the range of existing values. */
    // 如果需要的话，进行升级操作。如果进行升级，需要知道这个value要被插入在前面还是后面
    if (valenc > intrev32ifbe(is->encoding)) {
        /* This always succeeds, so we don't need to curry *success. */
        // 使用intsetUpgradeAndAdd进行升级并插入
        return intsetUpgradeAndAdd(is,value);
    } else {
        /* Abort if the value is already present in the set.
         * This call will populate "pos" with the right position to insert
         * the value when it cannot be found. */
        // 否则说明不需要升级，检查value是否已经存在
        if (intsetSearch(is,value,&pos)) {
            // 如果已经存在，则直接返回
            if (success) *success = 0;
            return is;
        }
        // 到这里说明value不在集合中，分配内存空间
        is = intsetResize(is,intrev32ifbe(is->length)+1);
        // 如果要插入的位置不是末尾，则需要将pos后面的所有元素整体后移，为value腾出位置
        if (pos < intrev32ifbe(is->length)) intsetMoveTail(is,pos,pos+1);
    }
    // 设置值
    _intsetSet(is,pos,value);
    // 修改计数
    is->length = intrev32ifbe(intrev32ifbe(is->length)+1);
    return is;
}

我们先来看最简单的情况，也就是不需要升级的情况。首先会使用intsetSearch检查待插入值是否已经存在：

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
// 根据value在is中寻找，如果is中有指定值，则返回1.并把位置赋值给pos
// 如果没有找到，返回0，并把pos设置为value可以被插入的位置
static uint8_t intsetSearch(intset *is, int64_t value, uint32_t *pos) {
    int min = 0, max = intrev32ifbe(is->length)-1, mid = -1;
    int64_t cur = -1;

    /* The value can never be found when the set is empty */
    if (intrev32ifbe(is->length) == 0) {
        // 如果is为空，返回0
        if (pos) *pos = 0;
        return 0;
    } else {
        /* Check for the case where we know we cannot find the value,
         * but do know the insert position. */
        // is不为空，检查其他失败的情况
        if (value > _intsetGet(is,max)) {
            // _intsetGet用于获得is中指定位置的值
            // 由于is从小到大有序，所以如果value比is中最大的元素还大，则肯定找不到
            // 此时直接把pos设置为is的末尾即可
            if (pos) *pos = intrev32ifbe(is->length);
            return 0;
        } else if (value < _intsetGet(is,0)) {
            // 同理，如果value比is中的最小值还要小，则直接把pos设置为开头
            if (pos) *pos = 0;
            return 0;
        }
    }
    // 否则，进行二分查找 O(logn)
    while(max >= min) {
        // 使用移位操作来代替除以2，佩服作者的优化功力
        mid = ((unsigned int)min + (unsigned int)max) >> 1;
        cur = _intsetGet(is,mid);
        if (value > cur) {
            min = mid+1;
        } else if (value < cur) {
            max = mid-1;
        } else {
            break;
        }
    }

    if (value == cur) {
        // 如果集合中已有指定值，则返回1，并把pos位置设置为当前mid
        if (pos) *pos = mid;
        return 1;
    } else {
        // 没有指定值，则返回0，并把pos设置为当前min
        if (pos) *pos = min;
        return 0;
    }
}

如果value不在集合中，重新为集合分配内存，并整体后移元素，最后使用_intsetSet设置值：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// 将value插入到is的pos位置
static void _intsetSet(intset *is, int pos, int64_t value) {
    uint32_t encoding = intrev32ifbe(is->encoding);

    if (encoding == INTSET_ENC_INT64) {
        ((int64_t*)is->contents)[pos] = value;
        memrev64ifbe(((int64_t*)is->contents)+pos);
    } else if (encoding == INTSET_ENC_INT32) {
        ((int32_t*)is->contents)[pos] = value;
        memrev32ifbe(((int32_t*)is->contents)+pos);
    } else {
        ((int16_t*)is->contents)[pos] = value;
        memrev16ifbe(((int16_t*)is->contents)+pos);
    }
}

接下来看另一种情况。如果需要升级，则会调用intsetUpgradeAndAdd：

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
static intset *intsetUpgradeAndAdd(intset *is, int64_t value) {
    // 当前的encoding
    uint8_t curenc = intrev32ifbe(is->encoding);
    // 新的encoding
    uint8_t newenc = _intsetValueEncoding(value);
    int length = intrev32ifbe(is->length);
    int prepend = value < 0 ? 1 : 0;

    /* First set new encoding and resize */
    // 设置新的encoding
    is->encoding = intrev32ifbe(newenc);
    // 调整is的大小
    is = intsetResize(is,intrev32ifbe(is->length)+1);

    /* Upgrade back-to-front so we don't overwrite values.
     * Note that the "prepend" variable is used to make sure we have an empty
     * space at either the beginning or the end of the intset. */
    // 从后向前进行元素的重新赋值，所以并不会覆盖掉原先的值
    // prepend用于保证is的前/后有空余的空间存储新value
    while(length--)
        _intsetSet(is,length+prepend,_intsetGetEncoded(is,length,curenc));

    /* Set the value at the beginning or the end. */
    // 将新value插入到最前面或最后面
    if (prepend)
        _intsetSet(is,0,value);
    else
        _intsetSet(is,intrev32ifbe(is->length),value);
    // 更新计数
    is->length = intrev32ifbe(intrev32ifbe(is->length)+1);
    return is;
}

为了直观地理解这个过程，举个例子：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
假设原来有 curenc 编码的三个元素，它们在数组中排列如下：
| x | y | z | 
假设分配内存后，整个内存空间如下（? 表示未使用的内存）：
| x | y | z | ? |   ?   |   ?   |
这时程序从数组后端开始，重新插入元素：
| x | y | z | ? |   z   |   ?   |
| x | y |   y   |   z   |   ?   |
|   x   |   y   |   z   |   ?   |
最后，程序可以将新元素添加到最后 ? 号标示的位置中：
|   x   |   y   |   z   |  new  |
上面演示的是新元素比原来的所有元素都大的情况，也即是 prepend == 0
当新元素比原来的所有元素都小时（prepend == 1），调整的过程如下：
| x | y | z | ? |   ?   |   ?   |
| x | y | z | ? |   ?   |   z   |
| x | y | z | ? |   y   |   z   |
| x | y |   x   |   y   |   z   |
当添加新值时，原本的 | x | y | 的数据将被新值代替
|  new  |   x   |   y   |   z   |

4.4 intsetMoveTail

现在回过头看一下整体移动的函数intsetMoveTail，在插入元素时，如果要插入的位置不是末尾，需要将pos后面的所有元素整体后移，为value腾出位置：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
static void intsetMoveTail(intset *is, uint32_t from, uint32_t to) {
    void *src, *dst;
    // 要移动的字节数
    uint32_t bytes = intrev32ifbe(is->length)-from;
    // 编码类型
    uint32_t encoding = intrev32ifbe(is->encoding);
    // 设置对应类型的src和dst
    if (encoding == INTSET_ENC_INT64) {
        src = (int64_t*)is->contents+from;
        dst = (int64_t*)is->contents+to;
        bytes *= sizeof(int64_t);
    } else if (encoding == INTSET_ENC_INT32) {
        src = (int32_t*)is->contents+from;
        dst = (int32_t*)is->contents+to;
        bytes *= sizeof(int32_t);
    } else {
        src = (int16_t*)is->contents+from;
        dst = (int16_t*)is->contents+to;
        bytes *= sizeof(int16_t);
    }
    // 从src复制bytes个字节到dst
    memmove(dst,src,bytes);
}

以插入元素时整体后移的操作intsetMoveTail(is,pos,pos+1)为例：

1
2
3
4
5
6
7
8
9
说明：[字母]表示已存储数据的内存块，[ ]代表已分配还没有数据的内存块
移动前
[a]  [b]  [c]  [d]  [e]  [ ]
                |    |    
               pos  pos+1
移动后
[a]  [b]  [c]  [d]  [d]  [e]
                |    |     
               pos  pos+1

移动后只需把pos位置覆盖为新值即完成了插入。

实际上，通过移动元素，还可以进行删除操作，假设参数为intsetMoveTail(is,pos+1,pos)：

1
2
3
4
5
6
7
8
9
说明：[字母]表示已存储数据的内存块，[ ]代表已分配还没有数据的内存块
移动前
[a]  [b]  [c]  [d]  [e]  [f]
                |    |    
               pos  pos+1
移动后
[a]  [b]  [c]  [e]  [f]  [f]
                |    |     
               pos  pos+1

游动后只需删除最后的元素，即可完成pos位置的删除。

理解了这个过程，下面来看看删除元素的逻辑。

4.5 删除元素

intsetRemove函数用于删除元素：

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
intset *intsetRemove(intset *is, int64_t value, int *success) {
    // 获取被删除元素对应的encoding
    uint8_t valenc = _intsetValueEncoding(value);
    uint32_t pos;
    // 首先假设删除不成功
    if (success) *success = 0;

    // 判断：要删除的元素类型必须小于等于is中的类型，且is中存在此值
    if (valenc <= intrev32ifbe(is->encoding) && intsetSearch(is,value,&pos)) {
        // 获取当前is长度
        uint32_t len = intrev32ifbe(is->length);

        /* We know we can delete */
        // 肯定能够删除，所以success设置为1
        if (success) *success = 1;

        /* Overwrite value with tail and update length */
        // 删除pos位置的值
        if (pos < (len-1)) intsetMoveTail(is,pos+1,pos);
        // 重新分配内存
        is = intsetResize(is,len-1);
        // 更新计数
        is->length = intrev32ifbe(len-1);
    }
    return is;
}

整数集合这种数据结构设置多种类型，既增加了灵活性，又可以按需分配内存，提高了内存空间的有效利用率，

缺点是它不支持降级，因此假设一开始存储非常大的数，然后都是很小的数，这样就失去了整数集合的优势。另外的缺点是插入和删除都需要进行内存拷贝，这也是redis为什么要把整数集合的使用限制在512元素个数以内的原因。

5 有序集合

redis采用跳表作为有序集合底层数据结构之一。

5.1 跳表

提到有序集合，就不得不提到跳表。它的数据结构定义在./src/server.h中，节点定义如下：

1
2
3
4
5
6
7
8
9
typedef struct zskiplistNode {
    sds ele;
    double score;
    struct zskiplistNode *backward;
    struct zskiplistLevel {
        struct zskiplistNode *forward;
        unsigned long span;
    } level[];
} zskiplistNode;

其中：

  • ele为节点中保存的元素，为sds类型
  • score表示元素对应的得分（权重）
  • backward是上一个节点的指针，基于该指针可以对跳表进行倒序查找
  • level为一个结构体数组，同时也是伸缩型数组成员，用来表示节点的层级，每一层：
    • forward为每一层的前向指针
    • span表示跨度，表示该层当前节点与forward指针指向节点之间的距离

在此之上，跳表的定义：

1
2
3
4
5
typedef struct zskiplist {
    struct zskiplistNode *header, *tail;
    unsigned long length;
    int level;
} zskiplist;

其中header和tail分别指向头节点和尾节点，length为整个列表的节点个数，level表示当前跳表中的最高层数。

zset的定义：

1
2
3
4
typedef struct zset {
    dict *dict;
    zskiplist *zsl;
} zset;

它同时使用了跳表+哈希表的底层实现。

5.2 创建跳表

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
zskiplist *zslCreate(void) {
    int j;
    zskiplist *zsl;
    // 分配内存
    zsl = zmalloc(sizeof(*zsl));
    // 初始层级为1
    zsl->level = 1;
    // 节点数为0
    zsl->length = 0;
    // 头节点
    zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);
    // 为header节点初始化ZSKIPLIST_MAXLEVEL层（32层）每一层的指针和跨度
    for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) {
        zsl->header->level[j].forward = NULL;
        zsl->header->level[j].span = 0;
    }
    // header后向指针和tail都设为NULL
    zsl->header->backward = NULL;
    zsl->tail = NULL;
    return zsl;
}

其中，zslCreateNode用于创建跳表节点：

1
2
3
4
5
6
7
8
9
zskiplistNode *zslCreateNode(int level, double score, sds ele) {
    // 分配内存
    zskiplistNode *zn =
        zmalloc(sizeof(*zn)+level*sizeof(struct zskiplistLevel));
    // 初始化分数和element
    zn->score = score;
    zn->ele = ele;
    return zn;
}

初始节点示意图如下：

img

5.3 跳表的插入

向跳表中插入一个指定分值（权重）的元素：

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
zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {
    // 创建一个指向node节点的指针数组update，以及x
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    unsigned int rank[ZSKIPLIST_MAXLEVEL];
    int i, level;

    serverAssert(!isnan(score));
    // x初始指向跳表的头节点
    x = zsl->header;
    // 从最高层开始，逐步向下
    for (i = zsl->level-1; i >= 0; i--) {
        /* store rank that is crossed to reach the insert position */
        // 每次循环，rank[zsl->level-1]的位置总跨度初始为0，其余位置初始为上一个位置rank[i+1]的跨度
        rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
        // 判断：当前层级前向指针不为NULL，并且：
        // 当前层级前向节点的分数小于当前分数，或者当前层级前向节点的分数等于当前分数时，前向节点的元素比当前元素小（通过sdscmp比较）
        while (x->level[i].forward &&
                (x->level[i].forward->score < score ||
                    (x->level[i].forward->score == score &&
                    sdscmp(x->level[i].forward->ele,ele) < 0)))
        {
            // 总结：前向节点存在，且前向节点比当前节点小，会进入循环
            // 此时会用span值累加rank[i]
            rank[i] += x->level[i].span;
            // 向后找
            x = x->level[i].forward;
        }
        // 并更新update[i]，updated[i]中保存的是新节点每层level的前驱
        update[i] = x;
    }
    /* we assume the element is not already inside, since we allow duplicated
     * scores, reinserting the same element should never happen since the
     * caller of zslInsert() should test in the hash table if the element is
     * already inside or not. */
    // 我们认为插入的元素不存在。
    // 由于允许重复的scores，调用该函数之前就应该确认这个元素到底在不在里面，所以重新插入相同的元素永远不会发生

    // 新插入节点的level随机
    level = zslRandomLevel();
    // 更新level
    if (level > zsl->level) {
        // 如果这个随机层数level > 跳表的最高高度，就将跳表的层数增加到level
        for (i = zsl->level; i < level; i++) {
            // 将多出来的层级对应的rank设置为0
            rank[i] = 0;
            // update节点设置为header
            update[i] = zsl->header;
            update[i]->level[i].span = zsl->length;
        }
        // 更新跳表的层高
        zsl->level = level;
    }
    // 新建节点，
    x = zslCreateNode(level,score,ele);
    for (i = 0; i < level; i++) {
        // 新节点的每层level进行头插
        x->level[i].forward = update[i]->level[i].forward;
        update[i]->level[i].forward = x;

        /* update span covered by update[i] as x is inserted here */
        // 新节点的span等于其前驱的span-(rank[0]-rank[i])
        x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
        // 新节点前驱的span等于(rank[0] - rank[i]) + 1
        update[i]->level[i].span = (rank[0] - rank[i]) + 1;
    }

    /* increment span for untouched levels */
    // 如果level本身小于跳表的level，还要更新高于level的部分
    for (i = level; i < zsl->level; i++) {
        update[i]->level[i].span++;
    }
    // 设置新节点的前向指针
    x->backward = (update[0] == zsl->header) ? NULL : update[0];
    if (x->level[0].forward)
        // 如果x有前驱，就将前驱的前向指针指向自己
        x->level[0].forward->backward = x;
    else
        // 否则说明x为最后一个节点，将跳表的tail指向x
        zsl->tail = x;
    // 节点长度+1
    zsl->length++;
    return x;
}

添加操作可以分成两个部分，第一部分是找到待插入的位置，也就是查找，第二部分是创建节点并插入。其中查找是比较关键的步骤，redis并没有为其封装单独的函数，可能是不同的操作（如删除、修改等）查找的步骤在细节上无法统一。不过查找的大致思路是相似的。另外，在每次创建新节点的时候，这个节点的层数是随机的，具体的随机逻辑如下：

1
2
3
4
5
6
int zslRandomLevel(void) {
    int level = 1;
    while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
        level += 1;
    return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}

其中ZSKIPLIST_P=0.25，ZSKIPLIST_MAXLEVEL=32。首先生成一个0-1的随机数，如果随机数小于0.25，则继续增加层数直到随机数大于0.25结束，最后层数如果超过了32，则按32返回。采用该算法可以得出结论：层数越高，概率越小；层数越低，概率越大。

5.4 删除节点

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
int zslDelete(zskiplist *zsl, double score, sds ele, zskiplistNode **node) {
    // 删除节点只需要update，也就是所有的前驱level
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    int i;
    // x初始指向跳表的头节点
    x = zsl->header;
    // 查找的过程和插入节点类似
    for (i = zsl->level-1; i >= 0; i--) {
        while (x->level[i].forward &&
                (x->level[i].forward->score < score ||
                    (x->level[i].forward->score == score &&
                     sdscmp(x->level[i].forward->ele,ele) < 0)))
        {
            x = x->level[i].forward;
        }
        update[i] = x;
    }
    /* We may have multiple elements with the same score, what we need
     * is to find the element with both the right score and object. */
    // 由于可能有多个元素拥有相同的得分，所以还要保证找到分数和对象都正确的节点
    // x = x->level[0].forward是可能要删除的节点
    x = x->level[0].forward;
    // 保证删除正确的节点
    if (x && score == x->score && sdscmp(x->ele,ele) == 0) {
        // 删除该节点
        zslDeleteNode(zsl, x, update);
        if (!node)
            // 如果没有传入node参数，直接释放节点
            zslFreeNode(x);
        else
            // 否则，将其赋值给node，以供调用者使用
            *node = x;
        // 找到节点返回1
        return 1;
    }
    // 没有找到节点 返回0
    return 0; /* not found */
}

删除节点的内部函数zslDeleteNode：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
// 根据update以及待删除节点x，在zsl中删除它
void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {
    int i;
    // 更新x所有level前驱的后继指针
    for (i = 0; i < zsl->level; i++) {
        if (update[i]->level[i].forward == x) {
            update[i]->level[i].span += x->level[i].span - 1;
            update[i]->level[i].forward = x->level[i].forward;
        } else {
            update[i]->level[i].span -= 1;
        }
    }
    if (x->level[0].forward) {
        // 如果待删除节点不是最后一个节点，更新其后面节点的前向指针
        x->level[0].forward->backward = x->backward;
    } else {
        // 否则说明是最后一个节点，更新zsl的tail指针
        zsl->tail = x->backward;
    }
    // 删除后更新level
    while(zsl->level > 1 && zsl->header->level[zsl->level-1].forward == NULL)
        zsl->level--;
    zsl->length--;
}

5.5 修改节点

至于修改节点，也分为查找和修改两部分，其中查找原理与上文基本相同，这里不再赘述，而修改分为两种情况，一种是修改后的位置仍然不变，则无需后续操作；另一种是修改后位置改变，则删除该位置的节点，重新插入，源码如下：

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
zskiplistNode *zslUpdateScore(zskiplist *zsl, double curscore, sds ele, double newscore) {
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    int i;

    /* We need to seek to element to update to start: this is useful anyway,
     * we'll have to update or remove it. */
    x = zsl->header;
    for (i = zsl->level-1; i >= 0; i--) {
        while (x->level[i].forward &&
                (x->level[i].forward->score < curscore ||
                    (x->level[i].forward->score == curscore &&
                     sdscmp(x->level[i].forward->ele,ele) < 0)))
        {
            x = x->level[i].forward;
        }
        update[i] = x;
    }

    /* Jump to our element: note that this function assumes that the
     * element with the matching score exists. */
    x = x->level[0].forward;
    serverAssert(x && curscore == x->score && sdscmp(x->ele,ele) == 0);

    /* If the node, after the score update, would be still exactly
     * at the same position, we can just update the score without
     * actually removing and re-inserting the element in the skiplist. */
    if ((x->backward == NULL || x->backward->score < newscore) &&
        (x->level[0].forward == NULL || x->level[0].forward->score > newscore))
    {
        x->score = newscore;
        return x;
    }

    /* No way to reuse the old node: we need to remove and insert a new
     * one at a different place. */
    zslDeleteNode(zsl, x, update);
    zskiplistNode *newnode = zslInsert(zsl,newscore,x->ele);
    /* We reused the old node x->ele SDS string, free the node now
     * since zslInsert created a new one. */
    x->ele = NULL;
    zslFreeNode(x);
    return newnode;
}

四、高级数据结构

1 对象

1.1 对象定义

redis在众多的基础数据结构之上，定义了redis对象系统，这个系统包含字符串对象、列表对象、哈希对象、集合对象和有序集合对象这五种类型的对象，每种对象都用到了至少一种我们前面所介绍的数据结构。基于数据对象的进一步封装，可以针对不同的使用场景， 为对象设置多种不同的数据结构实现，从而优化对象在不同场景下的使用效率。redis对象的定义在./src/server.h中：

1
2
3
4
5
6
7
8
9
typedef struct redisObject {
    unsigned type:4;
    unsigned encoding:4;
    unsigned lru:LRU_BITS; /* LRU time (relative to global lru_clock) or
                            * LFU data (least significant 8 bits frequency
                            * and most significant 16 bits access time). */
    int refcount;
    void *ptr;
} robj;

其中：

  • type占4位，表示redis对象的类型，也就是常说的五种数据结构，此外两个特殊的类型：

    1
    2
    3
    4
    5
    6
    7
    #define OBJ_STRING 0    /* 字符串对象 */
    #define OBJ_LIST 1      /* 列表对象 */
    #define OBJ_SET 2       /* 集合对象 */
    #define OBJ_ZSET 3      /* 有序集合对象 */
    #define OBJ_HASH 4      /* 字典（哈希）对象 */
    #define OBJ_MODULE 5    /* Module对象 */
    #define OBJ_STREAM 6    /* Stream对象 */

  • encoding占4位，表示该对象类型具体使用的底层数据结构

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    #define OBJ_ENCODING_RAW 0     /* 原始数据编码 */
    #define OBJ_ENCODING_INT 1     /* 整数编码 */
    #define OBJ_ENCODING_HT 2      /* 哈希表编码 */
    #define OBJ_ENCODING_ZIPMAP 3  /* zipmap编码（不再使用） */
    #define OBJ_ENCODING_LINKEDLIST 4 /* 双向链表编码（不再使用） */
    #define OBJ_ENCODING_ZIPLIST 5 /* 压缩链表编码 */
    #define OBJ_ENCODING_INTSET 6  /* 整数集合编码 */
    #define OBJ_ENCODING_SKIPLIST 7  /* 跳表编码 */
    #define OBJ_ENCODING_EMBSTR 8  /* 嵌入数据对字符串编码 */
    #define OBJ_ENCODING_QUICKLIST 9 /* quicklist编码 */
    #define OBJ_ENCODING_STREAM 10 /* listpack编码 */

  • lru默认占24位，表示进行过期键处理时的策略

  • refcount占4字节，引用计数，用于内存回收

  • ptr占8字节，是可以指向任何类型的指针

有关引用计数、过期处理等内容会单独介绍，这里首先关注高级数据结构的定义。

1.2 对象创建

createObject是通用的redis对象创建函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
robj *createObject(int type, void *ptr) {
    // 分配内存
    robj *o = zmalloc(sizeof(*o));
    // 设置类型
    o->type = type;
    // 默认编码方式为原始数据编码
    o->encoding = OBJ_ENCODING_RAW;
    // 设置指针
    o->ptr = ptr;
    // 引用计数初始为1
    o->refcount = 1;

    /* Set the LRU to the current lruclock (minutes resolution), or
     * alternatively the LFU counter. */
    // lru相关
    if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
        o->lru = (LFUGetTimeInMinutes()<<8) | LFU_INIT_VAL;
    } else {
        o->lru = LRU_CLOCK();
    }
    return o;
}

除此之外，redis也为不同的数据类型提供创建接口和释放接口，这些接口会在后续介绍。

2 字符串对象

2.1 编码

字符串对象拥有三种编码类型，分别是：

  • OBJ_ENCODING_RAW
  • OBJ_ENCODING_INT
  • OBJ_ENCODING_EMBSTR

在./src/object.c中tryObjectEncoding函数用于为字符串类型编码：

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
robj *tryObjectEncoding(robj *o) {
    // 传入一个redis字符串对象，为其选择合适的编码以节省空间
    long value;
    sds s = o->ptr;
    size_t len;

    /* Make sure this is a string object, the only type we encode
     * in this function. Other types use encoded memory efficient
     * representations but are handled by the commands implementing
     * the type. */
    // 确保这是一个字符串对象
    serverAssertWithInfo(NULL,o,o->type == OBJ_STRING);

    /* We try some specialized encoding only for objects that are
     * RAW or EMBSTR encoded, in other words objects that are still
     * in represented by an actually array of chars. */
    // 如果对象o的编码不是OBJ_ENCODING_RAW或者OBJ_ENCODING_EMBSTR，直接返回
    if (!sdsEncodedObject(o)) return o;

    /* It's not safe to encode shared objects: shared objects can be shared
     * everywhere in the "object space" of Redis and may end in places where
     * they are not handled. We handle them only as values in the keyspace. */
     // 如果引用计数大于1，返回；对于引用计数大于1的对象进行编码是不安全的
     if (o->refcount > 1) return o;

    /* Check if we can represent this string as a long integer.
     * Note that we are sure that a string larger than 20 chars is not
     * representable as a 32 nor 64 bit integer. */
    // 检查是否能将这个字符串表示为一个长整型
    // 获取字符串长度
    len = sdslen(s);
    // 如果长度小于20，并且可以from string to long（由字符串转长整型）
    if (len <= 20 && string2l(s,len,&value)) {
        /* This object is encodable as a long. Try to use a shared object.
         * Note that we avoid using shared integers when maxmemory is used
         * because every object needs to have a private LRU field for the LRU
         * algorithm to work well. */
        // 尝试用共享类型，共享类型指的是启动时就创建好的整数
        // 但是，当开启了maxmemory时不使用这种方式，因为对象的lru字段是私有的
        if ((server.maxmemory == 0 ||
            !(server.maxmemory_policy & MAXMEMORY_FLAG_NO_SHARED_INTEGERS)) &&
            value >= 0 &&
            value < OBJ_SHARED_INTEGERS)
        {
            // OBJ_SHARED_INTEGERS，默认为10000，即小于10000的数
            // 引用计数减一
            decrRefCount(o);
            // 将共享整数的引用加一并返回
            incrRefCount(shared.integers[value]);
            return shared.integers[value];
        } else {
            // 否则判断encoding是否为OBJ_ENCODING_RAW
            if (o->encoding == OBJ_ENCODING_RAW) {
                sdsfree(o->ptr);
                // 将encoding改为OBJ_ENCODING_INT
                o->encoding = OBJ_ENCODING_INT;
                // 直接使用指针存储
                o->ptr = (void*) value;
                return o;
            } else if (o->encoding == OBJ_ENCODING_EMBSTR) {
                // 判断encoding为OBJ_ENCODING_EMBSTR
                decrRefCount(o);
                // 调用createStringObjectFromLongLongForValue
                return createStringObjectFromLongLongForValue(value);
            }
        }
    }

    /* If the string is small and is still RAW encoded,
     * try the EMBSTR encoding which is more efficient.
     * In this representation the object and the SDS string are allocated
     * in the same chunk of memory to save space and cache misses. */
    // 如果进行到这里仍然是RAW编码，则尝试将其编码为EMBSTR编码
    // 在这种编码格式下，对象和SDS字符串被分配在同一块内存中（连续），以节省空间、增加缓存命中
    if (len <= OBJ_ENCODING_EMBSTR_SIZE_LIMIT) {
        robj *emb;

        if (o->encoding == OBJ_ENCODING_EMBSTR) return o;
        emb = createEmbeddedStringObject(s,sdslen(s));
        decrRefCount(o);
        return emb;
    }

    /* We can't encode the object...
     *
     * Do the last try, and at least optimize the SDS string inside
     * the string object to require little space, in case there
     * is more than 10% of free space at the end of the SDS string.
     *
     * We do that only for relatively large strings as this branch
     * is only entered if the length of the string is greater than
     * OBJ_ENCODING_EMBSTR_SIZE_LIMIT. */
    // 进行到这里说明该对象无法编码，进行最后的尝试
    // 对sds对象进行空间优化，保证其末尾的空闲空间不超过10%
    trimStringObjectIfNeeded(o);

    /* Return the original object. */
    return o;
}

总结一下编码逻辑为：

  1. 首先进行前置检查，确保字符串对象符合要求
  2. 判断字符串能否转为长整型存储（字符串为整型，且长度小于20）
    • 如果可以转化，首先判断这个整型是否小于10000，如果小于10000则使用共享引用的方式，不重复创建
    • 如果大于10000，且采用OBJ_ENCODING_RAW编码，则将其保存到指针中存储
    • 如果大于10000，采用OBJ_ENCODING_EMBSTR编码，则创建整型存储的字符串
  3. 如果不能转为长整型，判断字符串长度是否大于OBJ_ENCODING_EMBSTR_SIZE_LIMIT即44的限制，如果小于，则创建OBJ_ENCODING_EMBSTR类型的字符串
  4. 最后，以上都不满足，对sds对象进行空间优化，保证其末尾的空闲空间不超过10%

在创建整型数值时，最终调用的是createStringObjectFromLongLongWithOptions：

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
// 通过给定的value，以及参数valueobj，返回对应的字符串对象
robj *createStringObjectFromLongLongWithOptions(long long value, int valueobj) {
    robj *o;

    if (server.maxmemory == 0 ||
        !(server.maxmemory_policy & MAXMEMORY_FLAG_NO_SHARED_INTEGERS))
    {
        /* If the maxmemory policy permits, we can still return shared integers
         * even if valueobj is true. */
        // 即使valueobj为1，只要maxmemory策略允许，仍然可以使用共享对象
        valueobj = 0;
    }
    // 判断value以及valueobj的数值，如果满足
    // value >= 0 && value < 共享整数最大值10000 && valueobj == 0，
    if (value >= 0 && value < OBJ_SHARED_INTEGERS && valueobj == 0) {
        // 则直接从shared.integers中返回共享对象
        incrRefCount(shared.integers[value]);
        o = shared.integers[value];
    } else {
        // 否则，则判断value >= LONG_MIN && value <= LONG_MAX（LONG能表示的范围）
        if (value >= LONG_MIN && value <= LONG_MAX) {
            // 创建字符串对象
            o = createObject(OBJ_STRING, NULL);
            // 使用OBJ_ENCODING_INT编码方式
            o->encoding = OBJ_ENCODING_INT;
            // 将值value存储到robj.ptr指针上
            o->ptr = (void*)((long)value);
        } else {
            // 以上都不满足，则通过sdsfromlonglong创建一个sds
            // 使用该sds创建一个字符串对象
            o = createObject(OBJ_STRING,sdsfromlonglong(value));
        }
    }
    return o;
}

在这个函数的基础上，另外两个函数是其valueobj的两种情况：

  • createStringObjectFromLongLong，对应valueobj=0
  • createStringObjectFromLongLongForValue，对应valueobj=1

2.2 EMBSTR创建

接下来通过源码查看EMBSTR类型字符串的创建，该类型在内存上是连续、不可修改的：

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
robj *createEmbeddedStringObject(const char *ptr, size_t len) {
    // 分配的内存包括robj结构体+sdshdr8结构体+字符串长度+1（结尾的'\0'）
    robj *o = zmalloc(sizeof(robj)+sizeof(struct sdshdr8)+len+1);
    // o+1即紧挨着o后面就是sh结构体
    struct sdshdr8 *sh = (void*)(o+1);

    o->type = OBJ_STRING;
    o->encoding = OBJ_ENCODING_EMBSTR;
    // sh+1指向buf
    o->ptr = sh+1;
    o->refcount = 1;
    if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
        o->lru = (LFUGetTimeInMinutes()<<8) | LFU_INIT_VAL;
    } else {
        o->lru = LRU_CLOCK();
    }

    sh->len = len;
    sh->alloc = len;
    sh->flags = SDS_TYPE_8;
    if (ptr == SDS_NOINIT)
        sh->buf[len] = '\0';
    else if (ptr) {
        memcpy(sh->buf,ptr,len);
        sh->buf[len] = '\0';
    } else {
        memset(sh->buf,0,len+1);
    }
    return o;
}

2.3 长度限制

在./src/t_string.c中，定义了一些字符串操作的命令，其中对于最大长度的限制由checkStringLength定义：

1
2
3
4
5
6
7
static int checkStringLength(client *c, long long size) {
    if (!(c->flags & CLIENT_MASTER) && size > server.proto_max_bulk_len) {
        addReplyError(c,"string exceeds maximum allowed size (proto-max-bulk-len)");
        return C_ERR;
    }
    return C_OK;
}

proto_max_bulk_len默认512MB。