Redis3.0 及其之后的版本提供了 redis-cluster 集群支持，用于在多个redis节点间共享数据，以提高服务的可用性。构建 redis-cluster 集群可以通过 redis-trib.rb 工具来完成。redis-trib.rb 是redis官方提供的一个集群管理工具，集成在redis安装包的 src 目录下。redis-trib.rb 封装了redis提供的集群命令，使用简单、便捷。
一、搭建集群
（1）因为 redis-trib.rb 是由ruby语言编写的，所以使用该工具需要ruby语言环境的支持。
（2）要启用redis-cluster集群，需要先修改redis配置文件集群配置部分的内容（xxxxredis.conf）
- cluster-enabled yes ：启用redis-cluster集群
- cluster-config-file nodes-6379.conf：集群节点配置文件，该文件无需手工修改，由redis自动维护（创建和更新）， 需要注意，单机运行多实例时，确保该文件没有被其他实例覆盖（不允许重名）
- cluster-node-timeout 15000：节点超时时长（毫秒）
为了方便进行演示，这里分别以端口 6379、6380、6381 各启用一个实例来代表不同的redis服务器
开启3台redis服务，查看进程
（3）创建集群
创建集群使用create 命令完成，create 命令的格式为：
redis-cluster集群至少需要3个可用节点。
（4）查看集群：使用info命令
（5） 当系统因为意外宕机，重启所有节点，集群会自动恢复。
二、主从复制模型
刚才说到，redis-cluster至少需要保证3个节点可用。那么为了避免节点宕机导致服务不可用，我们就需要添加主从配置，为集群中的节点增加从节点；使其在主节点宕机时，能够将从节点提升为主节点代替原来的主节点进行工作。
在非集群的单节点环境中，配置主从关系的方法大致有 2 种：
- 修改从服务器配置文件 redis.conf 的 slaveof <masterip> <masterport> 选项；
- 在从服务器上使用slaveof 命令直接指定主服务器。
然而在redis-cluster集群环境中，启用了cluster配置的情况下slaveof 是不可用的。此时我们需要借助于 redis-trib.rb 工具来进行从节点的添加操作，先将节点加入集群中，然后再声明节点为原有节点的从节点。
（1）启用服务后先将该节点加入集群，直接使用redis-trib.rb的 add-node 命令即可实现：
格式：redis-cli --cluster add-node new_host:new_port existing_host:existing_port
查看到该节点已经成功加入了集群中，并且该节点并没有分配哈希槽
（2）声明该节点为集群中某一节点的从节点，需要使用客户端进入该节点（此处即为新增的从节点 6382）进行设置，使用 cluster replicate 命令指定所属主节点ID。
主节点ID可以使用客户端连接到集群后通过命令 cluster nodes 查看 ：
把6382节点设置为6379的从节点
查看集群,可以看到6379下又一个从节点
查询出 6379 端口服务对应的PID，然后通过 kill 将服务关闭，使该节点在集群上不可用；此时查看集群信息，能够发现从节点 6382 自动提升为主节点，顶替了不可用的 6379 节点。
重新启动已宕机的服务后，该节点将会被当做从节点添加到管理原先的哈希槽分配范围的节点上。这里也就是添加到了 6382 节点上，6382 节点管理的哈希槽就是原先由 6379节点所管理的
三、客户端连接 redis-cluster
客户端在连接 redis 服务器时带上参数 -c 即为连接到cluster集群
可以看到，在 6380 端口的服务器上存储一个string类型的键值对 k1= v1的时，操作被重定向到了 6381 端口的服务器上，而 k1= v1这个键值对最终也被存储在了 6381 端口的服务器里。
同理，在获取数据时，也会重定向到对应数据实际存储的服务器上，然后在该服务器上进行操作。
四、单独连接集群上的节点
需要注意的是，节点在加入集群后，如果不声明参数 -c 连接集群而是单独连接集群上的节点，那么在操作时如果需要重定向到其他的服务器，是无法成功重定向然后完成操作的。
例如键值对 k1= v1存储在 6381 端口的服务器上，此时如果我们单独连接到 6380 端口的服务器上进行操作，那么该操作是无法成功的。
而如果无需重定向，则能成功完成操作。
五、Spring整合 redis-cluster
<!-- jedis连接池配置 --><bean id="jedisPoolConfig"class="redis.clients.jedis.JedisPoolConfig"><!-- 同一时刻可分配最大连接数 默认值8 设置为负数时不做限制 --><property name="maxTotal" value="8"></property><!-- 最大空闲连接 默认8个 超出连接将被释放 --><property name="maxIdle" value="8"></property><!-- 请求连接最大等待时间(毫秒)，默认值 无限期 ，超出时间将抛出异常 --><property name="maxWaitMillis" value="3000"></property></bean><!-- jedis集群配置 --><bean id="jedisCluster" class="redis.clients.jedis.JedisCluster"><constructor-arg name="nodes"><set><bean class="redis.clients.jedis.HostAndPort"><constructor-arg name="host" value="192.168.147.128"></constructor-arg><constructor-arg name="port" value="6379" /></bean><bean class="redis.clients.jedis.HostAndPort"><constructor-arg name="host" value="192.168.147.128"></constructor-arg><constructor-arg name="port" value="6380" /></bean><bean class="redis.clients.jedis.HostAndPort"><constructor-arg name="host" value="192.168.147.128"></constructor-arg><constructor-arg name="port" value="6381" /></bean><bean class="redis.clients.jedis.HostAndPort"><constructor-arg name="host" value="192.168.147.128"></constructor-arg><constructor-arg name="port" value="6382" /></bean></set></constructor-arg><constructor-arg name="poolConfig"ref="jedisPoolConfig" /></bean>测试代码：
@Testpublic void test() {ApplicationContext context = new ClassPathXmlApplicationContext("applicationContext.xml");JedisCluster jedisCluster = context.getBean("jedisCluster",JedisCluster.class);Assert.assertNotNull(jedisCluster);String result = jedisCluster.set("k2", "v2");Assert.assertNotNull(result);Assert.assertEquals("OK", result);String value = jedisCluster.get("k2");Assert.assertNotNull(value);Assert.assertEquals("v2", value);}运行成功
如果本文对您有很大的帮助，还请点赞关注一下。
