Java操作Redis：集群搭建，Redis Cluster 集成​

一、Redis集群为何而生？

记得几年前我在做一个电商秒杀系统时，用单节点Redis处理高并发请求，结果遇到了严重的性能瓶颈。后来尝试主从复制方案，却发现数据一致性维护异常麻烦。直到尝试了Redis Cluster，才真正找到了分布式缓存的最优解。

Redis Cluster是Redis官方提供的分布式解决方案，它通过数据分片（Sharding）机制实现分布式存储，采用去中心化的架构设计，支持自动故障转移和横向扩展。当数据量超过单机内存容量时，集群模式能像拼积木一样方便地增加节点。

二、手把手搭建Redis Cluster集群环境（Redis 6.x版本）

2.1 集群规划

我们搭建一个最小可用集群：3主3从共6个节点。使用Docker容器模拟多节点环境：

# 创建Redis配置文件模板
for port in 7000 7001 7002 7003 7004 7005
do
  mkdir -p redis-cluster/${port}
  cat > redis-cluster/${port}/redis.conf <<EOF
port ${port}
cluster-enabled yes
cluster-config-file nodes.conf
cluster-node-timeout 5000
appendonly yes
EOF
done

# 批量启动容器
docker run -d --name redis-7000 -p 7000:7000 \
  -v $PWD/redis-cluster/7000:/usr/local/etc/redis \
  redis:6.2.6 redis-server /usr/local/etc/redis/redis.conf
# 重复执行上述命令修改端口启动其他节点...

# 创建集群（任意节点执行）
redis-cli --cluster create \
  172.17.0.2:7000 172.17.0.3:7001 172.17.0.4:7002 \
  172.17.0.5:7003 172.17.0.6:7004 172.17.0.7:7005 \
  --cluster-replicas 1

2.2 集群验证

执行cluster nodes查看节点关系，当所有主节点显示"connected"，从节点显示"slave"即表示集群就绪。我们可以用redis-cli -c -p 7000进入集群模式测试数据分布。

三、Spring Boot集成Redis Cluster实战（技术栈：Spring Boot 2.7 + Lettuce）

3.1 Maven依赖配置

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
    <groupId>io.lettuce</groupId>
    <artifactId>lettuce-core</artifactId>
    <version>6.1.8.RELEASE</version>
</dependency>

3.2 配置类编写

@Configuration
public class RedisConfig {

    @Value("${spring.redis.cluster.nodes}")
    private String clusterNodes;

    @Bean
    public RedisConnectionFactory redisConnectionFactory() {
        // 解析节点地址
        String[] nodes = clusterNodes.split(",");
        List<RedisNode> cluster = new ArrayList<>();
        for (String node : nodes) {
            String[] hostPort = node.split(":");
            cluster.add(new RedisNode(hostPort[0], Integer.parseInt(hostPort[1])));
        }

        // 集群配置
        RedisClusterConfiguration config = new RedisClusterConfiguration();
        config.setClusterNodes(cluster);
        config.setMaxRedirects(3); // 最大重定向次数

        // 连接池配置
        LettucePoolingClientConfiguration clientConfig = LettucePoolingClientConfiguration.builder()
                .poolConfig(genericObjectPoolConfig())
                .build();

        return new LettuceConnectionFactory(config, clientConfig);
    }

    private GenericObjectPoolConfig<?> genericObjectPoolConfig() {
        GenericObjectPoolConfig<?> poolConfig = new GenericObjectPoolConfig<>();
        poolConfig.setMaxTotal(8);
        poolConfig.setMaxIdle(4);
        poolConfig.setMinIdle(1);
        poolConfig.setMaxWait(Duration.ofMillis(2000));
        return poolConfig;
    }
}

3.3 业务层封装示例

@Service
public class ClusterCacheService {
    private final RedisTemplate<String, Object> redisTemplate;

    // 使用分布式的计数器
    public Long increment(String key, long delta) {
        return redisTemplate.opsForValue().increment(key, delta);
    }

    // 批量操作处理（注意跨槽位的处理）
    public void multiSet(Map<String, Object> map) {
        Map<RedisClusterNode, Map<byte[], byte[]>> mapping = 
            new HashMap<>();
        
        RedisClusterConnection connection = 
            (RedisClusterConnection)redisTemplate.getConnectionFactory()
                .getConnection();

        // 根据槽位分组数据
        map.forEach((k, v) -> {
            int slot = ClusterSlotHashUtil.calculateSlot(k.getBytes());
            RedisClusterNode node = connection.clusterGetNodeForSlot(slot);
            mapping.computeIfAbsent(node, n -> new HashMap<>())
                   .put(k.getBytes(), redisTemplate.getValueSerializer().serialize(v));
        });

        // 按节点分批执行
        mapping.forEach((node, entries) -> {
            connection.multi();
            entries.forEach((k, v) -> connection.set(k, v));
            connection.exec();
        });
    }
}

四、深度扩展：Pipeline与事务的集群适配

在集群环境下执行批量操作需要特别注意slot分布。这里我们实现一个自适应Pipeline操作：

public List<Object> executePipeline(RedisCallback<?> action) {
    return redisTemplate.execute(new SessionCallback<List<Object>>() {
        @Override
        public List<Object> execute(RedisOperations operations) {
            RedisConnection connection = operations.getConnectionFactory()
                .getConnection();
            
            if (connection instanceof RedisClusterConnection) {
                Map<RedisClusterNode, List<RedisCommand>> commands = 
                    ((RedisClusterConnection)connection)
                        .getCommands();
                List<Object> results = new ArrayList<>();
                
                commands.forEach((node, cmds) -> {
                    connection.execute("SELECT", node.getSlot());
                    connection.openPipeline();
                    cmds.forEach(cmd -> connection.execute(cmd));
                    results.addAll(connection.closePipeline());
                });
                
                return results;
            } else {
                connection.openPipeline();
                action.doInRedis(connection);
                return connection.closePipeline();
            }
        }
    });
}

五、技术选型与最佳实践

5.1 应用场景分析

海量数据缓存：当缓存数据超过单机内存容量时
分布式会话存储：需要高可用的会话共享场景
实时排行榜系统：高并发写入且需要持久化保证
地理位置服务：利用Geo命令实现附近的人功能

5.2 技术优劣势对比

优势：

自动分片与再平衡能力
原生高可用设计（Master-Slave自动切换）
支持线性扩展至1000+节点

挑战：

跨槽位操作需特殊处理（需使用hashtag）
Lua脚本的所有Key必须落在同一个slot
集群重新分片时存在性能波动

5.3 实战注意事项

数据冷热分离策略：通过key前缀设计将热点数据分配到独立节点
客户端连接数控制：每个Lettuce连接可维护多个底层物理连接
重试策略设计：推荐指数退避算法处理MOVED/ASK重定向
监控指标体系：重点关注cluster_slots_ok和keyspace_hits_ratio

六、总结与展望

经过实际项目验证，Redis Cluster在数据规模超过500GB、QPS超过10万的应用场景中表现优异。特别是在滚动升级过程中，通过CLUSTER FAILOVER命令实现零停机维护，极大提升了系统可用性。

未来随着Redis 7.0的多线程特性普及，单个分片的性能瓶颈将得到进一步突破。但也要注意伴随数据规模增大，CLUSTER SLOTS命令的响应时间会线性增长，建议在客户端缓存路由表信息。

敲码拾光专注于编程技术，涵盖编程语言、代码实战案例、软件开发技巧、IT前沿技术、编程开发工具，是您提升技术能力的优质网络平台。

Java操作Redis：集群搭建，Redis Cluster 集成