Redis watch機(jī)制的分析
我們可以用redis的watch和multi來(lái)處理一些涉及并發(fā)的操作，redis的watch+multi實(shí)際是一種樂觀鎖，今天我們來(lái)分析一下它的實(shí)現(xiàn)機(jī)制陕习。
常用的代碼段
$key = 'xxxx';
$redis->watch($key);
$redis->multi();
// 更新了key
$redis->set($key);
$flag = $redis->exec();
// 如果事務(wù)執(zhí)行失敗返回false
if ($flag === false) {
} else {
}
流程圖
redis_watch.png
當(dāng)客戶端A和客戶端B同時(shí)執(zhí)行這段代碼時(shí)候声滥，因?yàn)槭聞?wù)的執(zhí)行是串行的，假設(shè)A客戶端先于B執(zhí)行，那么當(dāng)A執(zhí)行完成時(shí)缚够，會(huì)將客戶端A從watch了這個(gè)key的列表中刪除隔缀，并且將列表中的所有客戶端都設(shè)置為CLIENT_DIRTY_CAS题造，之后當(dāng)B執(zhí)行的時(shí)候，事務(wù)發(fā)現(xiàn)B的狀態(tài)是CLIENT_DIRTY_CAS猾瘸，便終止事務(wù)并返回失敗界赔。
存儲(chǔ)
- redis 用了哈希表+鏈表存儲(chǔ)watch了key的客戶端：
哈希表key為redis的key，哈希表的value為客戶端組成的鏈表
定位key的復(fù)雜度O(1)牵触，查找和處理客戶端的復(fù)雜度為O(n)
Key1 => (client1->client2->client3...)
Key2 => (client1->client2->client3...)
- 每個(gè)客戶端也維護(hù)一個(gè)鏈表用來(lái)存儲(chǔ)已經(jīng)watch的key
相關(guān)源碼
涉及文件
multi.h
multi.c
db.c
t_string.c
watch
/* watch命令 */
void watchCommand(client *c) {
int j;
if (c->flags & CLIENT_MULTI) {
addReplyError(c,"WATCH inside MULTI is not allowed");
return;
}
for (j = 1; j < c->argc; j++)
watchForKey(c,c->argv[j]);
addReply(c,shared.ok);
}
typedef struct watchedKey {
robj *key;
redisDb *db;
} watchedKey;
/* watch一個(gè)key */
void watchForKey(client *c, robj *key) {
list *clients = NULL;
listIter li;
listNode *ln;
watchedKey *wk;
/* 檢查key是否已經(jīng)watch 如果已經(jīng)watch 直接返回 */
// 創(chuàng)建一個(gè)迭代器
listRewind(c->watched_keys,&li);
// 遍歷客戶端已經(jīng)watch的key
while((ln = listNext(&li))) {
wk = listNodeValue(ln);
// 當(dāng)發(fā)現(xiàn)已經(jīng)存在次key淮悼，直接返回
if (wk->db == c->db && equalStringObjects(key,wk->key))
return; /* Key already watched */
}
/* 沒有被watch，繼續(xù)一下處理 */
// 獲取hash表中當(dāng)前key的客戶端鏈表
clients = dictFetchValue(c->db->watched_keys,key);
// 如果不存在揽思，則創(chuàng)建一個(gè)鏈表用于存儲(chǔ)
if (!clients) {
clients = listCreate();
dictAdd(c->db->watched_keys,key,clients);
incrRefCount(key);
}
// 添加當(dāng)前客戶端到鏈表末尾
listAddNodeTail(clients,c);
/* 維護(hù)客戶端中的watch_keys 鏈表 */
wk = zmalloc(sizeof(*wk));
wk->key = key;
wk->db = c->db;
incrRefCount(key);
listAddNodeTail(c->watched_keys,wk);
}
multi
/* multi 命令 */
void multiCommand(client *c) {
// 如果已經(jīng)初始化了客戶端狀態(tài)袜腥，即已經(jīng)執(zhí)行了multi 則返回
if (c->flags & CLIENT_MULTI) {
addReplyError(c,"MULTI calls can not be nested");
return;
}
// 初始化客戶端狀態(tài)為CLIENT_MULTI
c->flags |= CLIENT_MULTI;
addReply(c,shared.ok);
}
/* 初始化客戶端狀態(tài) 清空事務(wù)隊(duì)列 */
void initClientMultiState(client *c) {
c->mstate.commands = NULL;
c->mstate.count = 0;
}
exec
/* exec 命令 */
void execCommand(client *c) {
int j;
robj **orig_argv;
int orig_argc;
struct redisCommand *orig_cmd;
int must_propagate = 0; /* Need to propagate MULTI/EXEC to AOF / slaves? */
int was_master = server.masterhost == NULL;
// 未執(zhí)行multi，則返回
if (!(c->flags & CLIENT_MULTI)) {
addReplyError(c,"EXEC without MULTI");
return;
}
/*
* 關(guān)鍵
* 處理客戶端狀態(tài) 以下兩種狀態(tài)會(huì)直接終止事務(wù)钉汗，不會(huì)執(zhí)行事務(wù)隊(duì)列中的命令
* 1. CLIENT_DIRTY_CAS => 當(dāng)因?yàn)閣atch的key被touch了
* 2. CLIENT_DIRTY_EXEC => 當(dāng)客戶端入隊(duì)了不存在的命令
*/
/* 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. */
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;
}
/* 執(zhí)行隊(duì)列中的命令 */
// 清空當(dāng)前客戶端中存儲(chǔ)的watch了的key羹令，和hash表中客戶端node
unwatchAllKeys(c); /* Unwatch ASAP otherwise we'll waste CPU cycles */
orig_argv = c->argv;
orig_argc = c->argc;
orig_cmd = c->cmd;
addReplyMultiBulkLen(c,c->mstate.count);
// 執(zhí)行隊(duì)列中的命令
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;
/* Propagate a MULTI request once we encounter the first command which
* is not readonly nor an administrative one.
* This way we'll deliver the MULTI/..../EXEC block as a whole and
* both the AOF and the replication link will have the same consistency
* and atomicity guarantees. */
if (!must_propagate && !(c->cmd->flags & (CMD_READONLY|CMD_ADMIN))) {
execCommandPropagateMulti(c);
must_propagate = 1;
}
// 這里會(huì)call相關(guān)的命令
// 如果是涉及到修改相關(guān)的命令，不管有沒有更改值损痰，都會(huì)將hash表中watch了key的客戶端的狀態(tài)置為CLIENT_DIRTY_CAS
call(c,CMD_CALL_FULL);
/* 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;
}
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 (must_propagate) {
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));
}
}
handle_monitor:
/* Send EXEC to clients waiting data from MONITOR. We do it here
* since the natural order of commands execution is actually:
* MUTLI, EXEC, ... commands inside transaction ...
* Instead EXEC is flagged as CMD_SKIP_MONITOR in the command
* table, and we do it here with correct ordering. */
if (listLength(server.monitors) && !server.loading)
replicationFeedMonitors(c,server.monitors,c->db->id,c->argv,c->argc);
}
/* 清空當(dāng)前事務(wù)數(shù)據(jù) */
void discardTransaction(client *c) {
freeClientMultiState(c);
initClientMultiState(c);
c->flags &= ~(CLIENT_MULTI|CLIENT_DIRTY_CAS|CLIENT_DIRTY_EXEC);
unwatchAllKeys(c);
}
/* Unwatch all the keys watched by this client. To clean the EXEC dirty
* flag is up to the caller. */
void unwatchAllKeys(client *c) {
listIter li;
listNode *ln;
if (listLength(c->watched_keys) == 0) return;
listRewind(c->watched_keys,&li);
while((ln = listNext(&li))) {
list *clients;
watchedKey *wk;
/* Lookup the watched key -> clients list and remove the client
* from the list */
wk = listNodeValue(ln);
clients = dictFetchValue(wk->db->watched_keys, wk->key);
serverAssertWithInfo(c,NULL,clients != NULL);
listDelNode(clients,listSearchKey(clients,c));
/* Kill the entry at all if this was the only client */
if (listLength(clients) == 0)
dictDelete(wk->db->watched_keys, wk->key);
/* Remove this watched key from the client->watched list */
listDelNode(c->watched_keys,ln);
decrRefCount(wk->key);
zfree(wk);
}
}
set,hset 以set命令為??
void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire, int unit, robj *ok_reply, robj *abort_reply) {
long long milliseconds = 0; /* initialized to avoid any harmness warning */
if (expire) {
if (getLongLongFromObjectOrReply(c, expire, &milliseconds, NULL) != C_OK)
return;
if (milliseconds <= 0) {
addReplyErrorFormat(c,"invalid expire time in %s",c->cmd->name);
return;
}
if (unit == UNIT_SECONDS) milliseconds *= 1000;
}
if ((flags & OBJ_SET_NX && lookupKeyWrite(c->db,key) != NULL) ||
(flags & OBJ_SET_XX && lookupKeyWrite(c->db,key) == NULL))
{
addReply(c, abort_reply ? abort_reply : shared.nullbulk);
return;
}
// 看這里?? 設(shè)置string的值
setKey(c->db,key,val);
server.dirty++;
if (expire) setExpire(c,c->db,key,mstime()+milliseconds);
notifyKeyspaceEvent(NOTIFY_STRING,"set",key,c->db->id);
if (expire) notifyKeyspaceEvent(NOTIFY_GENERIC,
"expire",key,c->db->id);
addReply(c, ok_reply ? ok_reply : shared.ok);
}
/* SET key value [NX] [XX] [EX <seconds>] [PX <milliseconds>] */
void setCommand(client *c) {
int j;
robj *expire = NULL;
int unit = UNIT_SECONDS;
int flags = OBJ_SET_NO_FLAGS;
for (j = 3; j < c->argc; j++) {
char *a = c->argv[j]->ptr;
robj *next = (j == c->argc-1) ? NULL : c->argv[j+1];
if ((a[0] == 'n' || a[0] == 'N') &&
(a[1] == 'x' || a[1] == 'X') && a[2] == '\0' &&
!(flags & OBJ_SET_XX))
{
flags |= OBJ_SET_NX;
} else if ((a[0] == 'x' || a[0] == 'X') &&
(a[1] == 'x' || a[1] == 'X') && a[2] == '\0' &&
!(flags & OBJ_SET_NX))
{
flags |= OBJ_SET_XX;
} else if ((a[0] == 'e' || a[0] == 'E') &&
(a[1] == 'x' || a[1] == 'X') && a[2] == '\0' &&
!(flags & OBJ_SET_PX) && next)
{
flags |= OBJ_SET_EX;
unit = UNIT_SECONDS;
expire = next;
j++;
} else if ((a[0] == 'p' || a[0] == 'P') &&
(a[1] == 'x' || a[1] == 'X') && a[2] == '\0' &&
!(flags & OBJ_SET_EX) && next)
{
flags |= OBJ_SET_PX;
unit = UNIT_MILLISECONDS;
expire = next;
j++;
} else {
addReply(c,shared.syntaxerr);
return;
}
}
c->argv[2] = tryObjectEncoding(c->argv[2]);
setGenericCommand(c,flags,c->argv[1],c->argv[2],expire,unit,NULL,NULL);
}
/* High level Set operation. This function can be used in order to set
* a key, whatever it was existing or not, to a new object.
*
* 1) The ref count of the value object is incremented.
* 2) clients WATCHing for the destination key notified.
* 3) The expire time of the key is reset (the key is made persistent).
*
* All the new keys in the database should be craeted via this interface. */
void setKey(redisDb *db, robj *key, robj *val) {
if (lookupKeyWrite(db,key) == NULL) {
dbAdd(db,key,val);
} else {
dbOverwrite(db,key,val);
}
incrRefCount(val);
removeExpire(db,key);
// 看這里?? 標(biāo)記hash表中所有已經(jīng)watch這個(gè)key的所有客戶端狀態(tài)為CLIENT_DIRTY_CAS
// 如果我原先的值為1福侈，這里set為1，也會(huì)執(zhí)行這個(gè)方法卢未。所以說(shuō)和值變沒變沒關(guān)系肪凛。
signalModifiedKey(db,key);
}
void signalModifiedKey(redisDb *db, robj *key) {
touchWatchedKey(db,key);
}
/* 更新hash表中相應(yīng)客戶端的狀態(tài)為CLIENT_DIRTY_CAS */
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;
}
}
