Redis Transaction Principle and Source Code Analysis

[guozhihao-224]

基本介绍

Redis对于事务处理的逻辑相对简单，归纳起来就是每个client记录监控了哪些key(watched_keys)和使用list来存储当前事务中需要执行的命令；对于redis db来说，同样是使用HashTable的方式记录某个key被哪些client监控了。涉及以下操作：

1. 使用watch命令监控指定的key，在client的watched_keys和db的HashTable中记录这个key（使用watched_key结构)
2. 使用unwatch命令取消当前client所有监控的key，既将client的watched_keys清空，同时将HashTable所有关于该client的key清除
3. multi命令开启事务，exec命令执行事务。

命令使用方式

• MULTI
  • 开启事务，但是MULTI命令不能嵌套使用
  • client在非事务状态下，执行MULTI命令会返回OK
• EXEC
  • 按顺序执行命令
  • 某个命令执行失败，不会影响队列里面其他命令的执行
  • 返回每一条命令的执行结果
• DISCARD
  • 放弃事务
• WATCH key [key....]
  • 监控指定的key
  • 如果在执行EXEC命令前，监控的key发生变化，此时发送EXEC命令会返回nil的错误，并取消事务
• UNWATCH
  • 取消所有key的监视（当前client）

命令调用链

• watchCommand(client* c)
  • -> watchForKey(client* c, robj* key) -> 保存在db->watched_keys, c->watched_keys中

Redis事务相关流程调用链（只关注与事务相关的逻辑）（基于Redis 7.0.15 版本）

命令处理（对应图中蓝色字部分）

详细代码解析（基于Redis 7.0.15 版本）

主要数据结构

 /*
  * 用于表示被watch的key
  * db->watched_keys 中， 相当于一个HashTable，key -> []watchedKey  , 可以直接看作为 key -> []client , 表示某个key被哪些client监控
  * client->watched_keys, []watchedKey
 */
 typedef struct watchedKey {
    robj *key;
    redisDb *db;
    client *client;
    // 是否过期，在7.0版本以后才有这个功能
    /* Flag that we're watching an already expired key. */
    unsigned expired:1; 
} watchedKey;

WATCH命令实现

void watchCommand(client *c) {
    ...
    // 不允许在multi状态中使用watch命令
    if (c->flags & CLIENT_MULTI) {
        addReplyError(c,"WATCH inside MULTI is not allowed");
        return;
    }
    /* No point in watching if the client is already dirty. */
    // client处于 client_dirty_cas 状态，说明之前的watch的命令已经发生改变了，再进行watch已经没有意义
    // 直接返回Ok
    if (c->flags & CLIENT_DIRTY_CAS) {
        addReply(c,shared.ok);
        return;
    }
    for (j = 1; j < c->argc; j++)
        watchForKey(c,c->argv[j]);
    addReply(c,shared.ok);
}

void watchForKey(client *c, robj *key) {
    /* Check if we are already watching for this key */
    // 检查该client是否已经watch了该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 */
    // 找到该db中，key为key的watched_key list
    clients = dictFetchValue(c->db->watched_keys,key);
    if (!clients) {
        clients = listCreate();
        dictAdd(c->db->watched_keys,key,clients);
        incrRefCount(key);
    }
    /* Add the new key to the list of keys watched by this client */
    wk = zmalloc(sizeof(*wk));
    wk->key = key;
    wk->client = c;
    wk->db = c->db;
    // 设置该key是否过期
    wk->expired = keyIsExpired(c->db, key);
    incrRefCount(key);
    // 将该key加入到该client的watched_key list中
    listAddNodeTail(c->watched_keys,wk);
    // 将该key插入到该db的该key的的list中
    listAddNodeTail(clients,wk);
}