阿里云-云小站（无限量代金券发放中）
【腾讯云】云服务器、云数据库、COS、CDN、短信等热卖云产品特惠抢购

Redis源码分析之事务Transaction

215次阅读
没有评论

共计 13401 个字符，预计需要花费 34 分钟才能阅读完成。

这周学习了一下 Redis 事务功能的实现原理，本来是想用一篇文章进行总结的，写完以后发现这块内容比较多，而且多个命令之间又互相依赖，放在一篇文章里一方面篇幅会比较大，另一方面文章组织结构会比较乱，不容易阅读。因此把事务这个模块整理成上下两篇文章进行总结。

这篇文章我们重点分析一下 redis 事务命令中的两个辅助命令：watch 跟 unwatch。

一、redis 事务辅助命令简介

依然从 server.c 文件的命令表中找到相应的命令以及它们对应的处理函数。

//watch，unwatch 两个命令我们把它们叫做 redis 事务辅助命令
{"watch",watchCommand,-2,"sF",0,NULL,1,-1,1,0,0},
{"unwatch",unwatchCommand,1,"sF",0,NULL,0,0,0,0,0},
  1. watch，用于客户端关注某个 key，当这个 key 的值被修改时，整个事务就会执行失败（注：该命令需要在事务开启前使用）。
  2. unwatch，用于客户端取消已经 watch 的 key。

用法举例如下：
clientA

127.0.0.1:6379> watch a
OK
127.0.0.1:6379> multi
OK
127.0.0.1:6379> set b b
QUEUED
// 在执行前插入 clientB 的操作如下，事务就会执行失败
127.0.0.1:6379> exec
(nil)
127.0.0.1:6379>

clientB

127.0.0.1:6379> set a aa
OK
127.0.0.1:6379>

二、redis 事务辅助命令源码分析

在看具体执行函数之前首先了解几个数据结构：

// 每个客户端对象中有一个 watched_keys 链表来保存已经 watch 的 key
typedef struct client {list *watched_keys;  
}
// 上述链表中每个节点的数据结构
typedef struct watchedKey {//watch 的 key
    robj *key;
    // 指向的 DB，后面细说
    redisDb *db;
} watchedKey;

关于事务的几个命令所对应的函数都放在了 multi.c 文件中。
一起看下 watch 命令对应处理函数的源码：

void watchCommand(client *c) {
    int j;
    // 如果客户端处于事务状态，则返回错误信息
    // 由此可以看出，watch 必须在事务开启前使用
    if (c->flags & CLIENT_MULTI) {addReplyError(c,"WATCH inside MULTI is not allowed");
        return;
    }
    // 依次 watch 客户端的各个参数（这里说明 watch 命令可以一次 watch 多个 key）
    // 注：0 表示命令本身，所以参数从 1 开始
    for (j = 1; j < c->argc; j++)
        watchForKey(c,c->argv[j]);
    // 返回结果
    addReply(c,shared.ok);
}
 
// 具体的 watch 操作，代码较长，慢慢分析
void watchForKey(client *c, robj *key) {list *clients = NULL;
    listIter li;
    listNode *ln;
    // 上面已经提到了数据结构
    watchedKey *wk;
 
    // 首先判断 key 是否已经被客户端 watch
    //listRewind 这个函数在发布订阅那篇文章里也有，就是把客户端的 watched_keys 赋值给 li
    listRewind(c->watched_keys,&li);
    while((ln = listNext(&li))) {wk = listNodeValue(ln);
        // 这里一个 wk 节点中有 db，key 两个字段
        if (wk->db == c->db && equalStringObjects(key,wk->key))
            return; 
    }
    // 开始 watch 指定 key
    // 整个 watch 操作保存了两套数据结构，一套是在 db->watched_keys 中的字典结构，如下：
    clients = dictFetchValue(c->db->watched_keys,key);
    // 如果是 key 第一次出现，则进行初始化
    if (!clients) {clients = listCreate();
        dictAdd(c->db->watched_keys,key,clients);
        incrRefCount(key);
    }
    // 把当前客户端加到该 key 的 watch 链表中
    listAddNodeTail(clients,c);
    // 另一套是在 c ->watched_keys 中的链表结构：如下
    wk = zmalloc(sizeof(*wk));
    // 初始化各个字段
    wk->key = key;
    wk->db = c->db;
    incrRefCount(key);
    // 加入到链表最后
    listAddNodeTail(c->watched_keys,wk);
}

整个 watch 的数据结构比较复杂，我这里画了一张图方便理解：
Redis 源码分析之事务 Transaction
简单解释一下上面的图，首先 redis 把每个客户端连接包装成了一个 client 对象，上图中 db，watch_keys 就是其中的两个字段（client 对象里面还有很多其他字段，包括上篇文章中提到的 pub/sub）。

  1. db 字段指向给该 client 对象分配的储存空间，db 对象 中也含有一个 watched_keys 字段，是字典类型（也就是哈希表），以想要 watch 的 key 做 key，存储的链表则是所有 watch 该 key 的客户端。
  2. watch_keys 字段则是一个链表类型，每个节点类型为 watch_key，其中包含两个字段，key 表示 watch 的 key，db 则指向了当前 client 对象的 db 字段，如上图。

看完 watch 命令的源码以后，再来看一下 unwatch 命令，如果搞明白了上面提到的两套数据结构，那么看 unwatch 的源码应该会比较容易，毕竟就是删除数据结构中对应的内容。

void unwatchCommand(client *c) {// 取消 watch 所有 key
    unwatchAllKeys(c);
    // 修改客户端状态
    c->flags &= (~CLIENT_DIRTY_CAS);
    addReply(c,shared.ok);
}
 
// 取消 watch 的 key
void unwatchAllKeys(client *c) {
    listIter li;
    listNode *ln;
    // 如果客户端没有 watch 任何 key，则直接返回
    if (listLength(c->watched_keys) == 0) return;
    // 注意这里操作的是链表字段
    listRewind(c->watched_keys,&li);
    while((ln = listNext(&li))) {list *clients;
        watchedKey *wk;
        // 遍历取出该客户端 watch 的 key
        wk = listNodeValue(ln);
        // 取出所有 watch 了该 key 的客户端，这里则是字典（即哈希表）
        clients = dictFetchValue(wk->db->watched_keys, wk->key);
        // 空指针判断
        serverAssertWithInfo(c,NULL,clients != NULL);
        // 从 watch 列表中删除该客户端
        listDelNode(clients,listSearchKey(clients,c));
        // 如果 key 只有一个当前客户端 watch，则删除
        if (listLength(clients) == 0)
            dictDelete(wk->db->watched_keys, wk->key);
        // 从当前 client 的 watch 列表中删除该 key
        listDelNode(c->watched_keys,ln);
        // 减少引用数
        decrRefCount(wk->key);
        // 释放内存
        zfree(wk);
    }
}

最后我们考虑一下 watch 机制的触发时机，现在我们已经把想要 watch 的 key 加入到了 watch 的数据结构中，可以想到触发 watch 的时机应该是修改 key 的内容时，通知到所有 watch 了该 key 的客户端。

感兴趣的用户可以任意选一个修改命令跟踪一下源码，例如 set 命令，我们发现所有对 key 进行修改的命令最后都会调用 touchWatchedKey()函数，而该函数源码就位于 multi.c 文件中，该函数就是触发 watch 机制的关键函数，源码如下：

// 这里入参 db 就是客户端对象中的 db，上文已经提到，不赘述
void touchWatchedKey(redisDb *db, robj *key) {list *clients;
    listIter li;
    listNode *ln;
    // 保存 watchkey 的字典为空，则返回
    if (dictSize(db->watched_keys) == 0) return;
    // 注意这里操作的是字典（即哈希表）数据结构
    clients = dictFetchValue(db->watched_keys, key);
    // 如果没有客户端 watch 该 key，则返回
    if (!clients) return;
    // 把 client 赋值给 li
    listRewind(clients,&li);
    // 遍历 watch 了该 key 的客户端，修改他们的状态
    while((ln = listNext(&li))) {client *c = listNodeValue(ln);
        c->flags |= CLIENT_DIRTY_CAS;
    }
}

跟我们猜测的一样，就是每当 key 的内容被修改时，则遍历所有 watch 了该 key 的客户端，设置相应的状态为 CLIENT_DIRTY_CAS。

三、redis 事务辅助命令总结

上面就是 redis 事务命令中 watch，unwatch 的实现原理，其中最复杂的应该就是 watch 对应的那两套数据结构了，跟之前的 pub/sub 类似，都是使用链表 + 哈希表的结构存储，另外也是通过修改客户端的状态位 FLAG 来通知客户端。

代码比较多，而且 C ++ 代码看上去会比较费劲，需要慢慢读，反复读。

更多详情见请继续阅读下一页的精彩内容：http://www.linuxidc.com/Linux/2017-11/148624p2.htm

接着上一篇，这篇文章分析一下 Redis 事务操作中 multi，exec，discard 三个核心命令。

一、redis 事务核心命令简介

redis 事务操作核心命令：

    // 用于开启事务
    {"multi",multiCommand,1,"sF",0,NULL,0,0,0,0,0},
    // 用来执行事务中的命令
    {"exec",execCommand,1,"sM",0,NULL,0,0,0,0,0},
    // 用来取消事务
    {"discard",discardCommand,1,"sF",0,NULL,0,0,0,0,0},

在 redis 中，事务并不具有 ACID 的概念，换句话说，redis 中的事务仅仅是保证一系列的命令按顺序一个一个执行，如果中间失败了，并不会进行回滚操作。

使用 redis 事务举例如下：

127.0.0.1:6379> multi
OK
127.0.0.1:6379> set a a
QUEUED
127.0.0.1:6379> set b b
QUEUED
127.0.0.1:6379> set c c
QUEUED
127.0.0.1:6379> exec
1) OK
2) OK
3) OK
127.0.0.1:6379>

二、redis 事务核心命令源码分析

关于事务的几个命令所对应的函数都放在 multi.c 文件中。

首先来看一下 multi 命令，该命令用于标记客户端开启事务状态，因此它做的就是修改客户端状态，代码很简单，如下：

void multiCommand(client *c) {// 如果客户端已经是事务模式，则返回错误提示信息
    if (c->flags & CLIENT_MULTI) {addReplyError(c,"MULTI calls can not be nested");
        return;
    }
    // 设置客户端为事务模式
    c->flags |= CLIENT_MULTI;
    // 返回结果
    addReply(c,shared.ok);
}

接下来看下 redis 处理命令逻辑中的一段源码：
这段代码在 server.c 文件中的 processCommand 方法中：

    // 如果客户端处于事务状态且当前执行的命令不是 exec，discard，multi 跟 watch 命令中的一个
    // 则把当前命令加入一个队列
    if (c->flags & CLIENT_MULTI &&
        c->cmd->proc != execCommand && c->cmd->proc != discardCommand &&
        c->cmd->proc != multiCommand && c->cmd->proc != watchCommand)
    {// 加入队列
        queueMultiCommand(c);
        // 返回结果
        addReply(c,shared.queued);
    } else {// 执行当前命令
        call(c,CMD_CALL_FULL);
        c->woff = server.master_repl_offset;
        if (listLength(server.ready_keys))
            handleClientsBlockedOnLists();}

看入队操作源码前，先来熟悉几个数据结构，redis 会把每个连接的客户端封装成一个 client 对象，该对象中含有大量字段用来保存需要的信息，发布订阅功能也使用对应的字段进行存储，事务当然也不例外，如下：

// 每个客户端对象中有一个 mstate 字段用来保存事务上下文
typedef struct client {multiState mstate;}
 
// 事务包装类型
typedef struct multiState {// 当前事务中需要执行的命令数组
    multiCmd *commands; 
    // 需要执行的命令数量
    int count;          
    // 需要同步复制的最小数量
    int minreplicas;       
    // 同步复制超时时间
    time_t minreplicas_timeout; 
} multiState;
 
// 事务中执行命令的封装类型
typedef struct multiCmd {// 参数
    robj **argv;
    // 参数数量
    int argc;
    // 命令本身
    struct redisCommand *cmd;
} multiCmd;

了解了基本的数据结构以后，再来看下入队操作：

void queueMultiCommand(client *c) {// 类型前面有说明
    multiCmd *mc;
    int j;
    // 扩容，每次扩容一个命令的大小
    c->mstate.commands = zrealloc(c->mstate.commands,
            sizeof(multiCmd)*(c->mstate.count+1));
    //c++ 中给数组最后一个元素赋值语法实在是有点难懂...
    mc = c->mstate.commands+c->mstate.count;
    // 初始化 mc 各个字段
    mc->cmd = c->cmd;
    mc->argc = c->argc;
    mc->argv = zmalloc(sizeof(robj*)*c->argc);
    // 把参数一个一个拷贝过来
    memcpy(mc->argv,c->argv,sizeof(robj*)*c->argc);
    for (j = 0; j < c->argc; j++)
        incrRefCount(mc->argv[j]);
    c->mstate.count++;
}

上面是把命令加入事务命令数组的中的逻辑，由于在执行事务过程中也会执行删除事务的操作，因此在看执行事务逻辑之前我们先看下删除事务的实现原理。

当事务执行完成，执行错误或者客户端想取消当前事务，都会跟 discard 命令有联系，一起看下源码：

void discardCommand(client *c) {// 如果当前客户端没有处于事务状态，则返回错误信息
    if (!(c->flags & CLIENT_MULTI)) {addReplyError(c,"DISCARD without MULTI");
        return;
    }
    // 删除事务
    discardTransaction(c);
    // 返回结果
    addReply(c,shared.ok);
}
 
// 具体的删除逻辑
void discardTransaction(client *c) {// 释放客户端事务资源
    freeClientMultiState(c);
    // 初始化客户端事务资源
    initClientMultiState(c);
    // 状态位还原
    c->flags &= ~(CLIENT_MULTI|CLIENT_DIRTY_CAS|CLIENT_DIRTY_EXEC);
    // 取消已 watch 的 key，该函数上面文章中已经进行过分析，不赘述
    unwatchAllKeys(c);
}
 
// 释放事务队列中的每个命令
void freeClientMultiState(client *c) {
    int j;
    for (j = 0; j < c->mstate.count; j++) {
        int i;
        multiCmd *mc = c->mstate.commands+j;
        // 挨个释放命令的参数
        for (i = 0; i < mc->argc; i++)
            decrRefCount(mc->argv[i]);
        zfree(mc->argv);
    }
     // 最后释放命令本身
    zfree(c->mstate.commands);
}
 
// 事务相关字段设为初始值
void initClientMultiState(client *c) {c->mstate.commands = NULL;
    c->mstate.count = 0;
}

到这里，我们已经了解了开启事务模式，把各个命令加入到事务命令执行数组中以及取消事务三个模块的执行原理，最后一起看下事务的执行过程，代码较长，需要慢慢看。

把一系列命令加入到事务命令数组中以后，客户端执行 exec 命令就可以把其中的所有命令挨个执行完成了，分析 exec 命令源码之前，我们应该可以想到 redis 的逻辑应该就是从客户端的事务命令数组中取出所有命令一个一个执行，源码如下：

void execCommand(client *c) {
    int j;
    robj **orig_argv;
    int orig_argc;
    struct redisCommand *orig_cmd;
    // 标记是否需要把 MULTI/EXEC 传递到 AOF 或者 slaves 节点
    int must_propagate = 0; 
    // 标记当前 redis 节点是否为主节点
    int was_master = server.masterhost == NULL;
    // 如果客户端没有处于事务状态，则返回错误提示信息
    if (!(c->flags & CLIENT_MULTI)) {addReplyError(c,"EXEC without MULTI");
        return;
    }
 
    // 首先对两个需要终止当前事务的条件进行判断
    //1. 当有 WATCH 的 key 被修改时则终止，返回一个 nullmultibulk 对象
    //2. 当之前有命令加入事务命令数组出错则终止，例如传入的命令参数数量不对，会返回 execaborterr
    if (c->flags & (CLIENT_DIRTY_CAS|CLIENT_DIRTY_EXEC)) {addReply(c, c->flags & CLIENT_DIRTY_EXEC ? shared.execaborterr :
                                                  shared.nullmultibulk);
        // 删除当前事务信息，前面已经分析过，不赘述
        discardTransaction(c);
        goto handle_monitor;
    }
 
    // 把 watch 的 key 都删除，上面文章已经分析过，不赘述
    unwatchAllKeys(c); 
    // 保存当前命令上下文
    orig_argv = c->argv;
    orig_argc = c->argc;
    orig_cmd = c->cmd;
    addReplyMultiBulkLen(c,c->mstate.count);
    // 遍历事务命令数组
    for (j = 0; j < c->mstate.count; j++) {// 把事务队列中的命令参数取出赋值给 client，因为命令是在 client 维度执行的
        c->argc = c->mstate.commands[j].argc;
        c->argv = c->mstate.commands[j].argv;
        c->cmd = c->mstate.commands[j].cmd;
 
        // 同步事务操作到 AOF 或者集群中的从节点
        if (!must_propagate && !(c->cmd->flags & (CMD_READONLY|CMD_ADMIN))) {execCommandPropagateMulti(c);
            must_propagate = 1;
        }
        // 执行具体命令
        call(c,CMD_CALL_FULL);
 
        // 由于命令可以修改参数的值或者数量，因此重新保存命令上下文
        c->mstate.commands[j].argc = c->argc;
        c->mstate.commands[j].argv = c->argv;
        c->mstate.commands[j].cmd = c->cmd;
    }
    // 恢复原始命令上下文
    c->argv = orig_argv;
    c->argc = orig_argc;
    c->cmd = orig_cmd;
    // 事务执行完成，删除该事务，前面已经分析过，不赘述
    discardTransaction(c);
 
    // 确保 EXEC 会进行传递
    if (must_propagate) {int is_master = server.masterhost == NULL;
        server.dirty++;
        if (server.repl_backlog && was_master && !is_master) {char *execcmd = "*1\r\n$4\r\nEXEC\r\n";
            feedReplicationBacklog(execcmd,strlen(execcmd));
        }
    }
//monitor 命令操作
handle_monitor:
    if (listLength(server.monitors) && !server.loading)
        replicationFeedMonitors(c,server.monitors,c->db->id,c->argv,c->argc);
}

上面就是事务命令执行的整个逻辑，可以先排除集群跟 AOF 的同步逻辑，专注理解核心逻辑，代码整体逻辑算是比较清晰的，搞明白了前面的几个模块以后，再看执行逻辑就不会太难。

三、redis 事务命令总结

通过上、下两篇文章对 redis 事务各个命令进行了分析，仔细阅读应该可以了解整个事务执行框架，如果有任何问题或者疑惑，欢迎留言评论。

本文永久更新链接地址：http://www.linuxidc.com/Linux/2017-11/148624.htm

这周学习了一下 Redis 事务功能的实现原理，本来是想用一篇文章进行总结的，写完以后发现这块内容比较多，而且多个命令之间又互相依赖，放在一篇文章里一方面篇幅会比较大，另一方面文章组织结构会比较乱，不容易阅读。因此把事务这个模块整理成上下两篇文章进行总结。

这篇文章我们重点分析一下 redis 事务命令中的两个辅助命令：watch 跟 unwatch。

一、redis 事务辅助命令简介

依然从 server.c 文件的命令表中找到相应的命令以及它们对应的处理函数。

//watch，unwatch 两个命令我们把它们叫做 redis 事务辅助命令
{"watch",watchCommand,-2,"sF",0,NULL,1,-1,1,0,0},
{"unwatch",unwatchCommand,1,"sF",0,NULL,0,0,0,0,0},
  1. watch，用于客户端关注某个 key，当这个 key 的值被修改时，整个事务就会执行失败（注：该命令需要在事务开启前使用）。
  2. unwatch，用于客户端取消已经 watch 的 key。

用法举例如下：
clientA

127.0.0.1:6379> watch a
OK
127.0.0.1:6379> multi
OK
127.0.0.1:6379> set b b
QUEUED
// 在执行前插入 clientB 的操作如下，事务就会执行失败
127.0.0.1:6379> exec
(nil)
127.0.0.1:6379>

clientB

127.0.0.1:6379> set a aa
OK
127.0.0.1:6379>

二、redis 事务辅助命令源码分析

在看具体执行函数之前首先了解几个数据结构：

// 每个客户端对象中有一个 watched_keys 链表来保存已经 watch 的 key
typedef struct client {list *watched_keys;  
}
// 上述链表中每个节点的数据结构
typedef struct watchedKey {//watch 的 key
    robj *key;
    // 指向的 DB，后面细说
    redisDb *db;
} watchedKey;

关于事务的几个命令所对应的函数都放在了 multi.c 文件中。
一起看下 watch 命令对应处理函数的源码：

void watchCommand(client *c) {
    int j;
    // 如果客户端处于事务状态，则返回错误信息
    // 由此可以看出，watch 必须在事务开启前使用
    if (c->flags & CLIENT_MULTI) {addReplyError(c,"WATCH inside MULTI is not allowed");
        return;
    }
    // 依次 watch 客户端的各个参数（这里说明 watch 命令可以一次 watch 多个 key）
    // 注：0 表示命令本身，所以参数从 1 开始
    for (j = 1; j < c->argc; j++)
        watchForKey(c,c->argv[j]);
    // 返回结果
    addReply(c,shared.ok);
}
 
// 具体的 watch 操作，代码较长，慢慢分析
void watchForKey(client *c, robj *key) {list *clients = NULL;
    listIter li;
    listNode *ln;
    // 上面已经提到了数据结构
    watchedKey *wk;
 
    // 首先判断 key 是否已经被客户端 watch
    //listRewind 这个函数在发布订阅那篇文章里也有，就是把客户端的 watched_keys 赋值给 li
    listRewind(c->watched_keys,&li);
    while((ln = listNext(&li))) {wk = listNodeValue(ln);
        // 这里一个 wk 节点中有 db，key 两个字段
        if (wk->db == c->db && equalStringObjects(key,wk->key))
            return; 
    }
    // 开始 watch 指定 key
    // 整个 watch 操作保存了两套数据结构，一套是在 db->watched_keys 中的字典结构，如下：
    clients = dictFetchValue(c->db->watched_keys,key);
    // 如果是 key 第一次出现，则进行初始化
    if (!clients) {clients = listCreate();
        dictAdd(c->db->watched_keys,key,clients);
        incrRefCount(key);
    }
    // 把当前客户端加到该 key 的 watch 链表中
    listAddNodeTail(clients,c);
    // 另一套是在 c ->watched_keys 中的链表结构：如下
    wk = zmalloc(sizeof(*wk));
    // 初始化各个字段
    wk->key = key;
    wk->db = c->db;
    incrRefCount(key);
    // 加入到链表最后
    listAddNodeTail(c->watched_keys,wk);
}

整个 watch 的数据结构比较复杂，我这里画了一张图方便理解：
Redis 源码分析之事务 Transaction
简单解释一下上面的图，首先 redis 把每个客户端连接包装成了一个 client 对象，上图中 db，watch_keys 就是其中的两个字段（client 对象里面还有很多其他字段，包括上篇文章中提到的 pub/sub）。

  1. db 字段指向给该 client 对象分配的储存空间，db 对象 中也含有一个 watched_keys 字段，是字典类型（也就是哈希表），以想要 watch 的 key 做 key，存储的链表则是所有 watch 该 key 的客户端。
  2. watch_keys 字段则是一个链表类型，每个节点类型为 watch_key，其中包含两个字段，key 表示 watch 的 key，db 则指向了当前 client 对象的 db 字段，如上图。

看完 watch 命令的源码以后，再来看一下 unwatch 命令，如果搞明白了上面提到的两套数据结构，那么看 unwatch 的源码应该会比较容易，毕竟就是删除数据结构中对应的内容。

void unwatchCommand(client *c) {// 取消 watch 所有 key
    unwatchAllKeys(c);
    // 修改客户端状态
    c->flags &= (~CLIENT_DIRTY_CAS);
    addReply(c,shared.ok);
}
 
// 取消 watch 的 key
void unwatchAllKeys(client *c) {
    listIter li;
    listNode *ln;
    // 如果客户端没有 watch 任何 key，则直接返回
    if (listLength(c->watched_keys) == 0) return;
    // 注意这里操作的是链表字段
    listRewind(c->watched_keys,&li);
    while((ln = listNext(&li))) {list *clients;
        watchedKey *wk;
        // 遍历取出该客户端 watch 的 key
        wk = listNodeValue(ln);
        // 取出所有 watch 了该 key 的客户端，这里则是字典（即哈希表）
        clients = dictFetchValue(wk->db->watched_keys, wk->key);
        // 空指针判断
        serverAssertWithInfo(c,NULL,clients != NULL);
        // 从 watch 列表中删除该客户端
        listDelNode(clients,listSearchKey(clients,c));
        // 如果 key 只有一个当前客户端 watch，则删除
        if (listLength(clients) == 0)
            dictDelete(wk->db->watched_keys, wk->key);
        // 从当前 client 的 watch 列表中删除该 key
        listDelNode(c->watched_keys,ln);
        // 减少引用数
        decrRefCount(wk->key);
        // 释放内存
        zfree(wk);
    }
}

最后我们考虑一下 watch 机制的触发时机，现在我们已经把想要 watch 的 key 加入到了 watch 的数据结构中，可以想到触发 watch 的时机应该是修改 key 的内容时，通知到所有 watch 了该 key 的客户端。

感兴趣的用户可以任意选一个修改命令跟踪一下源码，例如 set 命令，我们发现所有对 key 进行修改的命令最后都会调用 touchWatchedKey()函数，而该函数源码就位于 multi.c 文件中，该函数就是触发 watch 机制的关键函数，源码如下：

// 这里入参 db 就是客户端对象中的 db，上文已经提到，不赘述
void touchWatchedKey(redisDb *db, robj *key) {list *clients;
    listIter li;
    listNode *ln;
    // 保存 watchkey 的字典为空，则返回
    if (dictSize(db->watched_keys) == 0) return;
    // 注意这里操作的是字典（即哈希表）数据结构
    clients = dictFetchValue(db->watched_keys, key);
    // 如果没有客户端 watch 该 key，则返回
    if (!clients) return;
    // 把 client 赋值给 li
    listRewind(clients,&li);
    // 遍历 watch 了该 key 的客户端，修改他们的状态
    while((ln = listNext(&li))) {client *c = listNodeValue(ln);
        c->flags |= CLIENT_DIRTY_CAS;
    }
}

跟我们猜测的一样，就是每当 key 的内容被修改时，则遍历所有 watch 了该 key 的客户端，设置相应的状态为 CLIENT_DIRTY_CAS。

三、redis 事务辅助命令总结

上面就是 redis 事务命令中 watch，unwatch 的实现原理，其中最复杂的应该就是 watch 对应的那两套数据结构了，跟之前的 pub/sub 类似，都是使用链表 + 哈希表的结构存储，另外也是通过修改客户端的状态位 FLAG 来通知客户端。

代码比较多，而且 C ++ 代码看上去会比较费劲，需要慢慢读，反复读。

更多详情见请继续阅读下一页的精彩内容：http://www.linuxidc.com/Linux/2017-11/148624p2.htm

正文完
星哥玩云-微信公众号
post-qrcode
MongoDB MySQL NoSQL Oracle Oracle教程 Oracle数据库 PostgreSQL Redis事务 Redis源码分析 Redis源码分析之事务Transaction 数据库
发表至： 数据库技术
2022-01-22
 0
星锅
版权声明：本站原创文章，由 星锅 于2022-01-22发表，共计13401字。
转载说明：除特殊说明外本站文章皆由CC-4.0协议发布，转载请注明出处。
【腾讯云】推广者专属福利，新客户无门槛领取总价值高达2860元代金券，每种代金券限量500张，先到先得。
innobackupex的备份和恢复
MySQL数据库备份与恢复
MySQL集群读写分离的自定义实现
在Linux上开始安装 SQL Server on Linux
Redis主从配置及手动切换遇到的问题
SQL Server AlwaysOn 可用性组副本只读设置
Oracle broker 详解
利用SQL Server 2008 R2创建自动备份计划
阿里云-最新活动爆款每日限量供应
评论（没有评论）
验证码
星哥说事
星哥说事
分享互联网知识
阅读量
4049509
评论数
4
文章搜索
热门文章
申请腾讯混元的API Key并且使用LobeChat调用混元AI

申请腾讯混元的API Key并且使用LobeChat调用混元AI

申请腾讯混元的 API Key 并且使用 LobeChat 调用混元 AI 之前星哥写了一篇文章《手把手教拥有...
Docker部署搭建一个开源强大的图书管理系统

Docker部署搭建一个开源强大的图书管理系统

Docker 部署搭建一个开源强大的图书管理系统 大家好，我是星哥，作为一个电子图书收集者，想要管理自己的几百...
基于Docker快速搭建一个开源的IT人员在线工具箱-it-tools

基于Docker快速搭建一个开源的IT人员在线工具箱-it-tools

基于 Docker 快速搭建一个开源的 IT 人员在线工具箱 -it-tools 在日常的开发工作中，我们经常...
让每个人都可以轻松使用Git-腾讯自研Git客户端

让每个人都可以轻松使用Git-腾讯自研Git客户端

让每个人都可以轻松使用 Git- 腾讯自研 Git 客户端 Git，作为当今最流行的版本控制系统，其强大的功能...
使用Docker部署开源的WPS-Office

使用Docker部署开源的WPS-Office

使用 Docker 部署开源的 WPS-Office 一、前言 越来越多的企业和个人开始将应用程序部署在 Do...
CDN
阿里云CDN-提高用户访问的响应速度和成功率
随机文章
Linux修改网卡名称

Linux修改网卡名称

导读 网卡，即网络接口板，又称网络适配器或 NIC (网络接口控制器)，是一块被设计用来允许计算机在计算机网络...
解决修复ntp漏洞

解决修复ntp漏洞

CVE-2013-5211 漏洞说明： CVE-2013-5211 最早公布是 2014 年 1 月 10 日...
腾讯云服务器买一年送3个月优惠：2核、4核、8核和16核配置价格表

腾讯云服务器买一年送3个月优惠：2核、4核、8核和16核配置价格表

腾讯云服务器买一年送 3 个月优惠活动，轻量应用服务器和云服务器 CVM 均可享受，买一年送 3 个月入口：w...
Docker部署开源私有云相册，给你的照片一个家

Docker部署开源私有云相册，给你的照片一个家

Docker 部署开源私有云相册，给你的照片一个家 大家好，我是星哥，在这个数字化时代，照片不仅是一种记录生活...
Centos之赛门铁克存储搭建

Centos之赛门铁克存储搭建

导读 赛门铁克是互联网安全技术的全球领导厂商，为企业、个人用户和服务供应商提供广泛的内容和网络安全软件及硬件的...
一言一句话
照我思索，可理解我，照我思索，可以得人。
-「沈从文墓志铭」
【腾讯云】云服务器、云数据库、COS、CDN、短信等云产品特惠热卖中