Redis | 事务源码阅读
Redis | 事务源码阅读
Redis 的事务在前面的内容中已经介绍过了，分别对应的两篇文章如下：
因为 Redis 的特性是要快，因此 Redis 事务没有关系型数据库的事务那么强大，这是它们在设计上的权衡。Redis 的事务只有简单的几条命令，且较为简单。它的实现也相对简单许多。我们来阅读一下 Redis 关于事务的源码。
事务的命令
事务的命令只有为数不多的几个，比较常用的是 multi、discard 和 exec 三个命令。这三个命令的作用分别是 开启事务、取消事务 和 执行事务。在事务中还有两个命令用来提供乐观锁，分别是 watch 和 unwatch 两个命令。这些命令在前面的文章中已经进行过整理。这篇文章重点来整理事务相关的源码。
multi 命令
multi 命令的源码在 multi.c 文件中，且源码特别的短，源码如下。
/**
* multi命令对应的源码
*/void multiCommand(redisClient *c) { // 判断是否嵌套执行multi
if(c->flags & REDIS_MULTI) {
addReplyError(c,"MULTI calls can not be nested");
return;
}
// 开启multi标志位
c->flags |= REDIS_MULTI;
addReply(c,shared.ok);}multiCommand 函数中只完成了两个功能，第一个功能是判断 multi 命令是否嵌套执行，也就是说开启了事务之后，如果没有执行（exec）或取消(discard)事务，是不能再次开启(multi)事务的，如下图所示。
可以看到，在连续两次输入 multi 命令后，提示 "MULTI calls can not be nested"，这个错误提示和源码中的提示是相同的。
multiCommand 函数中在检查后发现事务没有嵌套执行时，则会开启事务，开启事务只是对 redisClient 结构体的 flags 属性设置了 REDIS_MULTI 标志位。位运算的 | 通常对标志位进行置位，位运算的 & 通常对标志位进行复位 或 标志位位检测。
discard 命令
discard 命令是用来取消事务的，当使用 multi 开启一个事务后，可以选择使用 exec 命令来以原子的方式执行事务命令队列中的所有命令，也可以使用 discard 命令来取消整个当前的事务。
discard 命令的源码如下。
/**
* discard命令对应的源码
*/void discardCommand(redisClient *c) { // 未开启事务
if(!(c->flags & REDIS_MULTI)) {
addReplyError(c,"DISCARD without MULTI");
return;
}
discardTransaction(c);
addReply(c,shared.ok);}discardCommand 函数中也基本上完成了两个功能，第一个功能是判断是否开启了事务，第二个功能是取消事务。判断事务是否开启，只要检测 redisClient 的属性 flags 即可。取消事务的源码如下。
/**
* 取消事务
*/void discardTransaction(redisClient *c) { // 释放事务队列中的所有命令
freeClientMultiState(c);
// 初始化multiState的
initClientMultiState(c);
c->flags &=~(REDIS_MULTI|REDIS_DIRTY_CAS|REDIS_DIRTY_EXEC);;
// 释放watch中所有的key
unwatchAllKeys(c);}在取消事务时一共要完成四件事情，首先要释放掉所有事务队列中的命令。当开启事务后，所有的命令不会马上执行，而是要进入暂存命令的队列，所以在取消事务时，需要将队列中的命令都释放掉。释放事务队列中的命令的代码如下。
/* Release all the resources associated with MULTI/EXEC state */void freeClientMultiState(redisClient *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);}在释放了事务队列中的命令后，接着将 multiState 进行初始化。初始化的代码如下。
/* Client state initialization for MULTI/EXEC */void initClientMultiState(redisClient *c) { c->mstate.commands = NULL;
c->mstate.count = 0;}初始化的过程很简单，mstate 的命令队列置空，队列长度置 0 即可。
初始化后，将 redisClient 的 flags 标志位中关于 事务 和 watch 对应的几个位进行复位。
对于 unwatchAllKeys 函数则是 unwatch 掉所有 watch 的 key，该函数暂时先不看。
redisClient 结构体
在上面的 multiCommand、discardCommand 等几个函数中，它们的参数都是一个 redisClient 结构体指针。它们基本都访问（读写）了两个属性，分别是 flags 和 mstate 。redisClient 结构体想必是一个重要的结构体，它是 Redis 客户端的一个结构体，我们来看一下它的定义，该结构体的定义在 redis.h 文件中。
typedefstruct redisClient {
……
// 标志位
int flags; /* REDIS_SLAVE | REDIS_MONITOR | REDIS_MULTI ... */
……
// 事务状态
multiState mstate; /* MULTI/EXEC state */
……
list *watched_keys; /* Keys WATCHED for MULTI/EXEC CAS */
……
} redisClient;redisClient 结构体的属性特别的多，这里只留下三个与事务相关的属性。分别是 flags、mstate 和 watched_keys 三个属性。第一个是 标志位，第二个是 事务状态，第三个是 watch 链。前两个已经见过了，最后这个暂时先不管。
mstate 是一个 multiState 的结构体，看 multiState 结构体的定义。
/**
* 事务状态
*/typedef struct multiState {
// 命令队列
multiCmd *commands; /* Array of MULTI commands */
// 入队命令数量
int count; /* Total number of MULTI commands */} multiState;该结构体的属性有两个，commands 是命令队列，count 表示命令队列的长度。无论是调用命令，还是清空队列，都是根据 count (长度)来逐个的访问 commands。commands 是 multiCmd 的结构体，该结构体的定义如下。
/* Client MULTI/EXEC state */ttypedef struct multiCmd { // 参数
robj **argv;
// 参数个数
int argc;
// 命令指针
struct redisCommand *cmd;} multiCmd;multiCmd 中保存了命令的参数（argv）、参数的个数（argc），以及命令指针（cmd）。对于 multiState 来说，commands 就是一个 multiCmd 的队列，也就是说，commands 是由多个 multiCmd 组成。
redisClient、multiState 和 multiCmd 三个结构体的大体关系如下图。
exec 命令
exec 命令是用来执行事务队列中命令的。exec 命令对应的源码比较长，我保留了比较重要的代码，其余代码进行了删除，源码如下。
/**
* exec命令对应的源码
*/
void execCommand(redisClient *c) {
……
// 判断是否开启事务
if(!(c->flags & REDIS_MULTI)) {
addReplyError(c,"EXEC without MULTI");
return;
}
……
// 判断是否有对应的watch，且watch的keys是否被修改
if(c->flags &(REDIS_DIRTY_CAS|REDIS_DIRTY_EXEC)) {
addReply(c, c->flags & REDIS_DIRTY_EXEC ? shared.execaborterr :
shared.nullmultibulk);
// 取消事务
discardTransaction(c);
goto handle_monitor;
}
// 执行事务前，将所有的wach的keys都unwatch掉
unwatchAllKeys(c); /* Unwatch ASAP otherwise we'll waste CPU cycles */
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++) {
……
// 执行事务
call(c,REDIS_CALL_FULL);
……
}
c->argv = orig_argv;
c->argc = orig_argc;
c->cmd = orig_cmd;
……
}执行事务时，会判断是否开启了事务，判断是否有 watch 的 keys，且 watch 的 keys 是否有被修改过，如果都没有问题，就将 watch 的 keys 全部 unwatch 掉，最后就开始执行队列中的命令了。
命令入队
最后来看一下开启事务后命令是如何进入队列的。该部分源码在 redis.c 文件中的 processCommand 函数中，该代码比较长，只保留开启事务后入队的代码，源码如下。
int processCommand(redisClient *c) {
……
/* Exec the command */
if(c->flags & REDIS_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,REDIS_CALL_FULL);
if(listLength(server.ready_keys))
handleClientsBlockedOnLists();
}
return REDIS_OK;}当 redis 开启事务后，输入的命令对应的处理方法不是 execCommand、discardCommand 等，就通过 queueMultiCommand 函数来将命令入队，它的源码在 multi.c 下，代码如下。
void queueMultiCommand(redisClient *c) {
multiCmd *mc;
int j;
c->mstate.commands = zrealloc(c->mstate.commands,
sizeof(multiCmd)*(c->mstate.count+1));
mc = c->mstate.commands+c->mstate.count;
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++;}在代码中，使用 zrealloc 调整了命令队列的空间，也就是为新的命令分配了内存空间大小，以便有足够的空间来保存新入队的命令。然后让 mc 指向了当前队列的尾部（也就是新命令要入队的内存的起始位置），从而让新的命令入队。最后增加命令队列的长度。
总结
关于事务的数据结构和代码，就整理这么多，基本上已经算是把事务的大体结构搞明白了，主要就是开始事务、命令入队、执行/取消事务 的源代码，整个代码的流程不算复杂。最后，还有关于 watch 和 unwatch 的部分没有进行阅读，留着下次再来进行整理。
腾讯云开发者
