redis源码分析之事务Transaction（下）

Posted 2020-08-18

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

原文地址：http://www.jianshu.com/p/e22615586595

看本篇文章前需要先对上面文章有所了解：
redis源码分析之事务Transaction（上）

一、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事务各个命令进行了分析，仔细阅读应该可以了解整个事务执行框架，如果有任何问题或者疑惑，欢迎留言评论。