redis之列表命令源码解析

形象化设计模式实战             HELLO!架构                     redis命令源码解析

 

一、Lpush，Rpush

t_list.c

 

void lpushCommand(redisClient *c) {
    pushGenericCommand(c,REDIS_HEAD);
}

void rpushCommand(redisClient *c) {
    pushGenericCommand(c,REDIS_TAIL);
}

lpush插入列表头部，rpush插入列表尾部

void pushGenericCommand(redisClient *c, int where) {

    int j, waiting = 0, pushed = 0;

    // 取出列表对象
    robj *lobj = lookupKeyWrite(c->db,c->argv[1]);

    // 如果列表对象不存在，那么可能有客户端在等待这个键的出现
    int may_have_waiting_clients = (lobj == NULL);

    if (lobj && lobj->type != REDIS_LIST) {
        addReply(c,shared.wrongtypeerr);
        return;
    }

    // 将列表状态设置为就绪
    if (may_have_waiting_clients) signalListAsReady(c,c->argv[1]);

    // 遍历所有输入值，并将它们添加到列表中
    for (j = 2; j < c->argc; j++) {

        // 编码值
        c->argv[j] = tryObjectEncoding(c->argv[j]);

        // 如果列表对象不存在，那么创建一个，并关联到数据库
        if (!lobj) {
            lobj = createZiplistObject();
            dbAdd(c->db,c->argv[1],lobj);
        }

        // 将值推入到列表
        listTypePush(lobj,c->argv[j],where);

        pushed++;
    }

    // 返回添加的节点数量
    addReplyLongLong(c, waiting + (lobj ? listTypeLength(lobj) : 0));

    // 如果至少有一个元素被成功推入，那么执行以下代码
    if (pushed) {
        char *event = (where == REDIS_HEAD) ? "lpush" : "rpush";

        // 发送键修改信号
        signalModifiedKey(c->db,c->argv[1]);

        // 发送事件通知
        notifyKeyspaceEvent(REDIS_NOTIFY_LIST,event,c->argv[1],c->db->id);
    }

    server.dirty += pushed;
}

1、从这代码可以看出，可以一次插入多个数据，lpush test 1 2 3 4....。

 

官网解释：

It is possible to push multiple elements using a single command call just specifying multiple arguments at the end of the command. Elements are inserted one after the other to the head of the list, from the leftmost element to the rightmost element. So for instance the command LPUSH mylist a b c will result into a list containing c as first element, b as second element and a as third element.

 

 

2、最终实现插入列表的函数是listTypePush

 

二、listTypePush

 

void listTypePush(robj *subject, robj *value, int where) {

    // 是否需要转换编码？
    listTypeTryConversion(subject,value);

    if (subject->encoding == REDIS_ENCODING_ZIPLIST &&
        ziplistLen(subject->ptr) >= server.list_max_ziplist_entries)//注意这里，类似哈希表的类型转换
            listTypeConvert(subject,REDIS_ENCODING_LINKEDLIST);

    // ZIPLIST
    if (subject->encoding == REDIS_ENCODING_ZIPLIST) {
        int pos = (where == REDIS_HEAD) ? ZIPLIST_HEAD : ZIPLIST_TAIL;
        // 取出对象的值，因为 ZIPLIST 只能保存字符串或整数
        value = getDecodedObject(value);
        subject->ptr = ziplistPush(subject->ptr,value->ptr,sdslen(value->ptr),pos);
        decrRefCount(value);

    // 双端链表
    } else if (subject->encoding == REDIS_ENCODING_LINKEDLIST) {
        if (where == REDIS_HEAD) {
            listAddNodeHead(subject->ptr,value);
        } else {
            listAddNodeTail(subject->ptr,value);
        }
        incrRefCount(value);

    // 未知编码
    } else {
        redisPanic("Unknown list encoding");
    }
}

 

 

 

void listTypeTryConversion(robj *subject, robj *value) {

    // 确保 subject 为 ZIPLIST 编码
    if (subject->encoding != REDIS_ENCODING_ZIPLIST) return;

    if (sdsEncodedObject(value) &&
        // 看字符串是否过长
        sdslen(value->ptr) > server.list_max_ziplist_value)
            // 将编码转换为双端链表
            listTypeConvert(subject,REDIS_ENCODING_LINKEDLIST);
}

1、当插入的数据的长度大于list_max_ziplist_value时，转换类型为REDIS_ENCODING_LINKEDLIST（双端列表）。

 

2、当压缩列表的节点数大于list_max_ziplist_entries时，转换类型为REDIS_ENCODING_LINKEDLIST（双端列表）。

双端列表是很常见的数据结构，具体实现请点击

 

三、list_max_ziplist_value与list_max_ziplist_entries

在redis.conf中找到相关设置

 

如果你还记得哈希表的话，会发现这两值与哈希表从压缩列表转换为字典的限制值是相等的，这也不足为奇，因为它们都是由压缩列表转换。

 

 

四、列表阻塞命令：blpop、brpop、brpoplpush

这命令可能用的比较少，就是当列表有元素时就pop，没有就阻塞客户端，直到timeout。下面来试试看：

停在这里不动了，阻塞了！

（0表示无限时阻塞）

不管你再输入什么命令，客户端都没有反应，哪怕你直接ctrl+c。

那我再开一个客户端，往test里lpush一个数据,lpush test 1，再回看阻塞的客户端

返回了值，并且阻塞解除了。

这里注意的是，

1、如里在阻塞的客户端运行lpush会解除阻塞吗？答案是否，就是自己不能解除自己的阻塞。

2、blpop可以同时对多个列表进行阻塞，只要有一个列表有返回值，阻塞即解除。

3、如果有其他客户端访问阻塞的列表，只要不是阻塞命令都不会被阻塞。

 

在这里需要介绍一下客户端的几个参数：（因为参数数量很多，此处不全列出，具体可参看源码）

1、redisDb *db;当前正在使用的数据库

 

typedef struct redisDb {

    // 数据库键空间，保存着数据库中的所有键值对
    dict *dict;                 /* The keyspace for this DB */

    // 键的过期时间，字典的键为键，字典的值为过期事件 UNIX 时间戳。之前在讲字典结构时说道过期时间结构也是一个字典。
    dict *expires;              /* Timeout of keys with a timeout set */

    // 正处于阻塞状态的键（注意这个参数）
    dict *blocking_keys;        /* Keys with clients waiting for data (BLPOP) */

    // 可以解除阻塞的键<span style="font-family: Arial, Helvetica, sans-serif;">（注意这个参数）</span>
    dict *ready_keys;           /* Blocked keys that received a PUSH */

    // 正在被 WATCH 命令监视的键
    dict *watched_keys;         /* WATCHED keys for MULTI/EXEC CAS */

    struct evictionPoolEntry *eviction_pool;    /* Eviction pool of keys */

    // 数据库号码
    int id;                     /* Database ID */

    // 数据库的键的平均 TTL ，统计信息
    long long avg_ttl;          /* Average TTL, just for stats */

} redisDb;

标红处是本文所要关注的

 

2、blockingState bpop;     /* blocking state */

 

typedef struct blockingState {

    /* Generic fields. */
    // 阻塞时限
    mstime_t timeout;       /* Blocking operation timeout. If UNIX current time
                             * is > timeout then the operation timed out. */

    /* REDIS_BLOCK_LIST */
    // 造成阻塞的键
    dict *keys;             /* The keys we are waiting to terminate a blocking
                             * operation such as BLPOP. Otherwise NULL. *
    // 在被阻塞的键有新元素进入时，需要将这些新元素添加到哪里的目标键
    // 用于 BRPOPLPUSH 命令
    robj *target;           /* The key that should receive the element,
                             * for BRPOPLPUSH. */

    /* REDIS_BLOCK_WAIT */
    // 等待 ACK 的复制节点数量
    int numreplicas;        /* Number of replicas we are waiting for ACK. */
    // 复制偏移量
    long long reploffset;   /* Replication offset to reach. */

} blockingState;

 

五、阻塞的内部实现

 

1. 将客户端的状态设为“正在阻塞” ，并记录阻塞这个客户端的各个键（记录到bpop.keys中），以及阻塞的最长时限（bpop.timeout）等数据。
2. 将客户端的信息记录到server.db[i]->blocking_keys 中（其中i 为客户端所使用的数据库号码）。
3. 继续维持客户端和服务器之间的网络连接，但不再向客户端传送任何信息，造成客户端阻塞。

 

步骤2 是将来解除阻塞的关键，server.db[i]->blocking_keys 是一个字典，字典的键是那些造成客户端阻塞的键，而字典的值是一个链表，链表里保存了所有因为这个键而被阻塞的客户端（被同一个键所阻塞的客户端可能不止一个）。

 

六、解除阻塞的内部实现

脱离阻塞状态有以下三种方法：
1. 被动脱离：有其他客户端为造成阻塞的键推入了新元素。
2. 主动脱离：到达执行阻塞原语时设定的最大阻塞时间。
3. 强制脱离：客户端强制终止和服务器的连接，或者服务器停机。

 

这里看下lpush是如何解除阻塞的。

1. 检查这个键是否存在于前面提到的server.db[i]->blocking_keys 字典里，如果是的话，那么说明有至少一个客户端因为这个key 而被阻塞，程序会为这个键创建一个
redis.h/readyList 结构，并将它添加到server.ready_keys 链表中。（在signalListAsReady函数中实现）
2. 将给定的值添加到列表键中。

好像并没有解除阻塞啊，只是添加到了server.ready_keys 链表中。Redis 的主进程在执行完pushGenericCommand 函数之后，会继续调用handleClientsBlockedOnLists 函数来完成解除阻塞，大致图解如下：

 

解除阻塞依据FBFS策略（先阻塞先服务）

 

最大阻塞时间解除在redis.c中的clientsCronHandleTimeout中有：

        // 检查被 BLPOP 等命令阻塞的客户端的阻塞时间是否已经到达
        // 如果是的话，取消客户端的阻塞
        if (c->bpop.timeout != 0 && c->bpop.timeout < now_ms) {
            // 向客户端返回空回复
            replyToBlockedClientTimedOut(c);
            // 取消客户端的阻塞状态
            unblockClient(c);
        }

有兴趣可以深入看下源码，这里不作分析。