Libz's Blog

1
/* 控制SDS的头类型大小 */
2
#define SDS_TYPE_5 0
3
#define SDS_TYPE_8 1
4
#define SDS_TYPE_16 2
5
#define SDS_TYPE_32 3
6
#define SDS_TYPE_64 4
7

8
/* sdshdr8 结构体名称 . 有很多个相同结构 如上面... */
9
struct _attribute_ ((_packed_)) sdshdr8 {
10
    /* uint8_t : 无符号8个bit位 */
11
    /* buf已保存的字符串字节数 , 不包含结束标识 */
12
    uint8_t len;
13
    /* buf申请的总字节数 , 不包含结束标识 */
14
    uint8_t alloc;
15
    /* 不同SDS的头类型 , 用来控制SDS的头大小 */
16
    unsigned char flags;
17
    char buf[];
18
}

1
/* 存放 encoding , 表示不同整型大小 */
2
/* 2字节 整数 , 类以java的short */
3
#define INTSET_ENC16 (sizeof(nt16t_t))
4
/* 4字节 整数 , 类以java的int */
5
#define INTSET_ENC32 (sizeof(nt32t_t))
6
/* 8字节 整数 , 类以java的long */
7
#define INTSET_ENC64 (sizeof(int64_t))
8

9
typedef struct intset {
10
    /* 编码方式 , 支特存放16位、32位、64位整数 */
11
  uint32_t encoding;
12
    /* 元素个数 */
13
  uint32_t length;
14
    /* 整数数组 , 保存集合数据 */
15
  int8_t contents[];
16
}intset;

1
typedef struct dict {
2
    // dict类型 , 内置不同的hash函数
3
  dictType *type;
4
    // 私有数据 , 在做特殊hash运算时用
5
  void *privdata;
6
    // 一个Dict包含两个哈希表 , 一个是当前数据 , 另一个 rehash时使用(一般为空)
7
  dictht ht[2];
8
    // rehash的进度 , -1:未进行; rehashidx > -1:运行
9
  long rehashidx;
10
    // rehash是否暂停 , 1:暂停;0:继续
11
  int16_t pauserehash;
12
}dict;
13

14
typedef struct dictht {
15
  // entry数组
16
  // 数组中保存的是指向entry的指针
17
  dictEntry **table;
18
  // 哈希表大小
19
  unsigned long size;
20
  // 哈希表大小的掩码 , 总等于size-1
21
  unsigned long sizemask;
22
  // entry个数
23
  unsigned long used;
24
}dictht;
25

26
// 哈希节点 dictEntry
27
typedef struct dictEntry {
28
    // 键 key
29
  void *key;
30
    // 值 val
31
  union {
32
    void *val;
33
    uint64_t u64;
34
    int64_t s64;
35
    double d;
36
  };
37
  // 下一个Entry的指针
38
  struct dictEntry *next;
39
}dictEntry;

1
typedef struct ziplist {
2

3
    // 记录整个ZipList的字节数(4字节)
4
    unsigned uint32_t zlbytes;
5
    // 指向ZipList的最后一个节点的偏移量(从开头指向最后元素的开始)(4字节)
6
    unsigned uint32_t zltail;
7
    // 记录ZipList中元素总量(2字节)
8
    unsigned uint16_t zllen;
9

10
    // 以下是实际的数据存储空间 , 这个空间是紧凑的字节数组
11
    // 每个节点包含一个前置节点长度的字段prevlen和一个实际存储数据的entry字段
12
    // entry字段可以是字符串、整数或者浮点数 , 长度根据实际数据类型使用不同的字节数
13
    unsigned entry data[];
14
    // 结束符(1字节)
15
    unsigned uint8_t zlend;
16

17
} ziplist;

1
typedef struct quicklist {
2
    // 头节点指针
3
    quicklistNode *head;
4
    // 尾节点指针
5
    quicklistNode *tail;
6
    // 所有ziplist的entry数量
7
    unsigned long count;
8
    // ziplists总数量
9
    unsigned long len;
10
    // ziplist的entry上限 , 默认-2
11
    int fill : QL_FILL_BITS;
12
    // 首尾不压缩的节点数量
13
    unsigned int compress : QL_COMP_BITS;
14
    // 内存重分配是的书签数量以及数组(一般怎么用)
15
    unsigned int bookmark_count: QL_BM_BITS;
16
    quicklistBookmark bookmarks[];
17
} quicklist;
18

19
// ziplist节点
20
typedef struct quicklistNode {
21
    // 前一个节点指针
22
    struct quicklistNode *prev;
23
    // 下一个节点指针
24
    struct quicklistNode *next;
25
    // 当前ziplist节点 指针
26
    unsigned char *zl;
27
    // 当前ziplist节点 字节大小
28
    unsigned int sz;
29
  // 当前ziplist节点 entry个数
30
    unsigned int count : 16;
31
    // 编码方式: 1.ziplist; 2.lzf压缩
32
    unsigned int encoding : 2;
33
    // 数据容器类型: 1.其他; 2.ziplist
34
    unsigned int container : 2;
35
    // 当前ziplist节点 是否被解压缩: 1.被解压了, 需要重新压缩
36
    unsigned int recompress : 1;
37
    // 测试用
38
    unsigned int attempted_compress : 1;
39
    // 预留字段
40
    unsigned int extra : 10;
41
} quicklistNode;

1
typedef struct zskiplist {
2
  // 头尾节点指针
3
  struct zskiplistNode *header, *tail;
4
  // 节点数量
5
  unsigned long length;
6
  // 最大的索引层级(默认1 ;最大20)
7
  int level;
8
}zskiplist;
9

10
typedef struct zskiplistNode {
11
    // 节点存储的值
12
  sds ele;
13
  // 节点分数 , 排序、查找用
14
  double score;
15
    // 前一个节点指针
16
  struct zskiplistNode *backward;
17
    // 多级 索引数组
18
  struct zskiplistLevel {
19
        // 下一个节点指针
20
    struct zskiplistNode *forward;
21
        // 索引跨度
22
    unsigned long span;
23
  }level[0]:
24
}zskiplistNode;

1
typedef struct redisobject {
2
    // 对象类型(4bit)(string;hsah;list;set;zset)
3
  unsigned type:4;
4
    // 底层编码方式(4bit)
5
  unsigned encoding:4;
6
    // 最后一次访问的时间戳(24bit)(判断闲置过久的key)
7
    // LRU: 以秒单位记录最近一次访问时间 , 长度(24bit)
8
    // LFU:
9
    //      - 高16位以分钟为单位记录最近一次访问时间
10
    //     - 低8位记录逻辑访问次数
11
  unsigned lru:LRU_BITS;
12
    // 对象引用计数器(4bytes)(0:对象无人引用; 每次引用+1)
13
  int refcount;
14
    // 真实对象数据指针(8bytes)
15
  void *ptr;
16
}robj;

1
/** push源码
2
    client: 客户端信息,当中包含有命令
3
    where: 0:头新增; 1:尾新增
4
    xx: 当前push的key存在
5
**/
6
void pushGenericCommand(client *c, int where, int xx) {
7
    int j;
8
  // 判断元素大小 , 不能超过LIST_MAX_ITEM_SIZE
9
    // LPUSH key v1 v2 ... (跳过前面两个 , 从2开始
10
    for (j = 2; j < c->argc; j++) {
11
        // 遍历参数 argv[]
12
        if (sdslen(c->argv[j]->ptr) > LIST_MAX_ITEM_SIZE) {
13
            addReplyError(c, "Element too large");
14
            return;
15
        }
16
    }
17

18
    // 尝试找到KEY对应的list . 选择指定库db , argv[1]: key
19
    robj *lobj = lookupKeyWrite(c->db, c->argv[1]);
20
    // 检查类型是否正确
21
    if (checkType(c,lobj,OBJ_LIST)) return;
22
    // 判断空
23
    if (!lobj) {
24
        if (xx) {
25
            addReply(c, shared.czero);
26
            return;
27
        }
28
    // 为空情况 , 创建新的QuickList
29
        lobj = createQuicklistObject();
30
        quicklistSetOptions(lobj->ptr, server.list_max_ziplist_size,server.list_compress_depth);
31
        dbAdd(c->db,c->argv[1],lobj);
32
    }
33

34
    for (j = 2; j < c->argc; j++) {
35
        listTypePush(lobj,c->argv[j],where);
36
        server.dirty++;
37
    }
38

39
    addReplyLongLong(c, listTypeLength(lobj));
40

41
    char *event = (where == LIST_HEAD) ? "lpush" : "rpush";
42
    signalModifiedKey(c,c->db,c->argv[1]);
43
    notifyKeyspaceEvent(NOTIFY_LIST,event,c->argv[1],c->db->id);
44
}

1
// zset结构
2
typedef struct zset {
3
  // Dict指针
4
  dict *dict;
5
  // SkipList指针
6
  zskiplist *zsl;
7
}zset;
8

9
// 创建
10
robj *createZsetobject(void) {
11
  zset *zm = zmalloc(sizeof(*zs));
12
  robj *o;
13
  // 创建Dict
14
  zs->dict = dictCreate(&zsetDictType,NULL);
15
  // 创建SkipList
16
  zs->zsl = zslCreate();
17
  o = createobject(OBJ_ZSET,zs);
18
  o->encoding = OBJ_ENCODING_SKIPLIST;
19
  return o;
20
}

1
// zadd添加元素时 , 先根据key找到zset,不存在则创建新的zset
2
zobj lookupKeyWrite(c->db,key);
3
if (checkType(c,zobj,OBJ_ZSET))goto cleanup;
4
  //判断是否存在
5
if(Zobj==NULL){
6
    // 判断采用 HT+SkipList/ZipList 方案
7
    // server.zset_max_ziplist_entries配置 禁用/启用
8
    // 或者 value大小超过了zset_max_ziplist_value
9
  if (server.zset_max_ziplist_entries == 0 || server.zset_max_ziplist_value sdslen(c->argv[scoreidx+1]->ptr)) {
10
    // 采用 HT+SkipList
11
    zobj = createzsetobject();
12
  }else{
13
        // 采用 ZipList
14
    zobj = createzsetziplistobject();
15
    }
16
  dbAdd(c->db,key,zobj);
17
}
18
//..
19
zsetAdd(zobj,score,ele,flags,&retflags,&newscore);

1
$25\r\n
2
This is a multi-line string\r\n
3
that spans multiple lines.\r\n
4
\r\n

1
*3\r\n
2
$3\r\nset\r\n
3
$4\r\nname\r\n
4
$4\r\nSans\r\n

1
public class Main {
2

3
    public static Socket socket = null;
4
    public static PrintWriter writer;
5
    public static BufferedReader reader;
6

7
    public static void main(String[] args) {
8
        String host = "192.xxx.xx";
9
        String password = "123123";
10
        int port = 6379;
11

12
        try {
13
            // 1. 建立连接
14
            socket = new Socket(host, port);
15
            // 2. 获取 输入/输出 流 (字符流)
16
            writer = new PrintWriter(
17
                    new OutputStreamWriter(socket.getOutputStream(), StandardCharsets.UTF_8));
18
            reader = new BufferedReader(
19
                    new InputStreamReader(socket.getInputStream(), StandardCharsets.UTF_8));
20
            // 3. 认证
21
            sendRequest("auth", password);
22
            Object obj = handleResponse();
23
            System.out.println("登录 => " + obj);
24
            // 4. 发送 命令请求
25
            // set测试
26
            //sendRequest("set", "name2", "张三");
27
            // get测试
28
            //sendRequest("get","name");
29
            sendRequest("mget", "name", "name2");
30
            // 异常测试 (不存在)
31
            //sendRequest("get","xa");
32
            // 5. 解析响应
33
            obj = handleResponse();
34
            System.out.println("res => " + obj);
35
        } catch (Exception e) {
36
            throw new RuntimeException(e);
37
        } finally {
38
            try {
39
                // 6. 释放连接
40
                if (reader != null) reader.close();
41
                if (writer != null) writer.close();
42
                if (socket != null) socket.close();
43
            } catch (Exception e) {
44
                e.printStackTrace();
45
            }
46

47
        }
48
    }
49

50
    private static Object handleResponse() throws IOException {
51
        // 读取首字节 , 判断类型
52
        int prefix = reader.read();
53
        switch (prefix) {
54
            case '+':
55
                return reader.readLine();
56
            case '-':
57
                throw new RuntimeException(reader.readLine());
58
            case ':':
59
                return Long.parseLong(reader.readLine());
60
            case ': // 新读取长度 , 后读数据
61
                long len = Long.parseLong(reader.readLine());
62
                if (len == -1) return null;
63
                if (len == 0) return "";
64
                return reader.readLine();
65
            case '*':
66
                return readBulkString();
67
            default:
68
                throw new RuntimeException("未知格式");
69
        }
70
    }
71

72
    private static Object readBulkString() throws IOException {
73
        // 循环读取
74
        long len = Long.parseLong(reader.readLine());
75
        if (len <= 0) return null;
76
        List<Object> list = new ArrayList<>();
77
        for (int i = 0; i < len; i++) {
78
            list.add(handleResponse());
79
        }
80
        return list;
81
    }
82

83
    private static void sendRequest(String... ages) {
84
        writer.println("*" + ages.length);
85
        for (String age : ages) {
86
            // 字节大小(并非字符个数)
87
            writer.println("$" + age.getBytes(StandardCharsets.UTF_8).length);
88
            writer.println(age);
89
        }
90
        writer.flush();
91
    }
92

93
}

1
typedef struct redisDb {
2
    // 存储 key-value本身
3
    dict *dict;
4
    // 存过期时间 key-ttl
5
    dict *expires;
6
    dict *blocking_keys;
7
    dict *ready_keys;
8
    dict *watched_keys;
9
    int id;                      /* Database ID , 0~15 */
10
    long long avg_ttl;           /* 记录平均TTL时长 */
11
    unsigned long expires_cursor;/* expire检查时在dict中抽样的索引位置. */
12
    list *defrag_later;          /* 等待碎片整理的key列表. */
13
} redisDb;

1
// 查找一个key执行写操作
2
robj *lookupKeyWriteWithFlags(redisDb *db, robj *key, int flags) {
3
    // 检查key是否过期
4
    expireIfNeeded(db,key);
5
    return lookupKey(db,key,flags);
6
}
7
// 查找一个key执行读操作
8
robj *lookupKeyReadWithFlags(redisDb *db, robj *key, int flags) {
9
    robj *val;
10
    // 检查key是否过期
11
    if (expireIfNeeded(db,key) == 1) {
12
        // ...
13
    }
14
    return NULL;
15
}
16

17
int expireIfNeeded(redisDb *db, robj *key) {
18
    // 判断是否过期 , 如果未过期直接结束并返回0
19
    if (!keyIsExpired(db,key)) return 0;
20
    // ...
21
    // 删除过期key
22
    deleteExpiredKeyAndPropagate(db,key);
23
    return 1;
24
}

1
// 初始化
2
void initServer(void){
3
    // ...
4
    // 创建定时器 , 关联回调函数serverCron , 处理周期取决于server.hz , 默认10
5
    aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL);
6
}
7

8
// 定期执行
9
int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
10
    // 更新lruclock到当前时间 , 为后期的LRU和LFU做准备
11
    unsigned int lruclock = getLRUClock();
12
    atomicSet(server.lruclock,lruclock);
13
    // 执行database的数据清理 , 例如过期key处理
14
    databasesCron();
15
    // 返回等待加载下一周期的时间 100(1000/10)
16
    return 1000/server.hz;
17
}
18

19
// 清除任务
20
void databasesCron(void) {
21
    // 尝试清理部分过期key , 清理模式默认为SLOW
22
    activeExpireCycle(ACTIVE_EXPIRE_CYCLE_SLOW);
23
}

1
// 事件触发
2
void aeMain(aeEventLoop *eventLoop){
3
    eventLoop->stop = 0;
4
    while(!eventLoop->stop){
5
        // beforeSleep() --> FAST模式清理
6
        // n = aeApiPoll()
7
        // 如果 n>0 , FD就绪 , 处理IO事件
8
        // 如果到了执行时间 , 则调用 serverCron() --> SLOW模式清理
9
    }
10
}
11

12
// 清除任务
13
void beforeSleep(struct aeEventLoop *eventLoop){
14
    // ...
15
    // 尝试清理部分过期key , 清理模式默认为FAST
16
    activeExpireCycle(ACTIVE_EXPIRE_CYCLE_FAST);
17
}