前言：

redis在redis3.0版本之后推出redis cluster模式集群，redis cluster是官方提供的分布式解决方案。当一个redis节点挂了可以快速切到另一个节点中。当遇到单机内存、并发瓶颈时可以考虑使用redis cluster。

集群的内容会比较长，这一章会分为两篇作为描述:

《集群之分布式cluster建立集群关系》与《集群之分布式cluster原理》

(一)cluster基础知识

1.1 了解cluser

一个redis集群通常由多个节点组成，起初每个阶段都是独立的个体。它们都在自己的集群当中。如果要构建一个真正的集群，我们必须将各个独立的节点连接在一起，构成一个包含多节点的集群。

如图中所示：

我们要构成这些节点，可以通过命令或者配置构成。

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中。