redisソース分析のトランザクションTransaction(下)
7836 ワード
次の記事では、redisトランザクション操作におけるmulti、exec、discardの3つのコアコマンドを分析します。
この記事を読む前に、redisソース分析のトランザクションTransaction(上)について理解する必要があります.
一、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のトランザクションは、一連のコマンドが順番に1つずつ実行されることを保証するだけで、途中で失敗した場合、ロールバック操作は行われません.
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>
二、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;
}
ここまで、トランザクションモードをオンにし、各コマンドをトランザクションコマンド実行配列に追加し、トランザクションをキャンセルする3つのモジュールの実行原理を理解しました.最後に、トランザクションの実行過程を見てみましょう.コードが長いので、ゆっくり見る必要があります.
一連のコマンドをトランザクションコマンド配列に追加すると、クライアントがexecコマンドを実行すると、その中のすべてのコマンドを1つずつ実行することができます.execコマンドのソースコードを分析する前に、redisの論理はクライアントのトランザクションコマンド配列からすべてのコマンドを1つずつ実行することであり、ソースコードは以下の通りです.
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
$4\r
EXEC\r
";
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の同期ロジックを排除することができて、核心ロジックに専念して理解して、コード全体のロジックは比較的にはっきりしていて、前のいくつかのモジュールを理解してから、実行ロジックを見るのはあまり難しくありません.
三、redis事務コマンドまとめ
上、下の2つの文章を通じてredisトランザクションの各コマンドを分析し、よく読むとトランザクション実行フレームワーク全体が理解できるはずです.何か問題や疑問があれば、コメントを歓迎します.