前言：
redis在redis3.0版本之后推出redis cluster模式集群，redis cluster是官方提供的分布式解决方案。当一个redis节点挂了可以快速切到另一个节点中。当遇到单机内存、并发瓶颈时可以考虑使用redis cluster。
集群的内容会比较长，这一章会分为两篇作为描述:
《集群之分布式cluster建立集群关系》与《集群之分布式cluster原理》
(一)cluster基础知识
1.1 了解cluser
一个redis集群通常由多个节点组成，起初每个阶段都是独立的个体。它们都在自己的集群当中。如果要构建一个真正的集群，我们必须将各个独立的节点连接在一起，构成一个包含多节点的集群。
如图中所示：
| 节点 | 主 | 从 |
| 第一组节点 | 127.0.0.1:6379 | 127.0.0.1:6389 |
| 第二组节点 | 127.0.0.1:6380 | 127.0.0.1:6390 |
| 第三组节点 | 127.0.0.1:6381 | 127.0.0.1:6391 |
我们要构成这些节点，可以通过命令或者配置构成。
1.2 配置cluster
1)配置信息
port 6389
#开启cluster模式
cluster-enabled yes
#配置节点之间超时时间
cluster-node-timeout 15000
#这个配置很重要，cluster开启必须重命名指定cluster-config-file
#不能与别的节点相同，否则会启动失败,最好按主机+端口命名
#其次，该文件保存了本节点与其他节点的信息及关系
cluster-config-file nodes-6389.conf
2)创建redis cluster
redis-cli --cluster create 127.0.0.1:6379 127.0.0.1:6380 127.0.0.1:6381 127.0.0.1:6389 127.0.0.1:6390 127.0.0.1:6391 --cluster-replicas 1
该条命令时redis5.0客户端才支持的。除了这种方式还可以使用redis-trib.rb
在创建节点过程中如果时遇到[ERR] Not all 16384 slots are covered by nodes.错误，
或者使用过程中遇到(error) CLUSTERDOWN The cluster is down。
可以执行如下命令:
redis-cli --cluster fix 127.0.0.1:6379
修复过程中，我们可以看到我们集群中有16384个槽可以分别指派给集群中的各个节点。
1.3 基础命令
#查看当前节点
CLUSTER NODES
#将 ip 和 port 所指定的节点添加到集群中
CLUSTER MEET .
#从集群中移除 node_id 指定的节点
CLUSTER FORGET
#将当前节点设置为 node_id 指定的节点的从节点
CLUSTER REPLICATE
#将节点的配置文件保存到硬盘里面
CLUSTER SAVECONFIG
#将一个或多个槽（slot）指派（assign）给当前节点
CLUSTER ADDSLOTS [slot ...]
#移除一个或多个槽对当前节点的指派
CLUSTER DELSLOTS [slot ...]
#移除当前节点所有槽
CLUSTER FLUSHSLOTS
#将槽 slot 指派给 node_id 指定的节点，如果槽已经指派给另一个节点，
#那么先让另一个节点删除该槽，然后再进行指派
CLUSTER SETSLOT NODE
#将本节点的槽 slot 迁移到 node_id 指定的节点中
CLUSTER SETSLOT MIGRATING
#从 node_id 指定的节点中导入槽 slot 到本节点
CLUSTER SETSLOT IMPORTING
#取消对槽 slot 的导入（import）或者迁移（migrate）
CLUSTER SETSLOT STABLE
#计算键 key 应该被放置在哪个槽上
CLUSTER KEYSLOT
#返回槽 slot 目前包含的键值对数量
CLUSTER COUNTKEYSINSLOT
#返回 count 个 slot 槽中的键
CLUSTER GETKEYSINSLOT
(二)源码分析
2.1基础结构
#define CLUSTER_SLOTS 16384 //对应卡槽最大数量
typedefstructclusterNode {
mstime_t ctime; /* 创建节点时间. */
char name[CLUSTER_NAMELEN]; /* 节点名称 hex 字节串, 40个字节 */
int flags; /* 节点标识,CLUSTER_NODE_... */
uint64_t configEpoch; /* 节点当前的配置纪元，用于实现故障转移 */
unsignedchar slots[CLUSTER_SLOTS/8]; /* slots handled by this node */
int numslots; /* 此节点处理的插槽数 */
int numslaves; /* 如果这是主节点，则从节点的数量 */
structclusterNode **slaves; /* 节点从节点指针 */
structclusterNode *slaveof; /* 指向主节点的指针。*/
mstime_t ping_sent; /* 最新一个 ping 时间 */
mstime_t pong_received; /* 最新一个回复 pong 时间*/
mstime_t fail_time; /* 设置失败标志的Unix时间 */
mstime_t voted_time; /* 上一次投票时间 */
mstime_t repl_offset_time; /* 我们收到此节点偏移量的Unix时间 */
mstime_t orphaned_time; /* 孤立主条件开始的时间 */
longlong repl_offset; /* 此节点的最后一个已知复制偏移量. */
char ip[NET_IP_STR_LEN]; /* 节点IP地址 */
int port; /* 节点端口 */
int cport; /* 此节点的最新已知群集端口. */
clusterLink *link; /* 节点 TCP/IP 连接信息 */
list *fail_reports; /* 失败节点列表 */
} clusterNode;
CLUSTER_SLOTS宏是我们cluster卡槽数量，link保存了连接节点所需的有关信息， 比如套接字描述符， 输入缓冲区和输出缓冲区：
typedefstructclusterLink {
mstime_t ctime; /* Link 创建时间 */
int fd; /* TCP socket 描述符 */
sds sndbuf; /* 输出缓冲区 */
sds rcvbuf; /* 输入缓冲区 */
structclusterNode *node; /* 与这个连接相关联的节点，如果没有的话就为 NULL */
} clusterLink;
每个连接都会都会维护一个clusterState状态。
typedefstructclusterState {
clusterNode *myself; /* 指向当前节点指针 */
uint64_t currentEpoch; /* 集群当前的配置纪元，用于故障恢复 */
int state; /* CLUSTER_OK, CLUSTER_FAIL, ... */
int size; /* 集群中至少处理着一个槽的节点的数量 */
dict *nodes; /* 集群节点名单 对应 clusterNode 结构体 */
//...省略
} clusterState;
2.2 初始化cluster
cluster命令方法：
voidclusterCommand(client *c) {
if (server.cluster_enabled == 0) { //判断是否开启cluster
addReplyError(c,"This instance has cluster support disabled");
return;
}
//判断参数必须是4个或者5个，且第二个参数等于meet
if (!strcasecmp(c->argv[1]->ptr,"meet") && (c->argc == 4 || c->argc == 5)) {
/* CLUSTER MEET [cport] */
longlong port, cport;
if (getLongLongFromObject(c->argv[3], &port) != C_OK) { //获得端口参数
addReplyErrorFormat(c,"Invalid TCP base port specified: %s",
(char*)c->argv[3]->ptr);
return;
}
if (c->argc == 5) { //5个参数时
if (getLongLongFromObject(c->argv[4], &cport) != C_OK) { // 获得cport
addReplyErrorFormat(c,"Invalid TCP bus port specified: %s",
(char*)c->argv[4]->ptr);
return;
}
} else {
cport = port + CLUSTER_PORT_INCR; //默认清楚port + 10000
}
if (clusterStartHandshake(c->argv[2]->ptr,port,cport) == 0 && . //握手
errno == EINVAL)
{
addReplyErrorFormat(c,"Invalid node address specified: %s:%s",
(char*)c->argv[2]->ptr, (char*)c->argv[3]->ptr);
} else {
addReply(c,shared.ok);
}
}
//。。。省略
}
握手函数源码
intclusterStartHandshake(char *ip, int port, int cport) {
clusterNode *n;
char norm_ip[NET_IP_STR_LEN];
structsockaddr_storagesa;
/* IP健全性检查 */
if (inet_pton(AF_INET,ip,
&(((struct sockaddr_in *)&sa)->sin_addr)))
{
sa.ss_family = AF_INET;
} elseif (inet_pton(AF_INET6,ip,
&(((struct sockaddr_in6 *)&sa)->sin6_addr)))
{
sa.ss_family = AF_INET6;
} else {
errno = EINVAL;
return0;
}
/* 端口健全性检测 */
if (port <= 0 || port > 65535 || cport <= 0 || cport > 65535) {
errno = EINVAL;
return0;
}
/* 网络ip地址转寒*/
memset(norm_ip,0,NET_IP_STR_LEN);
if (sa.ss_family == AF_INET)
inet_ntop(AF_INET,
(void*)&(((struct sockaddr_in *)&sa)->sin_addr),
norm_ip,NET_IP_STR_LEN);
else
inet_ntop(AF_INET6,
(void*)&(((struct sockaddr_in6 *)&sa)->sin6_addr),
norm_ip,NET_IP_STR_LEN);
//判断是否正在握手中，防止重复握手
if (clusterHandshakeInProgress(norm_ip,port,cport)) {
errno = EAGAIN;
return0;
}
/* 创建node节点结构*/
n = createClusterNode(NULL,CLUSTER_NODE_HANDSHAKE|CLUSTER_NODE_MEET);
memcpy(n->ip,norm_ip,sizeof(n->ip));
n->port = port;
n->cport = cport;
//添加nodes节点到server.cluster->nodes中
clusterAddNode(n);
return1;
}
加入node到server.cluster->nodes后，serverCron中会调用clusterCron。该函数中会判断处于握手状态的节点是否握手超时，如果是。则调用clusterDelNode函数删除节点。如果节点的节点 TCP/IP 连接信息 等于空时（link == NULL），会发起tcp/ip连接，且将fd信息保存到节点link->fd中。已经ping信息。（该部分下一章详细讲解）
2.3键值设置流程
1）第一步通过6379端口登录
#redis-cli -p 6379 -c
2）下断点processCommand
intprocessCommand(client *c) {
//。。。省略
/* 如果启用群集，请在此处执行群集重定向。
*但是，如果发生以下情况，则不执行重定向：
*1）这个命令的发送者是我们的master。
*2）命令没有键参数。*/
if (server.cluster_enabled &&
!(c->flags & CLIENT_MASTER) &&
!(c->flags & CLIENT_LUA &&
server.lua_caller->flags & CLIENT_MASTER) &&
!(c->cmd->getkeys_proc == NULL && c->cmd->firstkey == 0 &&
c->cmd->proc != execCommand))
{
int hashslot;
int error_code;
//获得key属于哪个slot，内部通过keyHashSlot函数计算slot
clusterNode *n = getNodeByQuery(c,c->cmd,c->argv,c->argc,
&hashslot,&error_code);
if (n == NULL || n != server.cluster->myself) { //判断node不是自己
if (c->cmd->proc == execCommand) {
discardTransaction(c);
} else {
flagTransaction(c);
}
clusterRedirectClient(c,n,hashslot,error_code); //跳转
return C_OK;
}
}
//。。。省略
}
计算slot分布函数，keyHashSlot
unsignedintkeyHashSlot(char *key, int keylen) {
int s, e; /* start-end indexes of { and } */
for (s = 0; s < keylen; s++) //计算{的位置
if (key[s] == '{') break;
/* 没有{情况下直接 crc16(key) % 16384 = crc16(key) & 0x3FFF */
if (s == keylen) return crc16(key,keylen) & 0x3FFF;
/* 有'{'情况，计算}的位置 */
for (e = s+1; e < keylen; e++)
if (key[e] == '}') break;
/* 没有 '}'或者不是{} 闭合 */
if (e == keylen || e == s+1) return crc16(key,keylen) & 0x3FFF;
/* {key}闭合时取中间的key */
return crc16(key+s+1,e-s-1) & 0x3FFF;
}
函数中计算slot的分布情况,以crc16(key) % 16384 。
上图为计算slots
上图为跳转函数
上图为跳转后展示
总结：
1.cluster中是采用hash槽，有16384个槽可以分别指派给集群中的各个节点。每个节点都会记录哪些槽分配给自己，哪些槽指派给其他节点。在node_xxx.conf中也可以体现。
2.redis5.0客户端通过redis-cli --cluster 可以创建cluster集群，通过redis-cli --cluster fix 可以修复集群重新分配槽。
3.调用命令时，会发起调用processCommand函数，函数中如果开启cluster，计算的slot后得到node节点不是自己。则客户端连接则会跳转。
4.CLUSTER MEET命令可以加入cluster，加入时发起握手相关操作，检测服务正常后，加入节点信息到server.cluster->nodes中。
