为了确保连续多个操作的原子性，一个成熟的数据库通常都会有事务支持，Redis也不例外。Redis的事务使用非常简单，不同于关系数据库，我们无需理解那么多复杂的事务模型了，就可以直接使用。不过也正是因为这种简单性，它的事务模型很不严格，这要求我们不能像使用关系数据库的事务一样使用Redis。

Redis事务的基本使用

每个事务的操作都有begin、commit、和rollback、begin指示事务的开始，commit指示事务的提交，rollback指示事务的回滚。

Redis与其差不多，对应的分别是multi/exec/discard。multi指示事务的开始，exec指示事务的执行，discard指示事务的丢弃。

1
2
3
4
5
6
7
8
9

>multi  
OK  
>incr books  
QUEUED  
>incr books  
QUEUED  
>exec  
(integer) 1
(integer）2

上面的指令演示了一个完整的事务过程，所有的指令在exec之前不执行，而是缓存在服务器的一个事务队列中，服务器一旦受到exec指令，就开始执行整个事务队列，执行完毕后一次性返回所有指令的运行结果。因为Redis的单线程特性，它不用担心自己在执行队列的时候被其他指令打搅，可以保证他们得到的[原子性]执行。

上图显示了以上事务过程完整的交互结果。QUEUED是一个简单字符串，同OK是一个形式，它表示指令已经被服务器缓存到队列里了。

大多数人一直对Redis管道有一个误解，他们以为这是Redis服务器提供的一种特别的技术，有了这种技术就可以加速Redis的存取效率。但是实际上Redis管道（Pipeline）本身并不是Redis服务器直接提供的技术，这个技术本质上是由客户端提供的，跟服务器没有什么直接的关系。

Redis的消息交互

当我们使用客户端对Redis进行一次操作时，如下图所示，客户端将请求传送给服务器，服务器处理完毕后，再将响应回复给客户端。这要花费一个网络数据包的来回时间。

Redis的数据全部在内存里，如果突然宕机，数据就会全部丢失，因此必须有一种机制来保证Redis的数据不会因为故障而丢失，这种机制就是Redis的持久化机制。

Redis的持久化机制有两种，第一种就是快照（RDB文件），第二种就是AOF日志。快照是一次全量备份，AOF日志是连续的增量备份。快照是内存数据的二进制序列化形式，在存储上非常紧凑，而AOF日志记录的是内存数据修改的指令记录文本。AOF日志在长期的运行过程中会变得非常大，数据库重启时需要加载AOF日志进行指令重放，这个时间会变得无比漫长。所以需要定期进行AOF重写，给AOF文件瘦身。

快照原理

我们知道Redis是单线程程序，这个线程要同时负责多个客户端套接字的并发读写操作和内存数据结构的逻辑读写。

在服务线上请求的同时，Redis还需要进行内存快照，内存快照要求Redis必须进行文件IO操作，可文件IO操作是不能使用多路复用API。

这意味着单线程同时在服务线上的请求和吉林文件IO操作，文件IO操作会严重拖垮服务器请求的性能。还有个重要的问题就是为了不阻塞线上的业务，就需要边持久化边响应客户端请求。持久化的同时，内存数据结构还在变化，比如一个大型的hash字典正在持久化，结果一个请求过来把它给删除了，还没持久化，这该怎么办？

Redis的作者认为数据库系统的瓶颈一般不在于网络流量，而是数据库自身内部逻辑处理上。所以即使Redis使用了浪费流量的文本协议，依然可以取得极高的访问性能。Redis将所有数据都放在内存，用一个单线程对外提供服务，单个节点在跑满一个CPU核心的情况下可以达到10w/s的超高QPS。

RESP

RESP（Redis Serialization Protocol）是Redis序列化协议的简写。它是一种直观的文本协议，优势在于实现异常简单，解析性能极好。

Redis歇息将传输的数据结构分为5种最小单元类型，单元结束时统一加上回车换行符号\r\n。

• 单行字符以+符号开头
• 多行字符以$符号开头，后跟字符串长度
• 整数值以：符号开头，后跟整数的字符串形式
• 错误消息以-符号开头
• 数组以*符号开头，后跟数组的长度

单行字符

1

+hello world\r\n

多行字符串

1
2

$11\r\n  
hello world\r\n

整数

1

:1024\r\n

错误

1

-WRONGTYPE Operation ....

数组[1,2,3]

1
2
3
4

*3\r\n  
:1\r\n
:2\r\n  
:3\r\n

Redis是一个单线程程序

非阻塞IO

当我们调用套接字的读写方法，默认他们是阻塞的，比如read方法要传递一个参数n，表示最多读取这门多字节后返回，如果一个字节都没有，那么线程就会卡在那里，直到新的数据到来或者连接关闭，read方法才可以返回，线程才能够继续处理。而write方法一般来说不会阻塞，除非内核为套接字分配的写缓冲区已经写满了，write方法才会阻塞，直到缓冲区有空闲空间。
非阻塞IO在套接字对象上提供了一个选项Non_Blocking,当这个选项打开时，读写方法不会阻塞。能读多少取决于内核为套接字分配的读缓冲区的空闲空间字节数，能写多少取决于内核为套接字分配的写缓冲区的空闲空间字节数。读方法和写方法都会通过返回值来告知程序实际读写了多少字节。

事件轮询（多路复用）

非阻塞IO有个问题，就是线程要读数据，但是读了一部分就返回了，线程如何知道什么时候应该继续读，也就是当数据到来时，线程如何得到通知。写也是一样。
事件轮询API就是用来解决这个问题的，最简单的时间轮询API是select函数，它是操作系统提供给用户程序的API。输入时读写描述符列表read_fds & write_fds，输出是与之对应的可读可写时间。同时还提供了一个timeout参数，如果没有任何事件的到来，那么最多等待timeout时间，线程处于阻塞状态。一旦期间有事件到来，就可以立即返回。线程就可以继续挨个处理相应的事件。处理完了继续过来轮询。于是线程就进入了一个死循环。我们把这个死循环称为事件轮询，一个循环为一个周期。
事件轮询API就是Java语言中的NIO技术

指令队列

Redis会将每个客户端套接字都关联一个指令队列。客户端的指令通过队列来排队进行顺序处理，先到先服务。

响应队列

Redis同样会为每个客户端套接字关联一个响应队列。
Redis服务器通过响应队列来将指令的返回结果回复给客户端。如果队列为空，那么以为这连接暂时处于空闲状态，不需要去获取写事件，也就是可以将当前的客户端描述符从write_fds里面移除。等到队列有数据了，再将描述符放进去。避免select系统调用立即返回写事件，结果发现没什么数据可以写。出现这种情况的线程会飙高CPU

定时任务

服务器处理要响应的IO事件外，还要处理其他事情。比如定时任务。
如果线程阻塞在select系统调用上，定时任务将无法得到准时调度。那么Redis是如何解决这个问题的呢？

Redis的定时任务会记录在一个最小堆的数据结构中。这个堆中，最快要执行的任务排在堆的最上方。在每个循环周期，Redis都会将最小堆里面已经到点的任务立即进行处理。处理完毕后，将最快要执行的任务还需要的时间记录下来，这个时间就是select系统调用的timeout参数。因为Redis知道未来timeout时间内，没有其他定时任务需要处理，所以可以安心睡眠timeout时间。

python装饰器就是用于扩展原来函数功能的一种函数，这个函数的特殊之处在于它的返回值也是一个函数。使用python装饰器的好处就是在不用更改原函数的代码前提下给函数增加新的功能。

1
2
3

原函数
def func():
	print("hello")

要想扩展一个函数的功能，最简单的方法就是直接修改原函数。

1
2
3
4

def func():
	print("before")
	print("hello")
	print("after")

如果不想修改原函数，还是想增强函数的功能时，可以另外定义一个函数调用原函数。（类似于设计模式中的装饰模式，有组合和代理两种方式）

1
2
3
4
5
6
7
8
9
10
11
12
13
14

def deco(func):
	print("before")
	func()
	print("after")

def func():
	print("hello")

if __name__ == '__main__':
    f = func
    deco(f)#只有把func()或者f()作为参数执行，新加入功能才会生效
    print("f.__name__ is",f.__name__)#f的name就是func()
    print()
    #func()

但是如果存在很多个类似于func的函数需要相同的扩展，那岂不是要执行deco函数许多次？
下面我们实现一个最简陋的装饰器：

1
2
3
4
5
6
7
8
9
10

def deco(func):
	def wrapper(*args, **kwargs):
		print("before")
		func(*args, **kwargs)
		print("after")
	return wrapper

@deco
def func():
	print("hello")

这里的deco函数就是最原始的装饰器，它的参数是一个函数，然后返回值也是一个函数。其中作为参数的这个函数func()就在返回函数wrapper()的内部执行。然后在函数func()前面加上@deco。
所以这里装饰器就像一个注入符号：有了它，拓展了原来函数的功能既不需要侵入函数内更改代码，也不需要重复执行原函数。
在func函数前还可以使用多个@的方式来执行多个装饰器，多个装饰器的执行顺序就是从最后一个装饰器开始执行到第一个装饰器，在执行函数本身。

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

def dec1(func):  
    print("1111")  
    def one():  
        print("2222")  
        func()  
        print("3333")  
    return one  

def dec2(func):  
    print("aaaa")  
    def two():  
        print("bbbb")  
        func()  
        print("cccc")  
    return two  

@dec1  
@dec2  
def test():  
    print("test test")  

test()  

aaaa  
1111  
2222  
bbbb  
test test  
cccc  
3333

压缩列表：压缩列表是列表键和哈希键的底层实现之一。为了提高内存存储效率而升级的。
当一个列表键只包含少量列表项，并且每个列表项要么就是小整数值，要么就是长度比较短的字符串，那么Redis会使用压缩列表来做列表键的底层实现。

注：redis 3.2以后，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

/*
 * 保存 ziplist 节点信息的结构
 */
typedef struct zlentry {

    // prevrawlen ：前置节点的长度
    // prevrawlensize ：编码 prevrawlen 所需的字节大小
    unsigned int prevrawlensize, prevrawlen;

    // len ：当前节点值的长度
    // lensize ：编码 len 所需的字节大小
    unsigned int lensize, len;

    // 当前节点 header 的大小
    // 等于 prevrawlensize + lensize
    unsigned int headersize;

    // 当前节点值所使用的编码类型
    unsigned char encoding;

    // 指向当前节点的指针
    unsigned char *p;

} zlentry;

虽然定义了这个结构体，但是根本就没有使用zlentry结构来作为压缩列表中用来存储数据节点中的结构。
因为这个结构存小整数或短字符是在太浪费空间了。这个结构总共在32位机占用28个字节，64位机占用32个字节。这不符合压缩列表的设计目的。
压缩列表的节点真正结构如下图：

• prev_entry_len：记录前驱节点的长度
• encoding：记录当前节点的value成员的数据类型以及长度
• value：根据encoding保存字节数组或整数

prev_entry_len

prev_entry_len成员实际上就是zlentry结构中prevrawlensize,和prevrawlen这两个成员的压缩版。
这两个成员都是int类型，因此将两者压缩为一个成员prev_entry_len。分别对不同长度的前驱节点使用不同的字节数表示：
1) 如果前置节点的长度小于 254 字节，那么程序将使用 1 个字节来保存这个长度值。
2) 如果前置节点的长度大于等于 254 字节，那么程序将使用 5 个字节来保存这个长度值：
a) 第 1 个字节的值将被设为 254 ，用于标识这是一个 5 字节长的长度值。
b) 之后的 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

/* Encode the length of the previous entry and write it to "p". Return the
 * number of bytes needed to encode this length if "p" is NULL. 
 *
 * 对前置节点的长度 len 进行编码，并将它写入到 p 中，
 * 然后返回编码 len 所需的字节数量。
 *
 * 如果 p 为 NULL ，那么不进行写入，仅返回编码 len 所需的字节数量。
 *
 * T = O(1)
 */
static unsigned int zipPrevEncodeLength(unsigned char *p, unsigned int len) {

    // 仅返回编码 len 所需的字节数量
    if (p == NULL) {
        return (len < ZIP_BIGLEN) ? 1 : sizeof(len)+1;

    // 写入并返回编码 len 所需的字节数量
    } else {

        // 1 字节
        if (len < ZIP_BIGLEN) {
            p[0] = len;
            return 1;

        // 5 字节
        } else {
            // 添加 5 字节长度标识
            p[0] = ZIP_BIGLEN;
            // 写入编码
            memcpy(p+1,&len,sizeof(len));
            // 如果有必要的话，进行大小端转换
            memrev32ifbe(p+1);
            // 返回编码长度
            return 1+sizeof(len);
        }
    }
}

encoding

和prev_entry_len一样，encoding成员同样可以看做成zlentry结构中lensize和len的压缩版。
同样的lensize和len都是占4个字节的，因此将两者压缩为一个成员encoding，只要encoding能够同时具有lensize和len成员的功能，而且对当前节点保存的是字节数组还是整数分别编码。

zlentry中len字段配合encoding字段进行了编码, 尽量压缩字段长度, 减少内存使用。
1）如果节点保存的是字符串，那么使用两个字节来保存编码字符串长度所使用的类型。之后的跟着的内容为字符串的实际长度。

1. |00pppppp| 字符串长度小于等于63字节（6bit)
2. |01pppppp|qqqqqqqq| 字符串长度小于等于16383字节(14bit)
3. |10______|qqqqqqqq|rrrrrrrr|ssssssss|tttttttt| 字符串大于等于16384字节；

2）如果节点保存的是征收值，则头两位设置为1，紧跟着的两位用于标识节点所保存的整数类型。

1. |11000000| 节点的值为 int16_t 类型的整数，长度为 2 字节。
2. |11010000| 节点的值为 int32_t 类型的整数，长度为 4 字节。
3. |11100000| 节点的值为 int64_t 类型的整数，长度为 8 字节。
4. |11110000| 节点的值为 24 位（3 字节）长的整数。
5. |11111110| 节点的值为 8 位（1 字节）长的整数。
6. |1111xxxx| 节点的值为介于 0 至 12 之间的无符号整数。取出来的值还得减一。
7. |11111111| ziplist 的结尾标识

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

/*字符串编码标识使用了最高2bit位 */
#define ZIP_STR_06B (0 << 6)  //6bit
#define ZIP_STR_14B (1 << 6)  //14bit
#define ZIP_STR_32B (2 << 6)  //32bit

/*zlentry中len字段进行编码过程*/
static unsigned int zipEncodeLength(unsigned char *p, unsigned char encoding, unsigned int rawlen) {
    unsigned char len = 1, buf[5];

    if (ZIP_IS_STR(encoding)) {
        /*
          *6bit可以存储, 占用空间为1个字节, 值存储在字节后6bit中.
          */
        if (rawlen <= 0x3f) {
            if (!p) return len;
            buf[0] = ZIP_STR_06B | rawlen;
        } else if (rawlen <= 0x3fff) {
            len += 1;
            if (!p) return len;
               /*14bit可以存储, 置前两个bit位为ZIP_STR_14B标志 */
            buf[0] = ZIP_STR_14B | ((rawlen >> 8) & 0x3f);
            buf[1] = rawlen & 0xff;
        } else {
            len += 4;
            if (!p) return len;
            buf[0] = ZIP_STR_32B;
            buf[1] = (rawlen >> 24) & 0xff;
            buf[2] = (rawlen >> 16) & 0xff;
            buf[3] = (rawlen >> 8) & 0xff;
            buf[4] = rawlen & 0xff;
        }
    } else {
        /* 内容编码为整型, 长度默认为1*/
        if (!p) return len;
        buf[0] = encoding;
    }

    /* Store this length at p */
    memcpy(p,buf,len);
    return len;
}
2.3 zlentry之encoding和p编码
zlentry中encoding和p表示元素编码和内容, 下面分析下具体编码规则, 可以看到这里对内存节省真是到了魔性的地步. encoding是保存在len字段第一个字节中, 第一个字节最高2bit标识字符串编码, 5和6bit位标识是整数编码, 解码时直接从第一个字节中获取编码信息.

/* 整数编码标识使用了5和6bit位 */
#define ZIP_INT_16B (0xc0 | 0<<4)  //16bit整数
#define ZIP_INT_32B (0xc0 | 1<<4)  //32bit整数
#define ZIP_INT_64B (0xc0 | 2<<4)  //64bit整数
#define ZIP_INT_24B (0xc0 | 3<<4)  //24bit整数
#define ZIP_INT_8B 0xfe            //8bit整数

#define ZIP_INT_IMM_MASK 0x0f
#define ZIP_INT_IMM_MIN 0xf1    /* 11110001 */
#define ZIP_INT_IMM_MAX 0xfd    /* 11111101 */

static int zipTryEncoding(unsigned char *entry, unsigned int entrylen, long long *v, unsigned char *encoding) {
    long long value;
    if (entrylen >= 32 || entrylen == 0) return 0;
   
    if (string2ll((char*)entry,entrylen,&value)) {
        /* 0-12之间的值, 直接在保存在了encoding字段中, 其他根据值大小, 直接设置为相应的编码*/
        if (value >= 0 && value <= 12) {
            *encoding = ZIP_INT_IMM_MIN+value;
        } else if (value >= INT8_MIN && value <= INT8_MAX) {
            *encoding = ZIP_INT_8B;
        } else if (value >= INT16_MIN && value <= INT16_MAX) {
            *encoding = ZIP_INT_16B;
        } else if (value >= INT24_MIN && value <= INT24_MAX) {
            *encoding = ZIP_INT_24B;
        } else if (value >= INT32_MIN && value <= INT32_MAX) {
            *encoding = ZIP_INT_32B;
        } else {
            *encoding = ZIP_INT_64B;
        }
        *v = value;
        return 1;
    }
    return 0;
}

连锁更新

当将一个新节点添加到某个节点之前时，如果原节点的header空间不足以保存新节点的长度，那么就需要对原节点的header空间进行扩展（从1字节扩展到5字节）。但是当对这个节点进行扩展后，可能又会引起后续节点的扩展，这就会引起连锁更新。这种情况在多个连续节点的长度都接近 ZIP_BIGLEN 时可能发生。

反过来说，因为节点的长度变小而引起的连续缩小也是可能出现的，为了避免扩展-缩小-扩展-缩小这样的情况反复出现（flapping，抖动），我们不处理这种情况，而是任由 prevlen 比所需的长度更长。

inset.c/inset.h

1
2
3
4
5
6
7
8
9
10
11
12

typedef struct intset {
    // 编码方式
    uint32_t encoding;
    // 集合包含的元素数量
    uint32_t length;
    // 保存元素的数组
    int8_t contents[];
} intset;

#define INTSET_ENC_INT16 (sizeof(int16_t))
#define INTSET_ENC_INT32 (sizeof(int32_t))
#define INTSET_ENC_INT64 (sizeof(int64_t))

length记录了整数集合中包含的元素个数，即contents数组的长度。
contents数组是整数集合的底层实现，整数集合中的一个元素就是contents数组的一个元素，其中的元素都是按照数值从小到大排序的，并且元素不重复。虽然contents数组被定义为int8_t类型，但其中的元素类型取决于encoding。

inset是Redis中的整数集合数据结构，只允许保存整数值。
inset之所以有三种表示编码格式的宏定义，是因为根据存储元素数值的大小，能够选取一个最“合适”的类型存储，既能够表示元素的大小，又可以节省空间。当插入的新元素编码要大于当前集合编码格式，则需要进行升级操作。

升级步骤主要分成三步：

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

   static intset *intsetUpgradeAndAdd(intset *is, int64_t value) { // 当前的编码方式 uint8_t curenc = intrev32ifbe(is->encoding); // 新值所需的编码方式 uint8_t newenc = _intsetValueEncoding(value); // 当前集合的元素数量 int length = intrev32ifbe(is->length); // 根据 value 的值，决定是将它添加到底层数组的最前端还是最后端 // 注意，因为 value 的编码比集合原有的其他元素的编码都要大 // 所以 value 要么大于集合中的所有元素，要么小于集合中的所有元素 // 因此，value 只能添加到底层数组的最前端或最后端 int prepend = value < 0 ? 1 : 0; /* First set new encoding and resize */ // 更新集合的编码方式 is->encoding = intrev32ifbe(newenc); // 根据新编码对集合（的底层数组）进行空间调整 // T = O(N) is = intsetResize(is,intrev32ifbe(is->length)+1); // T = O(N) while(length--) _intsetSet(is,length+prepend,_intsetGetEncoded(is,length,curenc)); /* Set the value at the beginning or the end. */ // 设置新值，根据 prepend 的值来决定是添加到数组头还是数组尾 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
19
20
21
22
23
24
25
26
27

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 = (Enc_t*)is->contents+from 记录移动开始的位置
    // dst = (Enc_t*)is_.contents+to 记录移动结束的位置
    // bytes *= sizeof(Enc_t) 计算一共要移动多少字节
    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);
    }
    // 进行移动
    // T = O(N)
    memmove(dst,src,bytes);
}

1
2
3
4
5
6
7
8
9
10
11
12
13

### API
创建并返回一个空的整数集合
intset *intsetNew(void);
static intset *intsetResize(intset *is, uint32_t len);//重新分配contents数组指定长度的内存
//用二分查找在contents数组中查找value，返回pos
static uint8_t intsetSearch(intset *is, int64_t value, uint32_t *pos);
//添加value到contents正确的位置，保证inset的有序性
intset *intsetAdd(intset *is, int64_t value, uint8_t *success);
//从整数集合中删除指定value
intset *intsetRemove(intset *is, int64_t value, int *success);
uint8_t intsetFind(intset *is, int64_t value);//在inset查找value
int64_t intsetRandom(intset *is);//从整数集合中随机获取一个value
uint8_t intsetGet(intset *is, uint32_t pos, int64_t *value);//获取整数集合指定位置的value