REDIS 事务机制

基本事务操作：

任何数据库都必须要保证一种原子执行操作：最基本的原子执行操作肯定是需要提供：

举一个例子来说明： 当对某个Key 做一个统计： 可能不同的Client做它那部分的统计，一段时间后，服务器端需要得出那个key的具体值

Client1: GET number

number = number +N1;

SET number number+N1;

Client2: GET number

number = number +N2;

SET number number+N2;

原本来讲 ，期望的值是NUMBER=NUMBER+N1+N2; 但是可能结果有其他的可能性，需要将上面的3个操作原子化，也就是这样的操作流是一个完整体，而不让pipeline被打乱~！

REDIS事务机制

像上述情况 必须得以解决 不然redis很难做？作者提供了2个事务机制





利用multi/exec来完成 multi被认为是放在同一个序列中的，按照序列化去执行命令操作

看看源码是如何写的：

MULTI操作
void multiCommand(redisClient *c) {
if (c->flags & REDIS_MULTI) {
addReplyError(c,"MULTI calls can not be nested");
return;
}
c->flags |= REDIS_MULTI;
addReply(c,shared.ok);
}

每个redisClient 在同一时间都只能压入一个multi 首先检测client是否含有multi标记，如果没有 就将标记位置为REDIS_MULTI.

命令入队操作
redis设计中，对于某个redis客户端来讲，server端首先需要查看是否含有MULTI标记位【src/redis.c】来判断是否该讲multi命令写入到待执行队列中~！如果满足要求 就执行下面函数体：

在执行MULTI之后 都需要做 一个命令入队操作：

【src/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;

printf("mc:%p\n",mc);

mc->cmd = c->cmd;

mc->argc = c->argc;

mc->argv = zmalloc(sizeof(robj*)*c->argc);

printf("mc->argv:%d\n",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++;

}

typedef struct multiCmd {

robj **argv;

int argc;

struct redisCommand *cmd;

} multiCmd;

line 5:给mstate重新分配一个命令所需要的空间 预分配count+1的指针

line 6: 指针操作，定位到commands的count个命令指针

line11以后:申请c->argc个robj*地址，将c中的argv的内容都复制给mc的argv中 这里的robj指针暂时没有看懂~！





执行命令操作
执行主体函数就是下面的

void execCommand(redisClient *c) {

int j;

robj **orig_argv;

int orig_argc;

struct redisCommand *orig_cmd;

int must_propagate = 0; /* Need to propagate MULTI/EXEC to AOF / slaves? */

if (!(c->flags & REDIS_MULTI)) {

addReplyError(c,"EXEC without MULTI-----SFWTOMS");

return;

}

/* Check if we need to abort the EXEC because:

* 1) Some WATCHed key was touched.

* 2) There was a previous error while queueing commands.

* A failed EXEC in the first case returns a multi bulk nil object

* (technically it is not an error but a special behavior), while

* in the second an EXECABORT error is returned. */

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;

}

/* Exec all the queued commands */

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;

/* Propagate a MULTI request once we encounter the first write op.

* 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. */

if (!must_propagate && !(c->cmd->flags & REDIS_CMD_READONLY)) {

execCommandPropagateMulti(c);

must_propagate = 1;

}

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;

discardTransaction(c);

/* Make sure the EXEC command will be propagated as well if MULTI

* was already propagated. */

if (must_propagate) server.dirty++;

handle_monitor:

/* Send EXEC to clients waiting data from MONITOR. We do it here

* since the natural order of commands execution is actually:

* MUTLI, EXEC, ... commands inside transaction ...

* Instead EXEC is flagged as REDIS_CMD_SKIP_MONITOR in the command

* table, and we do it here with correct ordering. */

if (listLength(server.monitors) && !server.loading)

replicationFeedMonitors(c,server.monitors,c->db->id,c->argv,c->argc);

}

从line27开始： 逐步执行每个指令

must_propagate 初始是0 经过line 40-41的一个2重判断：

发现c的命名不是只含有读的成分，也就是含有写的成分，execCommandPropagateMulti就成为必须的了

execCommandPropagateMulti 这个function是用来给所有的SLAVE和AOF文件发出这个命令， 要异步到SLAVE中 为啥要这样做呢？

首先想下： SLAVE在所有读操作是无需理会的，不会改变数据 写操作是必须得处理的~！ AOF也是同样一个道理

而40-43只会执行一次 所以也就有一个mush_propagate就只有一个变量

执行45：调用命令

执行完成之后就进行销毁命令 然后将dirty更改变量

所以可以看出 整个执行过程不会去支持事务的回滚机制，不管命令M是否执行成功 都不会影响M+1的一个执行


REDIS乐观锁
利用上面的事务机制 你会发现上面的任务是最开始的那个例子也是无法完全避免的 REDIS提出一种乐观锁的机制： 【我个人认为很有技巧性 】

check-and-set：检查并且重新更新 什么意思呢 就是说get 和set能够同时保持本地原子性，不会被其他客户端干扰~！

利用调用watch命令 来监控redis某个数据库的某些Key 确保某个Key是不会被其他客户来修改 如果其他客户想要修改 那么这个redis就会执行一个任务失败的操作

你会怎么写呢？

首先要有一个监控watchkey的数据结构 存放被监控的watchkey 如果每次做修改操作时，不同redisClient来尝试修改这个操作时,应该都会检查是不是这个watchkey 如果是 就不让其修改。

具体看下源代码如何实现：

这是WATCH的更底层实现

入口参数：客户端信息指针c

被监控的key

line 8-13: 判断是否是不是含有这个watchkey

如果含有这个key 就不需要监控 映射同一个key2个条件： db相同 内容相同

line 15 -20 如果没有key 就添加这个key

注意line18：

watch_key是一个观察key字典，把所有的被观察的key都放在一个dic结构体中。

怎么添加key和value呢？

key: 被观察的key

value:client客户端

这样更加验证了前面用void*作为value的必要性啊~！

这里就又有一个watch_key 新的字典

加入乐观锁

/* Watch for the specified key */

void watchForKey(redisClient *c, robj *key) {

list *clients = NULL;

listIter li;

listNode *ln;

watchedKey *wk;

/* Check if we are already watching for this key */

listRewind(c->watched_keys,&li);

while((ln = listNext(&li))) {

wk = listNodeValue(ln);

if (wk->db == c->db && equalStringObjects(key,wk->key))

return; /* Key already watched */

}

/* This key is not already watched in this DB. Let's add it */

clients = dictFetchValue(c->db->watched_keys,key);

if (!clients) {

clients = listCreate();

dictAdd(c->db->watched_keys,key,clients);

incrRefCount(key);

}

listAddNodeTail(clients,c);

/* Add the new key to the list of keys watched by this client */

wk = zmalloc(sizeof(*wk));

wk->key = key;

wk->db = c->db;

incrRefCount(key);

listAddNodeTail(c->watched_keys,wk);

}

因为这里确实有点抽象，我利用图标加上文字来说明：

Step 1：

寻找本客户端是否还有相同的key





发现没有相同的key Ok

进行Step 2: 就讲new_key里进行加入到watch_keys的数据库里





加入的方式是Key：就是watchkey

而value:是一个List指针 如果含有watchkey 则把redisClient加入到改List指针末尾





Step3: NewKey在第一步中的WATCH_Key中





touch乐观锁
如果触碰到乐观锁，会怎么样呢？ 不管怎么样，至少要保证一点：不能再乐观锁解除之前执行这个key的所有写操作

/* "Touch" a key, so that if this key is being WATCHed by some client the
* next EXEC will fail. */
void touchWatchedKey(redisDb *db, robj *key) {
list *clients;
listIter li;
listNode *ln;
if (dictSize(db->watched_keys) == 0) return;
clients = dictFetchValue(db->watched_keys, key);
if (!clients) return;
/* Mark all the clients watching this key as REDIS_DIRTY_CAS */
/* Check if we are already watching for this key */
listRewind(clients,&li);
while((ln = listNext(&li))) {
redisClient *c = listNodeValue(ln);
c->flags |= REDIS_DIRTY_CAS;
}
}

在这里可以看出： 触碰了之后，主调客户端没有失败，而那么加锁监控的客户端是失败的

这里的touchWatchedKey只有看了被调用的例子才能理解真正的redis怎么处理这个触碰乐观锁的情况：





当Session1是没有办法执行这个age的~！也就是chang-and-set操作

只要被watch的key在其他客户端修改 而该客户端也已经进入了multi，那么我们在mutli之间的操作将会无法做成功~！

通过源码内部看具体看下怎么实现的~

对于Session1 :

Step1：src/multi.c里的watchForKey(session1,age) 加入watch的key中

Step2:执行multi函数：准备把接下来所有的命令都加入到QUEUE队列中

Step3:Session2 执行set(age,30) 这个时候set命令会查看这个是不是在watch_key里 到相应的watch_key字典中，如果含有 那就傻逼了，就调用Db.c里的singalTtachKey()，改写这个key的每个需要监控的客户端的REDIS_DIRTY_CAS字段为1

Step4：Session1就非常高兴的调用execCommand（session1）结果一发现现在这个REDIS_DIRTY_CAS字段就是一个1，就全部不执行 直接返回。

引入这个MULTI的原因
redis本身是一个单线程，按照常理来说，指令都是序列化的，一堆需要原子操作的命令放在服务器端执行 也是按照顺序往下执行，Client A 和Client B 只需要一个或者加Watch 某个Key 不管有没有multi命令是不是就确保了其会进行原子操作呢？ 在ClientA 和ClientB中，如果watch 了一个age,如果没有multi，那么假设Client A加了watch 执行了一个其中命令，而另外一个命令准备执行时，ClientB就修改了这个age，而那个时候ClientA即使读到REDIS_DIRTY_CAS为1 也起不到作用了，因为没办法进行事务的回滚操作~！

所以只能把操作放在队列中，要么不执行，要么一下子全部执行完~！