Redis cluster的思路是,当需要存储一个key时,先将key模拟发布在一个slot槽,在将key实际分布在slot关联的实际服务器上.
从头到尾看一下,跟key读写相关的源码.
1int main(int argc, char **argv) 2{ 3 . 4 . 5 . 6 /* 确保只有0号数据库里有数据 */ 7 if (server.cluster_enabled) { 8 if (verifyClusterConfigWithData() == REDIS_ERR) { 9 redisLog(REDIS_WARNING, 10 "You can't have keys in a DB different than DB 0 when in " 11 "Cluster mode. Exiting."); 12 exit(1); 13 } 14 } 15 . 16 . 17 . 18} 19 20 21 22/* 检查当前节点的节点配置是否正确,包含的数据是否正确 */ 23int verifyClusterConfigWithData(void) { 24 int j; 25 int update_config = 0; 26 27 if (nodeIsSlave(myself)) return REDIS_OK;//不对从节点进行检查 28 29 /* 确保只有0号数据库有数据 */ 30 for (j = 1; j < server.dbnum; j++) { 31 if (dictSize(server.db[j].dict)) return REDIS_ERR; 32 } 33 34 /* 检查槽表是否都有相应的节点，如果不是的话，进行修复 */ 35 for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) { 36 if (!countKeysInSlot(j)) continue; //空槽直接跳过 37 38 if (server.cluster->slots[j] == myself ||server.cluster->importing_slots_from[j] != NULL) continue;// 跳过正在导入的槽 39 40 /* If we are here data and cluster config don't agree, and we have 41 * slot 'j' populated even if we are not importing it, nor we are 42 * assigned to this slot. Fix this condition. */ 43 44 update_config++; 45 if (server.cluster->slots[j] == NULL) { 46 /* 处理未被分配的槽 */ 47 redisLog(REDIS_WARNING, "I've keys about slot %d that is " 48 "unassigned. Taking responsability " 49 "for it.",j); 50 clusterAddSlot(myself,j); 51 } else { 52 /* 如果一个槽已经被其他节点接管,那么将槽中的资料发送给对方 */ 53 redisLog(REDIS_WARNING, "I've keys about slot %d that is " 54 "already assigned to a different node. " 55 "Setting it in importing state.",j); 56 server.cluster->importing_slots_from[j] = server.cluster->slots[j]; 57 } 58 } 59 60 if (update_config) clusterSaveConfigOrDie(1);//保存 nodes.conf 文件 61 62 return REDIS_OK; 63}
当客户端传来一条命令时,会执行processCommand()函数,该函数与cluster相关的代码如下:
1int processCommand(redisClient *c) 2{ 3 . 4 . 5 . 6 /* 7 * 如果开启了集群模式，那么在这里进行转向操作。 8 * 不过，如果有以下情况出现，那么节点不进行转向： 9 * 1) 命令的发送者是本节点的主节点 10 * 2) 命令没有 key 参数 11 */ 12 if (server.cluster_enabled && !(c->flags & REDIS_MASTER) && !(c->cmd->getkeys_proc == NULL && c->cmd->firstkey == 0)) 13 { 14 int hashslot; 15 16 if (server.cluster->state != REDIS_CLUSTER_OK) {//集群已下线 17 flagTransaction(c); 18 addReplySds(c,sdsnew("-CLUSTERDOWN The cluster is down. Use CLUSTER INFO for more information\r\n")); 19 return REDIS_OK; 20 21 /* 集群运作正常 */ 22 } else { 23 int error_code; 24 clusterNode *n = getNodeByQuery(c,c->cmd,c->argv,c->argc,&hashslot,&error_code); 25 // 不能执行多键处理命令 26 if (n == NULL) { 27 flagTransaction(c); 28 if (error_code == REDIS_CLUSTER_REDIR_CROSS_SLOT) { 29 addReplySds(c,sdsnew("-CROSSSLOT Keys in request don't hash to the same slot\r\n")); 30 } else if (error_code == REDIS_CLUSTER_REDIR_UNSTABLE) { 31 /* The request spawns mutliple keys in the same slot, 32 * but the slot is not "stable" currently as there is 33 * a migration or import in progress. */ 34 addReplySds(c,sdsnew("-TRYAGAIN Multiple keys request during rehashing of slot\r\n")); 35 } else { 36 redisPanic("getNodeByQuery() unknown error."); 37 } 38 return REDIS_OK; 39 40 // 命令针对的槽和键不是本节点处理的，进行转向 41 } else if (n != server.cluster->myself) { 42 flagTransaction(c); 43 /* 例如 -ASK 10086 127.0.0.1:12345 */ 44 addReplySds(c,sdscatprintf(sdsempty(), "-%s %d %s:%d\r\n", (error_code == REDIS_CLUSTER_REDIR_ASK) ? "ASK" : "MOVED", 45 hashslot,n->ip,n->port)); 46 47 return REDIS_OK; 48 } 49 50 // 如果执行到这里，说明键 key 所在的槽由本节点处理 51 // 或者客户端执行的是无参数命令 52 } 53 } 54 . 55 . 56 . 57}
重点就是getNodeByQuery(),该函数可以根据客户端输入的命令,计算要映射到哪个slot槽上
1clusterNode *getNodeByQuery(redisClient *c, struct redisCommand *cmd, robj **argv, int argc, int *hashslot, int *error_code) { 2 3 // 初始化为 NULL ， 4 // 如果输入命令是无参数命令，那么 n 就会继续为 NULL 5 clusterNode *n = NULL; 6 7 robj *firstkey = NULL; 8 int multiple_keys = 0; 9 multiState *ms, _ms; 10 multiCmd mc; 11 int i, slot = 0, migrating_slot = 0, importing_slot = 0, missing_keys = 0; 12 13 /* Set error code optimistically for the base case. */ 14 if (error_code) *error_code = REDIS_CLUSTER_REDIR_NONE; 15 16 /* We handle all the cases as if they were EXEC commands, so we have 17 * a common code path for everything */ 18 // 集群可以执行事务， 19 // 但必须确保事务中的所有命令都是针对某个相同的键进行的 20 // 这个 if 和接下来的 for 进行的就是这一合法性检测 21 if (cmd->proc == execCommand) { 22 /* If REDIS_MULTI flag is not set EXEC is just going to return an 23 * error. */ 24 if (!(c->flags & REDIS_MULTI)) return myself; 25 ms = &c->mstate; 26 } else { 27 /* In order to have a single codepath create a fake Multi State 28 * structure if the client is not in MULTI/EXEC state, this way 29 * we have a single codepath below. */ 30 ms = &_ms; 31 _ms.commands = &mc; 32 _ms.count = 1; 33 mc.argv = argv; 34 mc.argc = argc; 35 mc.cmd = cmd; 36 } 37 38 /* Check that all the keys are in the same hash slot, and obtain this 39 * slot and the node associated. */ 40 for (i = 0; i < ms->count; i++) { 41 struct redisCommand *mcmd; 42 robj **margv; 43 int margc, *keyindex, numkeys, j; 44 45 mcmd = ms->commands[i].cmd; 46 margc = ms->commands[i].argc; 47 margv = ms->commands[i].argv; 48 49 // 定位命令的键位置 50 keyindex = getKeysFromCommand(mcmd,margv,margc,&numkeys); 51 // 遍历命令中的所有键 52 for (j = 0; j < numkeys; j++) { 53 robj *thiskey = margv[keyindex[j]]; 54 int thisslot = keyHashSlot((char*)thiskey->ptr, 55 sdslen(thiskey->ptr)); 56 57 if (firstkey == NULL) { 58 // 这是事务中第一个被处理的键 59 // 获取该键的槽和负责处理该槽的节点 60 /* This is the first key we see. Check what is the slot 61 * and node. */ 62 firstkey = thiskey; 63 slot = thisslot; 64 n = server.cluster->slots[slot]; 65 redisAssertWithInfo(c,firstkey,n != NULL); 66 /* If we are migrating or importing this slot, we need to check 67 * if we have all the keys in the request (the only way we 68 * can safely serve the request, otherwise we return a TRYAGAIN 69 * error). To do so we set the importing/migrating state and 70 * increment a counter for every missing key. */ 71 if (n == myself && 72 server.cluster->migrating_slots_to[slot] != NULL) 73 { 74 migrating_slot = 1; 75 } else if (server.cluster->importing_slots_from[slot] != NULL) { 76 importing_slot = 1; 77 } 78 } else { 79 /* If it is not the first key, make sure it is exactly 80 * the same key as the first we saw. */ 81 if (!equalStringObjects(firstkey,thiskey)) { 82 if (slot != thisslot) { 83 /* Error: multiple keys from different slots. */ 84 getKeysFreeResult(keyindex); 85 if (error_code) 86 *error_code = REDIS_CLUSTER_REDIR_CROSS_SLOT; 87 return NULL; 88 } else { 89 /* Flag this request as one with multiple different 90 * keys. */ 91 multiple_keys = 1; 92 } 93 } 94 } 95 96 /* Migarting / Improrting slot? Count keys we don't have. */ 97 if ((migrating_slot || importing_slot) && 98 lookupKeyRead(&server.db[0],thiskey) == NULL) 99 { 100 missing_keys++; 101 } 102 } 103 getKeysFreeResult(keyindex); 104 } 105 106 /* No key at all in command? then we can serve the request 107 * without redirections or errors. */ 108 if (n == NULL) return myself; 109 110 /* Return the hashslot by reference. */ 111 if (hashslot) *hashslot = slot; 112 113 /* This request is about a slot we are migrating into another instance? 114 * Then if we have all the keys. */ 115 116 /* If we don't have all the keys and we are migrating the slot, send 117 * an ASK redirection. */ 118 if (migrating_slot && missing_keys) { 119 if (error_code) *error_code = REDIS_CLUSTER_REDIR_ASK; 120 return server.cluster->migrating_slots_to[slot]; 121 } 122 123 /* If we are receiving the slot, and the client correctly flagged the 124 * request as "ASKING", we can serve the request. However if the request 125 * involves multiple keys and we don't have them all, the only option is 126 * to send a TRYAGAIN error. */ 127 if (importing_slot && 128 (c->flags & REDIS_ASKING || cmd->flags & REDIS_CMD_ASKING)) 129 { 130 if (multiple_keys && missing_keys) { 131 if (error_code) *error_code = REDIS_CLUSTER_REDIR_UNSTABLE; 132 return NULL; 133 } else { 134 return myself; 135 } 136 } 137 138 /* Handle the read-only client case reading from a slave: if this 139 * node is a slave and the request is about an hash slot our master 140 * is serving, we can reply without redirection. */ 141 if (c->flags & REDIS_READONLY && 142 cmd->flags & REDIS_CMD_READONLY && 143 nodeIsSlave(myself) && 144 myself->slaveof == n) 145 { 146 return myself; 147 } 148 149 /* Base case: just return the right node. However if this node is not 150 * myself, set error_code to MOVED since we need to issue a rediretion. */ 151 if (n != myself && error_code) *error_code = REDIS_CLUSTER_REDIR_MOVED; 152 153 // 返回负责处理槽 slot 的节点 n 154 return n; 155} 156 157 158 159// 计算给定键应该被分配到那个槽 160unsigned int keyHashSlot(char *key, int keylen) { 161 int s, e; /* start-end indexes of { and } */ 162 163 for (s = 0; s < keylen; s++) 164 if (key[s] == '{') break; 165 166 /* No '{' ? Hash the whole key. This is the base case. */ 167 if (s == keylen) return crc16(key,keylen) & 0x3FFF; 168 169 /* '{' found? Check if we have the corresponding '}'. */ 170 for (e = s+1; e < keylen; e++) 171 if (key[e] == '}') break; 172 173 /* No '}' or nothing betweeen {} ? Hash the whole key. */ 174 if (e == keylen || e == s+1) return crc16(key,keylen) & 0x3FFF; 175 176 /* If we are here there is both a { and a } on its right. Hash 177 * what is in the middle between { and }. */ 178 return crc16(key+s+1,e-s-1) & 0x3FFF; 179}