1. RedisCluster集群结构

 Redis-Cluster采用无中心结构，每个节点保存数据和整个集群状态,每个节点都和其他所有节点连接。2. RedisCluster集群特点

① 所有的redis节点彼此互联(PING-PONG机制),内部使用二进制协议优化传输速度和带宽；

② 每个主节点都有一个从节点，当主节点故障，Cluster会按照规则实现主备的高可用性；

③ 可以相对平滑扩/缩容节点；

④ Redis Cluster采用虚拟槽分区，所有的key根据哈希函数映射到0~16383槽内，key是数据分区的最小粒度，因为不能讲一个大的键值对象，如hash，list等映射到不同的节点上；

从RedisCluster结构及其主要特点分析，当共享平台用户量增加时，如果业务访问的key分布在不通的槽分区，不同的节点上，此时即可大大增加Redis的并发访问数，较单节点模式在性能方面会有较大的提升。3. maven依赖整改

使用RedisCluster模式，需要使用2.9.0以上版本才能支持Redis密码认证。由于产品共享平台是典型的Spring项目，所以在spring-data-redis的基础上配置spring实现操作Redis Cluster。

               redis.clients       jedis       2.9.0                 org.springframework.data        spring-data-redis        2.0.8.RELEASE   

4. 项目配置文件配置项整改

RedisCluster集群至少需要6个节点（3主3从，每个主节点分别拥有一个从节点）

    #redis cluster config    #RedisCluster集群节点及端口信息    share.redis.cluster.nodes=127.0.0.1:6379,127.0.0.1:6380,127.0.0.1:6381,127.0.0.1:6382,127.0.0.1:6383,127.0.0.1:6384    #Redis密码    share.redis.password=    #在群集中执行命令时要遵循的最大重定向数目    share.redis.cluster.max-redirects=5    #Redis连接池在给定时间可以分配的最大连接数。使用负值无限制    share.redis.jedis.pool.max-active=1000    #以毫秒为单位的连接超时时间    share.redis.timeout=2000    #池中“空闲”连接的最大数量。使用负值表示无限数量的空闲连接    share.redis.jedis.pool.max-idle=8    #目标为保持在池中的最小空闲连接数。这个设置只有在设置max-idle的情况下才有效果    share.redis.jedis.pool.min-idle=5    #连接分配在池被耗尽时抛出异常之前应该阻塞的最长时间量（以毫秒为单位）。使用负值可以无限期地阻止    share.redis.jedis.pool.max-wait=1000    #redis cluster只使用db0    share.redis.index=0

5. 新增RedisCluster配置项类

新增RedisCluster配置项类用户收集配置文件中的配置项

    @Component    public class ShareClusterRedisProperties {    @Value("${spring.share.redis.timeout}")        private Integer redisTimeout;    @Value("${spring.share.redis.jedis.pool.max-active}")        private Integer poolMaxActive;        @Value("${spring.share.redis.jedis.pool.max-idle}")        private Integer poolMaxIdle;        @Value("${spring.share.redis.jedis.pool.min-idle}")        private Integer poolMinIdle;        @Value("${spring.share.redis.jedis.pool.max-wait}")        private Integer poolMaxWait;        @Value("${spring.share.redis.cluster.nodes}")        private List clusterNodes;        @Value("${spring.share.redis.cluster.max-redirects}")        private Integer clusterMaxRedirects;        @Value("${spring.share.redis.password}")        private String password;        -- getter/setter方法省略 --    }

6. 新增RedisCluster集群配置类

    @Configuration    public class ShareJedisClusterConfig {        @Autowired        ShareClusterRedisProperties redisProperties;    /**         * 注意：         * 这里返回的JedisCluster是单例的，并且可以直接注入到其他类中去使用         * @return         */        @Bean("shareJedisCluster")        public JedisCluster getJedisCluster() {            Set nodes = new HashSet();            for (String ipPort : redisProperties.getClusterNodes()) {                String[] ipPortPair = ipPort.split(":");                nodes.add(new HostAndPort(ipPortPair[0].trim(), Integer.valueOf(ipPortPair[1].trim())));            }            //需要密码连接的创建对象方式            if (StringUtils.isBlank(redisProperties.getPassword())) {                return new JedisCluster(nodes,redisProperties.getRedisTimeout(),1000,1,new GenericObjectPoolConfig());            } else {                return new JedisCluster(nodes,redisProperties.getRedisTimeout(),1000,1,redisProperties.getPassword() ,new GenericObjectPoolConfig());            }        }        @Bean        public RedisConnectionFactory redisConnectionFactory() {            JedisPoolConfig poolConfig = new JedisPoolConfig();            poolConfig.setMaxTotal(redisProperties.getPoolMaxActive());            poolConfig.setMaxIdle(redisProperties.getPoolMaxIdle());            poolConfig.setMinIdle(redisProperties.getPoolMinIdle());            poolConfig.setMaxWaitMillis(redisProperties.getPoolMaxWait());            JedisClientConfiguration clientConfig = JedisClientConfiguration.builder()                    .usePooling()                    .poolConfig(poolConfig)                    .and()                    .readTimeout(Duration.ofMillis(redisProperties.getRedisTimeout()))                    .build();             // cluster模式            RedisClusterConfiguration redisConfig = new RedisClusterConfiguration();            redisConfig.setMaxRedirects(redisProperties.getClusterMaxRedirects());            for (String ipPort :redisProperties.getClusterNodes()){                String[] ipPortArr = ipPort.split(":");                redisConfig.clusterNode(ipPortArr[0], Integer.parseInt(ipPortArr[1].trim()));            }             return new JedisConnectionFactory(redisConfig, clientConfig);        }    }

7. Redis公共类整改

在操作Redis Cluster集群的时候，只需自动注入JedisCluster即可，用JedisCluster替换现有的Redis公共方法中使用到redis的地方即可。

但是 JedisCluster 从性能方面等考虑，并没有提供 Jedis 下提供的 keys 方法， keys 方法，主要用于通配符模式匹配返回满足条件 的 key。keys 方法 还是比较有用的，所以这里需要在Jedis的基础上进行扩展。

    public int keys(final String pattern) {            try {                final Set keySet = new HashSet();                final Map nodes = jedisCluster.getClusterNodes();                for (String k : nodes.keySet()) {                    JedisPool pool = nodes.get(k);                      //获取Jedis对象，Jedis对象支持keys模糊查询                    Jedis jedis = pool.getResource();                      final Set set = jedis.keys(pattern);                    keySet.addAll(set);                }                                return keySet.size();            } catch (Exception e) {                logger.error("异常信息", e);                throw e;            }        }        /**         * 模糊查询给定的pattern的所有keys列表         *         * @param pattern 模糊查询         * @return 返回当前pattern可匹配的对象keys列表         */        public Set keyList(final String pattern) {                 try {                final Set keySet = new HashSet();                final Map nodes = jedisCluster.getClusterNodes();                for (String k : nodes.keySet()) {                    JedisPool pool = nodes.get(k);                      //获取Jedis对象，Jedis对象支持keys模糊查询                    Jedis jedis = pool.getResource();                      final Set set = jedis.keys(pattern);                    keySet.addAll(set);                }                return keySet;            } catch (Exception e) {                logger.error("异常信息", e);                throw e;            }