searchusermenu
  • 发布文章
  • 消息中心
点赞
收藏
评论
分享
原创

redis设计与实现读书笔记之事务机制解析

2023-10-10 10:34:04
25
0

redis通过MULTI、EXEC、WATCH等命令实现事务功能。事务提供了一种将多个命令请求打包,然后一次性,按顺序地执行多个命令的机制,将事务中所有命令执行完毕,才会去执行其他客户端的请求。

事务以MULTI开始,以EXEC提交执行:

事务的实现

一个事务从开始到结束会经历三个阶段,事务开始,事务入队,事务执行

事务开始

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

事务入队

处于事务状态时,执行命令,该命令会入队,简单来说就是保存到一个队列中,缓存起来,就是缓存在服务器中的该client的数据结构中

struct client {
	multiState mstate;      /* MULTI/EXEC state */
};

typedef struct multiState {
	  // 事务队列,FIFO顺序,记录所有的命令
    multiCmd *commands;     /* Array of MULTI commands */
    int count;              /* Total number of MULTI commands */
    int cmd_flags;          /* The accumulated command flags OR-ed together.
                               So if at least a command has a given flag, it
                               will be set in this field. */
    int cmd_inv_flags;      /* Same as cmd_flags, OR-ing the ~flags. so that it
                               is possible to know if all the commands have a
                               certain flag. */
} multiState;

/* Client MULTI/EXEC state */
typedef struct multiCmd {
    robj **argv;  // 参数
    int argc;  // 参数个数
    struct redisCommand *cmd; // 命令指针
} multiCmd;

int processCommand(client *c) {
	...
	/* Exec the command */
    if (c->flags & CLIENT_MULTI &&  // 处于事务状态
        c->cmd->proc != execCommand && c->cmd->proc != discardCommand &&
        c->cmd->proc != multiCommand && c->cmd->proc != watchCommand &&
        c->cmd->proc != resetCommand)
    {
        queueMultiCommand(c); // 加入队列
        addReply(c,shared.queued);
    } else {
        call(c,CMD_CALL_FULL);
        c->woff = server.master_repl_offset;
        if (listLength(server.ready_keys))
            handleClientsBlockedOnKeys();
    }

    return C_OK;
}

/* Add a new command into the MULTI commands queue */
void queueMultiCommand(client *c) {
    multiCmd *mc;
    int j;

    /* No sense to waste memory if the transaction is already aborted.
     * this is useful in case client sends these in a pipeline, or doesn't
     * bother to read previous responses and didn't notice the multi was already
     * aborted. */
    if (c->flags & CLIENT_DIRTY_EXEC)
        return;

    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++;
    c->mstate.cmd_flags |= c->cmd->flags;
    c->mstate.cmd_inv_flags |= ~c->cmd->flags;
}

举例,如果执行下面命令:

入队后的数据结构示意图如下:

事务执行

当执行EXEC命令时,server会遍历命令队列,依次执行,并将结果全部返回

void execCommand(client *c) {
    int j;
    robj **orig_argv;
    int orig_argc;
    struct redisCommand *orig_cmd;
    int was_master = server.masterhost == NULL;

    if (!(c->flags & CLIENT_MULTI)) {
        addReplyError(c,"EXEC without MULTI");
        return;
    }

    // 如果有到期失效的key,则不会执行事务
    /* EXEC with expired watched key is disallowed*/
    if (isWatchedKeyExpired(c)) {
        c->flags |= (CLIENT_DIRTY_CAS);
    }

    /* 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. */
    // 判断两个标识决定是否能执行事务
		// CLIENT_DIRTY_EXEC标识是当watched key被修改时会被设置
    if (c->flags & (CLIENT_DIRTY_CAS|CLIENT_DIRTY_EXEC)) {
        addReply(c, c->flags & CLIENT_DIRTY_EXEC ? shared.execaborterr :
                                                   shared.nullarray[c->resp]);
        discardTransaction(c);
        return;
    }

    uint64_t old_flags = c->flags;

    /* we do not want to allow blocking commands inside multi */
    c->flags |= CLIENT_DENY_BLOCKING;

	  // 事务执行完毕时,设置不WATCH所有key,watch key -> client列表的哈希表,
    // 如果client列表中有该client则删除该client,表示不监听该key
    // 每个事务执行完毕后,该client不会watch任何key,恢复到初始状态
    /* Exec all the queued commands */
    unwatchAllKeys(c); /* Unwatch ASAP otherwise we'll waste CPU cycles */

    server.in_exec = 1;

    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;

        /* ACL permissions are also checked at the time of execution in case
         * they were changed after the commands were queued. */
        int acl_errpos;
        int acl_retval = ACLCheckAllPerm(c,&acl_errpos);
        if (acl_retval != ACL_OK) {
            char *reason;
            switch (acl_retval) {
            case ACL_DENIED_CMD:
                reason = "no permission to execute the command or subcommand";
                break;
            case ACL_DENIED_KEY:
                reason = "no permission to touch the specified keys";
                break;
            case ACL_DENIED_CHANNEL:
                reason = "no permission to access one of the channels used "
                         "as arguments";
                break;
            default:
                reason = "no permission";
                break;
            }
            addACLLogEntry(c,acl_retval,acl_errpos,NULL);
            addReplyErrorFormat(c,
                "-NOPERM ACLs rules changed between the moment the "
                "transaction was accumulated and the EXEC call. "
                "This command is no longer allowed for the "
                "following reason: %s", reason);
        } else {
            call(c,server.loading ? CMD_CALL_NONE : CMD_CALL_FULL);
            serverAssert((c->flags & CLIENT_BLOCKED) == 0);
        }

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

    // restore old DENY_BLOCKING value
    if (!(old_flags & CLIENT_DENY_BLOCKING))
        c->flags &= ~CLIENT_DENY_BLOCKING;

    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 (server.propagate_in_transaction) {
        int is_master = server.masterhost == NULL;
        server.dirty++;
        /* If inside the MULTI/EXEC block this instance was suddenly
         * switched from master to slave (using the SLAVEOF command), the
         * initial MULTI was propagated into the replication backlog, but the
         * rest was not. We need to make sure to at least terminate the
         * backlog with the final EXEC. */
        if (server.repl_backlog && was_master && !is_master) {
            char *execcmd = "*1\r\n$4\r\nEXEC\r\n";
            feedReplicationBacklog(execcmd,strlen(execcmd));
        }
        afterPropagateExec();
    }

    server.in_exec = 0;
}

WATCH命令的实现

通过watch命令监控key,如果发现key被修改了,则拒绝执行事务

每个数据库都保存着dict *watched_keys字典,该字典的key是数据库的健,value是一个链表,保存着监控该key的客户端

所有对数据库的修改操作,如set,sadd等命令,最后会调用到touchWatchedKey检查是否有客户端在对该key进行监听,如果有,则将客户端的CLIENT_DIRTY_CAS标识打开,标识事务安全性已被破坏,在execCommand函数中,会检查该标识,标识打开了则执行事务不会执行。

/* "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 CLIENT_DIRTY_CAS */
    /* Check if we are already watching for this key */
    listRewind(clients,&li);
    while((ln = listNext(&li))) {
        client *c = listNodeValue(ln);

        c->flags |= CLIENT_DIRTY_CAS;
    }
}

事务的ACID性质

事务总是具有原子性一致性隔离性,并且当Redis 运行在某种特定的持久化模式下时,事务也具有耐久性

原子性

要么就执行事务中的所有操作,要么就一个操作也不执行。

Redis的事务和传统的关系型数据库事务的最大区别在于,Redis不支持事务回滚机制,即使事务队列中的某个命令在执行期间出现了错误,整个事务也会继续执行下去,直到将事务队列中的所有命令都执行完毕为止。

一致性

  • 入队错误:如果一个事务在入队命令的过程中,出现了命令不存在,或者命令的格式不正确等情况,那么Redis将拒绝执行这个事务。
  • 执行错误:即使在事务的执行过程中发生了错误,服务器也不会中断事务的执行,它会继续执行事务中余下的其他命令,并且已执行的命令(包括执行命令所产生的结果)不会被出错的命令影响。
  • 服务器停机:如果服务器运行在无持久化的内存模式下,那么重启之后的数据库将是空白的;如果服务器运行在RDB 模式下,服务器可以根据现有的RDB 文件来恢复数据;如果服务器运行在AOF模式下,根据现有的AOF文件来恢复数据。

隔离性

因为Redis使用单线程的方式来执行事务,所以Redis的事务总是以串行的方式运行的,并且事务也总是具有隔离性的。

耐久性

事务的耐久性指,当一个事务执行完毕时,执行这个事务所得的结果已经被保存到永久性存储介质里面了,即使服务器在事务执行完毕之后停机,执行事务所得的结果也不会丢失。

重点回顾

 

0条评论
0 / 1000
9****m
15文章数
1粉丝数
9****m
15 文章 | 1 粉丝
原创

redis设计与实现读书笔记之事务机制解析

2023-10-10 10:34:04
25
0

redis通过MULTI、EXEC、WATCH等命令实现事务功能。事务提供了一种将多个命令请求打包,然后一次性,按顺序地执行多个命令的机制,将事务中所有命令执行完毕,才会去执行其他客户端的请求。

事务以MULTI开始,以EXEC提交执行:

事务的实现

一个事务从开始到结束会经历三个阶段,事务开始,事务入队,事务执行

事务开始

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

事务入队

处于事务状态时,执行命令,该命令会入队,简单来说就是保存到一个队列中,缓存起来,就是缓存在服务器中的该client的数据结构中

struct client {
	multiState mstate;      /* MULTI/EXEC state */
};

typedef struct multiState {
	  // 事务队列,FIFO顺序,记录所有的命令
    multiCmd *commands;     /* Array of MULTI commands */
    int count;              /* Total number of MULTI commands */
    int cmd_flags;          /* The accumulated command flags OR-ed together.
                               So if at least a command has a given flag, it
                               will be set in this field. */
    int cmd_inv_flags;      /* Same as cmd_flags, OR-ing the ~flags. so that it
                               is possible to know if all the commands have a
                               certain flag. */
} multiState;

/* Client MULTI/EXEC state */
typedef struct multiCmd {
    robj **argv;  // 参数
    int argc;  // 参数个数
    struct redisCommand *cmd; // 命令指针
} multiCmd;

int processCommand(client *c) {
	...
	/* Exec the command */
    if (c->flags & CLIENT_MULTI &&  // 处于事务状态
        c->cmd->proc != execCommand && c->cmd->proc != discardCommand &&
        c->cmd->proc != multiCommand && c->cmd->proc != watchCommand &&
        c->cmd->proc != resetCommand)
    {
        queueMultiCommand(c); // 加入队列
        addReply(c,shared.queued);
    } else {
        call(c,CMD_CALL_FULL);
        c->woff = server.master_repl_offset;
        if (listLength(server.ready_keys))
            handleClientsBlockedOnKeys();
    }

    return C_OK;
}

/* Add a new command into the MULTI commands queue */
void queueMultiCommand(client *c) {
    multiCmd *mc;
    int j;

    /* No sense to waste memory if the transaction is already aborted.
     * this is useful in case client sends these in a pipeline, or doesn't
     * bother to read previous responses and didn't notice the multi was already
     * aborted. */
    if (c->flags & CLIENT_DIRTY_EXEC)
        return;

    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++;
    c->mstate.cmd_flags |= c->cmd->flags;
    c->mstate.cmd_inv_flags |= ~c->cmd->flags;
}

举例,如果执行下面命令:

入队后的数据结构示意图如下:

事务执行

当执行EXEC命令时,server会遍历命令队列,依次执行,并将结果全部返回

void execCommand(client *c) {
    int j;
    robj **orig_argv;
    int orig_argc;
    struct redisCommand *orig_cmd;
    int was_master = server.masterhost == NULL;

    if (!(c->flags & CLIENT_MULTI)) {
        addReplyError(c,"EXEC without MULTI");
        return;
    }

    // 如果有到期失效的key,则不会执行事务
    /* EXEC with expired watched key is disallowed*/
    if (isWatchedKeyExpired(c)) {
        c->flags |= (CLIENT_DIRTY_CAS);
    }

    /* 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. */
    // 判断两个标识决定是否能执行事务
		// CLIENT_DIRTY_EXEC标识是当watched key被修改时会被设置
    if (c->flags & (CLIENT_DIRTY_CAS|CLIENT_DIRTY_EXEC)) {
        addReply(c, c->flags & CLIENT_DIRTY_EXEC ? shared.execaborterr :
                                                   shared.nullarray[c->resp]);
        discardTransaction(c);
        return;
    }

    uint64_t old_flags = c->flags;

    /* we do not want to allow blocking commands inside multi */
    c->flags |= CLIENT_DENY_BLOCKING;

	  // 事务执行完毕时,设置不WATCH所有key,watch key -> client列表的哈希表,
    // 如果client列表中有该client则删除该client,表示不监听该key
    // 每个事务执行完毕后,该client不会watch任何key,恢复到初始状态
    /* Exec all the queued commands */
    unwatchAllKeys(c); /* Unwatch ASAP otherwise we'll waste CPU cycles */

    server.in_exec = 1;

    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;

        /* ACL permissions are also checked at the time of execution in case
         * they were changed after the commands were queued. */
        int acl_errpos;
        int acl_retval = ACLCheckAllPerm(c,&acl_errpos);
        if (acl_retval != ACL_OK) {
            char *reason;
            switch (acl_retval) {
            case ACL_DENIED_CMD:
                reason = "no permission to execute the command or subcommand";
                break;
            case ACL_DENIED_KEY:
                reason = "no permission to touch the specified keys";
                break;
            case ACL_DENIED_CHANNEL:
                reason = "no permission to access one of the channels used "
                         "as arguments";
                break;
            default:
                reason = "no permission";
                break;
            }
            addACLLogEntry(c,acl_retval,acl_errpos,NULL);
            addReplyErrorFormat(c,
                "-NOPERM ACLs rules changed between the moment the "
                "transaction was accumulated and the EXEC call. "
                "This command is no longer allowed for the "
                "following reason: %s", reason);
        } else {
            call(c,server.loading ? CMD_CALL_NONE : CMD_CALL_FULL);
            serverAssert((c->flags & CLIENT_BLOCKED) == 0);
        }

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

    // restore old DENY_BLOCKING value
    if (!(old_flags & CLIENT_DENY_BLOCKING))
        c->flags &= ~CLIENT_DENY_BLOCKING;

    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 (server.propagate_in_transaction) {
        int is_master = server.masterhost == NULL;
        server.dirty++;
        /* If inside the MULTI/EXEC block this instance was suddenly
         * switched from master to slave (using the SLAVEOF command), the
         * initial MULTI was propagated into the replication backlog, but the
         * rest was not. We need to make sure to at least terminate the
         * backlog with the final EXEC. */
        if (server.repl_backlog && was_master && !is_master) {
            char *execcmd = "*1\r\n$4\r\nEXEC\r\n";
            feedReplicationBacklog(execcmd,strlen(execcmd));
        }
        afterPropagateExec();
    }

    server.in_exec = 0;
}

WATCH命令的实现

通过watch命令监控key,如果发现key被修改了,则拒绝执行事务

每个数据库都保存着dict *watched_keys字典,该字典的key是数据库的健,value是一个链表,保存着监控该key的客户端

所有对数据库的修改操作,如set,sadd等命令,最后会调用到touchWatchedKey检查是否有客户端在对该key进行监听,如果有,则将客户端的CLIENT_DIRTY_CAS标识打开,标识事务安全性已被破坏,在execCommand函数中,会检查该标识,标识打开了则执行事务不会执行。

/* "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 CLIENT_DIRTY_CAS */
    /* Check if we are already watching for this key */
    listRewind(clients,&li);
    while((ln = listNext(&li))) {
        client *c = listNodeValue(ln);

        c->flags |= CLIENT_DIRTY_CAS;
    }
}

事务的ACID性质

事务总是具有原子性一致性隔离性,并且当Redis 运行在某种特定的持久化模式下时,事务也具有耐久性

原子性

要么就执行事务中的所有操作,要么就一个操作也不执行。

Redis的事务和传统的关系型数据库事务的最大区别在于,Redis不支持事务回滚机制,即使事务队列中的某个命令在执行期间出现了错误,整个事务也会继续执行下去,直到将事务队列中的所有命令都执行完毕为止。

一致性

  • 入队错误:如果一个事务在入队命令的过程中,出现了命令不存在,或者命令的格式不正确等情况,那么Redis将拒绝执行这个事务。
  • 执行错误:即使在事务的执行过程中发生了错误,服务器也不会中断事务的执行,它会继续执行事务中余下的其他命令,并且已执行的命令(包括执行命令所产生的结果)不会被出错的命令影响。
  • 服务器停机:如果服务器运行在无持久化的内存模式下,那么重启之后的数据库将是空白的;如果服务器运行在RDB 模式下,服务器可以根据现有的RDB 文件来恢复数据;如果服务器运行在AOF模式下,根据现有的AOF文件来恢复数据。

隔离性

因为Redis使用单线程的方式来执行事务,所以Redis的事务总是以串行的方式运行的,并且事务也总是具有隔离性的。

耐久性

事务的耐久性指,当一个事务执行完毕时,执行这个事务所得的结果已经被保存到永久性存储介质里面了,即使服务器在事务执行完毕之后停机,执行事务所得的结果也不会丢失。

重点回顾

 

文章来自个人专栏
redis代码剖析
9 文章 | 1 订阅
0条评论
0 / 1000
请输入你的评论
1
1