如何使用Redis哨兵模式和集群模式？

为了提高查询效率，会使用读写分离的方案。主库负责写操作，从库负责读取操作并且为只读属性。使用一主两从的拓扑关系讲述redis的读写分离方案，如图：

redis复制

redis的读写分离基于redis的复制机制实现，环境搭建的过程可以参考这位网友的介绍Redis集群主从复制（一主两从）搭建配置教程【Windows环境】。该机制存在读写分离的使用场景下有如下隐患：

• 复制数据有延迟

master采用异步复制的方式把数据复制到slave从库。master生成RDB压缩镜像文件，然后发送给slave节点；slave节点成功接收RDB文件后，清理当前节点的数据，然后执行RDB文件录入数据；slave节点如果开启了AOF持久化方式，会生成AOF文件，如果文件比较大的话甚至会对AOF文件执行重写操作。所以，从库数据与主库数据不一致的情景。

• 节点故障

master节点故障，写操作无法执行，这时需要人工介入手动切换master节点，从从节点中选择一个节点作为新的master节点。slave节点故障，连接到该节点的查询操作无法执行。当出现这两种节点故障时，客户端应用程序没有得到通知，就无法继续正常工作。

redis sentinel哨兵模式

redis使用Redis Sentinel哨兵模式实现了redis的高可用方案，环境搭建过程可以参考这位网友的介绍Redis哨兵（Sentinel）模式。一主二从的拓扑关系图如下：

该模式的主要功能如下：

• 监控

每个Sentinel节点，都会监控所有redis数据节点，特别是master节点的监控，如果redis数据节点故障，就会sentinel集合就会马上进行故障转移。客户端应用程序是通过sentinel集合中的单个sentinel节点获取redis master主节点的配置配置信息。所以，sentinel节点也会监控Sentinel集合中其余的Sentiel节点，即使其中某个sentinel节点故障，也不会影响主从的高可用，提高了Sentinel节点的容错性。

• 故障转移

Sentienl集合检测到master节点故障，会通过主观下线和客观下线机制，master选举，等一些列措施从slave节点中选举新的master节点，并完成新的主从模式的配置自动切换。

• 消息通知

当redis数据节点出现故障时，Sentinel节点会通知客户端应用程序，master节点的地址切换。客户端应用程序通过master-name主节点名称（标识redis master主节点）和sentinel节点列表实现与redis数据节点的配置信息的获取，类似于服务注册于发现的方式。例如，客户端应用程序通过master-name主节点名称，获取主节点的配置信息。调用sentinel节点的sentinel get-master-addr-by-name master-name API就可获取对应主节点的相关信息。

Redis Cluster集群模式

当遇到单机内存，并发，流程等瓶颈时，就可以采用集群模式，达到数据的分区，以及负载均衡的目的。Redis Cluster数据分片的规则，采用虚拟槽分区的规则，公式为slot=CRC16（key）&16383，维护虚拟槽与数据键值的映射关系，由于虚拟槽的hash比较均匀，缓存的数据比较均衡的分配到各个虚拟槽中。集群中，redis的数据节点维护与虚拟槽的关系，解耦了与缓存key的映射关系。

redission客户端集成springboot

现在比较流行的java客户端redission，支持单机，哨兵模式，以及集群的模式的集成。redission整合springboot的示例代码如下：

@Configuration
@EnableConfigurationProperties({RedisProperties.class})
public class RedisConfig {

    private final static String SSL_REDIS = "redis://";

    @Autowired
    private RedisProperties redisProperties;

    /**
     * redis FastJson序列化
     */
    @Bean
    @ConditionalOnMissingBean(value = RedisSerializer.class)
    public RedisSerializer<Object> redisFastJsonSerializer() {
        return new GenericFastJsonRedisSerializer();
    }

    @Bean
    public StringRedisTemplate stringRedisTemplate(@Autowired RedisConnectionFactory redisConnectionFactory) {

        StringRedisTemplate template = new StringRedisTemplate();
        template.setConnectionFactory(redisConnectionFactory);
        return template;
    }

    /**
     * 设置自定义序列化的RedisTemplate
     *
     * @param redisConnectionFactory
     * @param redisFastJsonSerializer 序列化
     * @return
     */
    @Bean
    public RedisTemplate<String, Object> redisTemplate(@Autowired RedisConnectionFactory redisConnectionFactory,
                                                       @Autowired RedisSerializer<Object> redisFastJsonSerializer) {

        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(redisConnectionFactory);
        // value，hash value设置FastJson序列化
        template.setHashValueSerializer(redisFastJsonSerializer);
        template.setValueSerializer(redisFastJsonSerializer);
        // key，hash key使用String序列化
        RedisSerializer<String> stringRedisSerializer = RedisSerializer.string();
        template.setHashKeySerializer(stringRedisSerializer);
        template.setKeySerializer(stringRedisSerializer);
        return template;
    }

    /**
     * 使用RedissonConnectionFactory
     */
    @Bean
    public RedissonConnectionFactory redissonConnectionFactory() {
        return new RedissonConnectionFactory(redissonClient());
    }

    /**
     * redisson 客户端配置
     */
    @Bean(destroyMethod = "shutdown")
    @ConditionalOnMissingBean({RedisProperties.class})
    public org.redisson.api.RedissonClient redissonClient() {

        Config config = new Config();

        RedisProperties.Sentinel sentinel;
        RedisProperties.Cluster cluster;
        // 哨兵模式
        if (Objects.nonNull(sentinel = this.redisProperties.getSentinel())) {

            List<String> newNodes = sentinel.getNodes()
                    .stream().map((index) -> index.startsWith(SSL_REDIS) ? index : SSL_REDIS + index)
                    .collect(Collectors.toList());

            SentinelServersConfig serverConfig = config.useSentinelServers()
                    .addSentinelAddress(newNodes.toArray(new String[0]))
                    .setPassword(this.redisProperties.getPassword())
                    .setMasterName(sentinel.getMaster())
                    .setReadMode(ReadMode.SLAVE);
            serverConfig.setDatabase(redisProperties.getDatabase());
        }
        // 集群模式
        else if (Objects.nonNull(cluster = this.redisProperties.getCluster())) {

            config.useClusterServers().addNodeAddress(cluster.getNodes().toArray(new String[0]));
        }
        // 单机模式
        else {

            String address = this.redisProperties.getHost() + ":" + this.redisProperties.getPort();
            config.useSingleServer().setAddress(address);
        }
        return Redisson.create(config);
    }
}

 并且，继承实现了java的Lock的各类分布式锁的实现，例如，读写锁，可重入锁。分布式锁的详细内容可以参考什么是分布式锁?Zookeeper和Redis是如何实现的？。以RedissonClient实现分布式锁的使用示例如下：

public MenuVo save(SaveMenuDto saveMenuDto) {

        LoginUser loginUser = LoginUserUtils.getLoginUser();
        String lockKey = MENU_LOCK + loginUser.getUserName();
        RLock lock = redissonClient.getLock(lockKey);
        // 加锁
        lock.lock();
        try {
            
            // 执行逻辑
            Menu addingMenu = MenuConverter.INSTANCE.fromSaveMenuDto(saveMenuDto);
            addingMenu.setVisible(MenuVisible.show.getValue());
            addingMenu.setStatus(MenuStatus.normal.getValue());
            menuService.save(addingMenu);
            Menu menu = menuService.getById(addingMenu.getId());
            return MenuConverter.INSTANCE.toMenuVo(menu);
        } finally {
            
            // 释放锁
            lock.unlock();
        }
    }