Redis学习之事务

介绍

   Redis 事务可以一次执行多个命令， 并且带有以下两个重要的保证：

• 批量操作在发送 EXEC 命令前被放入队列缓存。

• 收到 EXEC 命令后进入事务执行，事务中任意命令执行失败，其余的命令依然被执行。

事务过程不同的客户端是没有事务之间的影响的，一个事务从开始到执行会经历以下三个阶段：

• 开始事务

• 命令入队

• 执行事务

一、相关命令

• Exec 执行所有事务块内的命令。

• Watch 监视一个(或多个) key ，如果在事务执行之前这个(或这些) key 被其他命令所改动，那么事务将被打断。

• Discard 取消事务，放弃执行事务块内的所有命令。

• Unwatch 取消 WATCH 命令对所有 key 的监视。

• Multi 标记一个事务块的开始。

二、数据结构

typedef struct client {
    ...
    list *watched_keys; /* 事务操作 监控的keys链表 值为watchedKey结构*/
    multiState mstate;  /* 事务状态 */
    ...
}client;
typedef struct multiState {
    multiCmd *commands;     /*事务命令队列 */
    int count;              /*命令总个数*/
    ...
} multiState;
其中、client.watched_keys里面放的Node都是数据时如下结构体
//监视key
typedef struct watchedKey {
    robj *key;//监视的key
    redisDb *db;//哪个db里面的
} watchedKey;

//redisDb结构
typedef struct redisDb {
    ...
    dict *watched_keys;         /*数据库的监控key字典用来实现事务的 key -> client链表*/
    ...
} redisDb;

三、API

void multiCommand(client *c);//事务开始
void execCommand(client *c);//事务执行
void discardCommand(client *c);//取消事务
void watchCommand(client *c);//监控客户端的key
void unwatchCommand(client *c);//取消监控客户端正在监控的key

四、重要API解析

• multiCommand

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

• watchCommand

  /* 监视特定的key */
  void watchForKey(client *c, robj *key) {
      list *clients = NULL;
      listIter li;
      listNode *ln;
      watchedKey *wk;
      listRewind(c->watched_keys,&li);
      while((ln = listNext(&li))) {
          wk = listNodeValue(ln);
          if (wk->db == c->db && equalStringObjects(key,wk->key))
              return; /* 如果这个key已经被监控了 直接返回*/
      }
      /* 否则加入监视 */
      clients = dictFetchValue(c->db->watched_keys,key);
      if (!clients) {
          clients = listCreate();
          dictAdd(c->db->watched_keys,key,clients);
          incrRefCount(key);
      }
      listAddNodeTail(clients,c);//加入到监视keys的链表
      wk = zmalloc(sizeof(*wk));
      wk->key = key;
      wk->db = c->db;
      incrRefCount(key);
      listAddNodeTail(c->watched_keys,wk);//加入到watched_keys中
  }
  
  /* 键如果被改变 事务执行会失败*/
  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;
  
      //如果key被改变了，那么监视这个key的所有client都要被设置CLIENT_DIRTY_CAS
      listRewind(clients,&li);
      while((ln = listNext(&li))) {
          client *c = listNodeValue(ln);
  
          c->flags |= CLIENT_DIRTY_CAS;
      }
  }
  

• execCommand

  void execCommand(client *c) {//执行事务
      int j;
      robj **orig_argv;
      int orig_argc;
      struct redisCommand *orig_cmd;
      int must_propagate = 0; /* 需要将 MULTI/EXEC 同步到 AOF / slaves? */
      int was_master = server.masterhost == NULL;
  
      if (!(c->flags & CLIENT_MULTI)) {//如果并没有设置MULTI标志，无操作
          addReplyError(c,"EXEC without MULTI");
          return;
      }
      //如果事务期间被监视的key被修改了 停止事务，返回nil
      //入队的命令如果有误的话，执行事务error
      if (c->flags & (CLIENT_DIRTY_CAS|CLIENT_DIRTY_EXEC)) {
          addReply(c, c->flags & CLIENT_DIRTY_EXEC ? shared.execaborterr :
                                                     shared.nullarray[c->resp]);
          discardTransaction(c);
          goto handle_monitor;
      }
      //如果对于从库有修改key的操作 取消事务
      if (!server.loading && server.masterhost && server.repl_slave_ro &&
          !(c->flags & CLIENT_MASTER) && c->mstate.cmd_flags & CMD_WRITE)
      {
          addReplyError(c,
              "Transaction contains write commands but instance "
              "is now a read-only replica. EXEC aborted.");
          discardTransaction(c);
          goto handle_monitor;
      }
  
      /* 执行所有命令*/
      unwatchAllKeys(c); /* 取消此用户所有监视key 提升cpu效率 */
      orig_argv = c->argv;
      orig_argc = c->argc;
      orig_cmd = c->cmd;
      addReplyArrayLen(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;
          //如果有修改key的操作的话 需要同步到 aof 和 从库
          if (!must_propagate && !(c->cmd->flags & (CMD_READONLY|CMD_ADMIN))) {
              execCommandPropagateMulti(c);
              must_propagate = 1;
          }
          //执行命令
          call(c,server.loading ? CMD_CALL_NONE : CMD_CALL_FULL);
          /* 以上函数可能会改变argc argv. */
          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);
      /* 需要将 MULTI/EXEC 同步到 AOF / slaves */
      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));
          }
      }
  handle_monitor:
      if (listLength(server.monitors) && !server.loading)
          replicationFeedMonitors(c,server.monitors,c->db->id,c->argv,c->argc);
  }
  

参考

• 《Redis设计与实现》
• https://redis.io/