Redis 事务实现分析

通过对 Redis 源码中的 multi.c 文件进行分析，解释 Redis 事务(transaction)功能的实现原理。

1、Redis 的事务用法

Redis 的事务使用 MULTI 命令 和 EXEC 命令 包围，处在这两条命令之间的一条或多条命令，会以 FIFO 的方式运行：

redis> MULTI # 标记事务开始
OK
redis> INCR user_id
QUEUED
redis> SET greeting "hello moto"
QUEUED
redis> GET replay
QUEUED
redis> EXEC # 标记事务结束，并执行事务
1) (integer) 1
2) OK
3) "hello world"

需要注意的是，Redis 的事务和关系数据库的事务并不一样：Redis 的事务并不保证 ACID 性质。

也就是说，在 Redis 事务的执行过程中，因为服务器失败而造成数据不一致的情况是可能存在的，在后面对代码进行分析的时候，就会清晰地看到这一点。

2、MULTI 命令

每个 Redis 事务都以 MULTI 命令开始，而 MULTI 命令本身的实现则非常简单：

void multiCommand(redisClient *c) {
// MULTI 不可以嵌套使用
if (c->flags & REDIS_MULTI) {
addReplyError(c,"MULTI calls can not be nested");
return;
}
c->flags |= REDIS_MULTI; // 打开 FLAG
addReply(c,shared.ok);
}

multiCommand 的主要动作就是对 redisClient 结构的 flags 进行检查和设置。

它首先检查 flags ，确保没有嵌套使用 MULTI 命令。

如果检查通过，那么就使用位或操作，将 REDIS_MULTI 这个 FLAG 打开。

最后向客户端返回 OK 。

3、命令的入队

当 REDIS_MULTI 这个 FLAG 被打开之后， 传入 Redis 客户端的命令就不会马上被执行(部分命令如EXEC 除外)， 这些未被执行的命令会被 queueMultiCommand 以 FIFO 的方式放入一个数组里，储存起来。

redis.c 文件里的 processCommand 函数说明了这一点：

int processCommand(redisClient *c) {
// 其他代码 ...
/* Exec the command */
// 如果 REDIS_MULTI 被打开
// 且要执行的命令不是 EXEC 、 DISCARD 、 MULTI 或 WATCH
// 那么将这个命令入队
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);
}
return REDIS_OK;
}

queueMultiCommand 函数将要执行的命令、命令的参数个数以及命令的参数放进 multiCmd 结构中，并将这个结构保存到 redisClient.mstate.command 数组的末尾，从而形成一个保存了要执行的命令的 FIFO 队列：

/* Add a new command into the MULTI commands queue */
// 添加新命令到 MULTI 的执行队列中(FIFO)
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++;
}

queueMultiCommand 函数用到的几个结构（在 redis.h 文件）：

typedef struct redisClient {
// 其他属性 ...
redisDb *db; // 当前 DB
int flags; // 标记事务状态，以及 WATCH 状态
multiState mstate; // 事务中的所有命令
list *watched_keys; // 这个客户端 WATCH 的所有 KEY
// 其他属性 ...
} redisClient;

typedef struct multiState {
multiCmd *commands; // 保存事务中所有命令的数组
int count; // 命令的数量
} multiState;

typedef struct multiCmd {
robj **argv; // 命令参数
int argc; // 命令参数数量
struct redisCommand *cmd; // 命令
} multiCmd;

回到文章开头的例子，在执行以下几个命令之后：

redis> MULTI # 标记事务开始

OK

redis> INCR user_id

QUEUED

redis> SET greeting "hello moto"

QUEUED

redis> GET replay

QUEUED

redisClient.mstate 的值应该类似这个样子(用 JSON 结构来表示)：

redisClient.mstate = {

'count': 3,

'commands': [

{

'argv': ['user_id'],

'argc': 1,

'cmd': 'incrCommand',

},

{

'argv': ['greeting', 'hello moto'],

'argc': 2,

'cmd': 'setCommand',

},

{

'argv': ['replay'],

'argc': 1,

'cmd': 'getCommand',

}

]

}

4、执行事务

事务的执行由 execCommand 函数进行，它的定义如下：

void execCommand(redisClient *c) {
int j;
robj **orig_argv;
int orig_argc;
struct redisCommand *orig_cmd;
// 如果没执行过 MULTI ，报错
if (!(c->flags & REDIS_MULTI)) {
addReplyError(c,"EXEC without MULTI");
return;
}

/* Check if we need to abort the EXEC if some WATCHed key was touched.
* A failed EXEC will return a multi bulk nil object. */
// 如果在执行事务之前，有监视中(WATCHED)的 key 被改变
// 那么取消这个事务
if (c->flags & REDIS_DIRTY_CAS) {
freeClientMultiState(c);
initClientMultiState(c);
c->flags &= ~(REDIS_MULTI|REDIS_DIRTY_CAS);
unwatchAllKeys(c);
addReply(c,shared.nullmultibulk);
return;
}

/* Replicate a MULTI request now that we are sure the block is executed.
* This way we'll deliver the MULTI/..../EXEC block as a whole and
* both the AOF and the replication link will have the same consistency
* and atomicity guarantees. */
// 为保证一致性和原子性
// 如果处在 AOF 模式中，向 AOF 文件发送 MULTI
// 如果处在复制模式中，向附属节点发送 MULTI
execCommandReplicateMulti(c);
/* Exec all the queued commands */
// 开始执行所有事务中的命令(FIFO 方式)
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++) {
c->argc = c->mstate.commands[j].argc; // 取出参数数量
c->argv = c->mstate.commands[j].argv; // 取出参数
c->cmd = c->mstate.commands[j].cmd; // 取出要执行的命令
call(c,REDIS_CALL_FULL); // 执行命令
/* Commands may alter argc/argv, restore mstate. */
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;
// 重置事务状态
freeClientMultiState(c);
initClientMultiState(c);

c->flags &= ~(REDIS_MULTI|REDIS_DIRTY_CAS);
/* Make sure the EXEC command is always replicated / AOF, since we
* always send the MULTI command (we can't know beforehand if the
* next operations will contain at least a modification to the DB). */
// 更新状态值，确保事务执行之后的状态为脏
server.dirty++;
}

execCommand 的主要作用只是一个个地执行储存在 redisClient.mstate 数组中的命令， 命令在执行之前并没有使用日志之类的保护机制， 这是为什么 Redis 的事务并不支持 ACID 这些性质的(其中一个)原因。

在 execCommand 的前半部分，调用了 execCommandReplicateMulti 函数， 如果有需要的话， Redis 就会向 AOF 文件和其他复制实例(replication)发送 MULTI 命令， 通知正在执行事务。

如果事务在执行过程中失败，那么 AOF 文件和复制实例都会察觉到， 这时 Redis 实例会报错并退出，然后等待管理员使用 redis-check-aof 命令来进行数据修复， 具体请参考： Redis 官方网站上的 Transaction 介绍 。

5、取消事务

DISCARD 命令就是用来中途取消事务的， 它的实现由 discardTransaction 和 discardCommand两个函数实现：

// 放弃执行事务
void discardTransaction(redisClient *c) {
freeClientMultiState(c); // 释放事务资源
initClientMultiState(c); // 重置事务状态
c->flags &= ~(REDIS_MULTI|REDIS_DIRTY_CAS);; // 关闭 FLAG
unwatchAllKeys(c); // 取消对所有 key 的 WATCH
}
// 放弃执行事务(命令)
void discardCommand(redisClient *c) { //事务的状态存储在redisClient的标识里
// 如果没有调用过MULTI，报错
if (!(c->flags & REDIS_MULTI)) {
addReplyError(c,"DISCARD without MULTI");
return;
}
discardTransaction(c);
addReply(c,shared.ok);
}

其中 freeClientMultiState 和 initClientMultiState 两个函数用于重置redisClient.mstate 数组，从而达到删除所有入队命令的作用：

/* Client state initialization for MULTI/EXEC */
// 初始化客户端状态，为执行事务作准备
void initClientMultiState(redisClient *c) {
c->mstate.commands = NULL; // 清空命令数组
c->mstate.count = 0; // 清空命令计数器
}

/* 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);
}