redis源碼分析之事務Transaction（下）

接着上一篇，這篇文章分析一下redis事務操做中multi，exec，discard三個核心命令。

原文地址：www.jianshu.com/p/e22615586…

看本篇文章前須要先對上面文章有所瞭解：
redis源碼分析之事務Transaction（上）

1、redis事務核心命令簡介

redis事務操做核心命令：

//用於開啓事務
    {"multi",multiCommand,1,"sF",0,NULL,0,0,0,0,0},
    //用來執行事務中的命令
    {"exec",execCommand,1,"sM",0,NULL,0,0,0,0,0},
    //用來取消事務
    {"discard",discardCommand,1,"sF",0,NULL,0,0,0,0,0},複製代碼

在redis中，事務並不具備ACID的概念，換句話說，redis中的事務僅僅是保證一系列的命令按順序一個一個執行，若是中間失敗了，並不會進行回滾操做。

使用redis事務舉例以下：

127.0.0.1:6379> multi
OK
127.0.0.1:6379> set a a
QUEUED
127.0.0.1:6379> set b b
QUEUED
127.0.0.1:6379> set c c
QUEUED
127.0.0.1:6379> exec
1) OK
2) OK
3) OK
127.0.0.1:6379>複製代碼

2、redis事務核心命令源碼分析

關於事務的幾個命令所對應的函數都放在multi.c文件中。

首先來看一下multi命令，該命令用於標記客戶端開啓事務狀態，所以它作的就是修改客戶端狀態，代碼很簡單，以下：

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);
}複製代碼

接下來看下redis處理命令邏輯中的一段源碼：
這段代碼在server.c文件中的processCommand方法中：

//若是客戶端處於事務狀態且當前執行的命令不是exec，discard，multi跟watch命令中的一個
    //則把當前命令加入一個隊列
    if (c->flags & CLIENT_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,CMD_CALL_FULL);
        c->woff = server.master_repl_offset;
        if (listLength(server.ready_keys))
            handleClientsBlockedOnLists();
    }複製代碼

看入隊操做源碼前，先來熟悉幾個數據結構，redis會把每一個鏈接的客戶端封裝成一個client對象，該對象中含有大量字段用來保存須要的信息，發佈訂閱功能也使用對應的字段進行存儲，事務固然也不例外，以下：

//每一個客戶端對象中有一個mstate字段用來保存事務上下文
typedef struct client {
    multiState mstate; 
}

//事務包裝類型
typedef struct multiState {
    //當前事務中須要執行的命令數組
    multiCmd *commands; 
    //須要執行的命令數量
    int count;          
    //須要同步複製的最小數量
    int minreplicas;       
    //同步複製超時時間
    time_t minreplicas_timeout; 
} multiState;

//事務中執行命令的封裝類型
typedef struct multiCmd {
    //參數
    robj **argv;
    //參數數量
    int argc;
    //命令自己
    struct redisCommand *cmd;
} multiCmd;複製代碼

瞭解了基本的數據結構之後，再來看下入隊操做：

void queueMultiCommand(client *c) {
    //類型前面有說明
    multiCmd *mc;
    int j;
    //擴容，每次擴容一個命令的大小
    c->mstate.commands = zrealloc(c->mstate.commands,
            sizeof(multiCmd)*(c->mstate.count+1));
    //c++中給數組最後一個元素賦值語法實在是有點難懂...
    mc = c->mstate.commands+c->mstate.count;
    //初始化mc各個字段
    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++;
}複製代碼

上面是把命令加入事務命令數組的中的邏輯，因爲在執行事務過程當中也會執行刪除事務的操做，所以在看執行事務邏輯以前咱們先看下刪除事務的實現原理。

當事務執行完成，執行錯誤或者客戶端想取消當前事務，都會跟discard命令有聯繫，一塊兒看下源碼：

void discardCommand(client *c) {
    //若是當前客戶端沒有處於事務狀態，則返回錯誤信息
    if (!(c->flags & CLIENT_MULTI)) {
        addReplyError(c,"DISCARD without MULTI");
        return;
    }
    //刪除事務
    discardTransaction(c);
    //返回結果
    addReply(c,shared.ok);
}

//具體的刪除邏輯
void discardTransaction(client *c) {
    //釋放客戶端事務資源
    freeClientMultiState(c);
    //初始化客戶端事務資源
    initClientMultiState(c);
    //狀態位還原
    c->flags &= ~(CLIENT_MULTI|CLIENT_DIRTY_CAS|CLIENT_DIRTY_EXEC);
    //取消已watch的key，該函數上面文章中已經進行過度析，不贅述
    unwatchAllKeys(c);
}

//釋放事務隊列中的每一個命令
void freeClientMultiState(client *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);
}

//事務相關字段設爲初始值
void initClientMultiState(client *c) {
    c->mstate.commands = NULL;
    c->mstate.count = 0;
}複製代碼

到這裏，咱們已經瞭解了開啓事務模式，把各個命令加入到事務命令執行數組中以及取消事務三個模塊的執行原理，最後一塊兒看下事務的執行過程，代碼較長，須要慢慢看。

把一系列命令加入到事務命令數組中之後，客戶端執行exec命令就能夠把其中的全部命令挨個執行完成了，分析exec命令源碼以前，咱們應該能夠想到redis的邏輯應該就是從客戶端的事務命令數組中取出全部命令一個一個執行，源碼以下：

void execCommand(client *c) {
    int j;
    robj **orig_argv;
    int orig_argc;
    struct redisCommand *orig_cmd;
    //標記是否須要把MULTI/EXEC傳遞到AOF或者slaves節點
    int must_propagate = 0; 
    //標記當前redis節點是否爲主節點
    int was_master = server.masterhost == NULL;
    //若是客戶端沒有處於事務狀態，則返回錯誤提示信息
    if (!(c->flags & CLIENT_MULTI)) {
        addReplyError(c,"EXEC without MULTI");
        return;
    }

    //首先對兩個須要終止當前事務的條件進行判斷
    //1.當有WATCH的key被修改時則終止，返回一個nullmultibulk對象
    //2.當以前有命令加入事務命令數組出錯則終止，例如傳入的命令參數數量不對，會返回execaborterr
    if (c->flags & (CLIENT_DIRTY_CAS|CLIENT_DIRTY_EXEC)) {
        addReply(c, c->flags & CLIENT_DIRTY_EXEC ? shared.execaborterr :
                                                  shared.nullmultibulk);
        //刪除當前事務信息，前面已經分析過，不贅述
        discardTransaction(c);
        goto handle_monitor;
    }

    //把watch的key都刪除，上面文章已經分析過，不贅述
    unwatchAllKeys(c); 
    //保存當前命令上下文
    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++) {
        //把事務隊列中的命令參數取出賦值給client，由於命令是在client維度執行的
        c->argc = c->mstate.commands[j].argc;
        c->argv = c->mstate.commands[j].argv;
        c->cmd = c->mstate.commands[j].cmd;

        //同步事務操做到AOF或者集羣中的從節點
        if (!must_propagate && !(c->cmd->flags & (CMD_READONLY|CMD_ADMIN))) {
            execCommandPropagateMulti(c);
            must_propagate = 1;
        }
        //執行具體命令
        call(c,CMD_CALL_FULL);

        //因爲命令能夠修改參數的值或者數量，所以從新保存命令上下文
        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);

    //確保EXEC會進行傳遞
    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));
        }
    }
//monitor命令操做
handle_monitor:
    if (listLength(server.monitors) && !server.loading)
        replicationFeedMonitors(c,server.monitors,c->db->id,c->argv,c->argc);
}複製代碼

上面就是事務命令執行的整個邏輯，能夠先排除集羣跟AOF的同步邏輯，專一理解核心邏輯，代碼總體邏輯算是比較清晰的，搞明白了前面的幾個模塊之後，再看執行邏輯就不會太難。

3、redis事務命令總結

經過上、下兩篇文章對redis事務各個命令進行了分析，仔細閱讀應該能夠了解整個事務執行框架，若是有任何問題或者疑惑，歡迎留言評論。