Redis 集群方案详解：从主从复制到 Cluster ⭐⭐⭐

/**
 * 全量同步流程
 * 
 * 1. 从节点发送 PSYNC ? -1
 *    表示：我是新从节点，请进行全量同步
 * 
 * 2. 主节点响应 +FULLRESYNC <replication_id> <offset>
 *    replication_id：主节点的唯一标识
 *    offset：主节点当前的复制偏移量
 * 
 * 3. 主节点执行 BGSAVE，生成 RDB 快照
 *    同时将新的写命令缓存到缓冲区
 * 
 * 4. 主节点将 RDB 文件发送给从节点
 *    从节点接收并保存为临时文件
 * 
 * 5. 从节点加载 RDB 文件
 *    恢复主节点的所有数据
 * 
 * 6. 主节点将缓冲区的命令发送给从节点
 *    从节点执行这些命令，完成同步
 */

/**
 * 增量同步流程
 * 
 * 1. 从节点发送 PSYNC <replication_id> <offset>
 *    表示：我之前同步过，请从 offset 位置继续同步
 * 
 * 2. 主节点检查：
 *    - replication_id 是否匹配？
 *    - offset 是否在缓冲区范围内？
 * 
 * 3. 如果都匹配，主节点响应 +CONTINUE
 *    表示：可以进行增量同步
 * 
 * 4. 主节点将缓冲区中 offset 之后的命令发送给从节点
 *    从节点执行这些命令，完成同步
 * 
 * 5. 之后，主节点的每个写命令都会实时发送给从节点
 */

/**
 * 问题：主从复制是异步的，从节点的数据可能落后于主节点
 * 
 * 场景：
 * 1. 客户端写入数据到主节点
 * 2. 主节点立即返回成功
 * 3. 主节点异步发送命令给从节点
 * 4. 客户端立即从从节点读取数据
 * 5. 从节点还没收到命令，返回旧数据
 * 
 * 解决方案：
 * - 对于强一致性要求的数据，必须从主节点读取
 * - 使用 WAIT 命令等待从节点确认
 * - 使用 Redis Sentinel 或 Cluster 提高可用性
 */

// 使用 WAIT 命令
// WAIT numreplicas timeout
// 等待至少 numreplicas 个从节点确认写操作

// 示例：等待至少 1 个从节点确认，超时 1000ms
redis.set("key", "value");
redis.wait(1, 1000);  // 返回确认的从节点数

/**
 * 问题：主节点故障，从节点升级为主节点时，可能丢失数据
 * 
 * 场景：
 * 1. 主节点收到写命令，返回成功
 * 2. 主节点还没来得及发送给从节点
 * 3. 主节点故障宕机
 * 4. 从节点升级为主节点
 * 5. 这条命令永久丢失
 * 
 * 解决方案：
 * - 使用 Redis Sentinel 的 min-slaves-to-write 配置
 * - 主节点必须至少有 N 个从节点确认，才能接受写操作
 * - 这样可以保证数据不丢失
 */

// redis.conf
min-slaves-to-write 1      # 至少 1 个从节点
min-slaves-max-lag 10      # 从节点延迟不超过 10 秒

// 如果从节点数量不足或延迟过大，主节点拒绝写操作

/**
 * Sentinel 故障检测流程
 * 
 * 1. 定期发送 PING 命令
 *    Sentinel 每秒向主从节点发送 PING 命令
 * 
 * 2. 判断节点状态
 *    - 如果收到 PONG 或 LOADING 或 MASTERDOWN 响应，认为节点在线
 *    - 如果没有收到响应，认为节点可能故障
 * 
 * 3. 主观下线（Subjective Down）
 *    单个 Sentinel 认为节点故障
 *    条件：down-after-milliseconds 时间内没有收到有效响应
 * 
 * 4. 客观下线（Objective Down）
 *    多个 Sentinel 都认为节点故障
 *    条件：至少 quorum 个 Sentinel 认为节点故障
 * 
 * 5. 触发故障转移
 *    当主节点被判定为客观下线时，触发故障转移
 */

// sentinel.conf 配置
sentinel monitor mymaster 127.0.0.1 6379 2
// 监控主节点 mymaster，地址 127.0.0.1:6379
// quorum = 2，至少 2 个 Sentinel 认为故障才转移

sentinel down-after-milliseconds mymaster 30000
// 30 秒内没有收到响应，认为主观下线

sentinel parallel-syncs mymaster 1
// 故障转移时，最多 1 个从节点同时进行全量同步

/**
 * 新主节点选择算法
 * 
 * 优先级1：replica-priority（从节点优先级）
 *   - 配置中指定，默认 100
 *   - 优先级越低越优先（0 最高）
 *   - 可以手动调整，让某个从节点优先成为主节点
 * 
 * 优先级2：复制偏移量（Replication Offset）
 *   - 从节点复制的数据越多，偏移量越大
 *   - 偏移量最大的从节点数据最新
 *   - 优先选择数据最新的从节点
 * 
 * 优先级3：运行 ID（Run ID）
 *   - Redis 实例启动时生成的唯一标识
 *   - 如果前两个条件相同，选择 ID 最小的
 */

// sentinel.conf 配置
sentinel replica-priority mymaster 100

// 调整从节点优先级
// 在从节点的 redis.conf 中配置
replica-priority 10  # 优先级 10，比默认 100 更高

/**
 * 问题：网络分区导致 Sentinel 和主节点通信中断
 * 
 * 场景：
 * 1. 主节点和 Sentinel 之间网络中断
 * 2. Sentinel 认为主节点故障，触发故障转移
 * 3. 从节点升级为新主节点
 * 4. 网络恢复，原主节点仍然接受写操作
 * 5. 现在有两个主节点，数据不一致
 * 
 * 解决方案：
 * - 使用 min-slaves-to-write 和 min-slaves-max-lag
 * - 主节点必须至少有 N 个从节点确认，才能接受写操作
 * - 如果从节点数量不足或延迟过大，主节点拒绝写操作
 * - 这样可以防止脑裂时的数据不一致
 */

// redis.conf
min-slaves-to-write 1      # 至少 1 个从节点
min-slaves-max-lag 10      # 从节点延迟不超过 10 秒

// 如果从节点数量不足或延迟过大，主节点拒绝写操作
// 这样可以防止脑裂时的数据不一致

/**
 * 问题：Sentinel 节点故障，无法进行故障转移
 * 
 * 解决方案：
 * - 部署多个 Sentinel 节点（通常 3 个或 5 个）
 * - 使用 Raft 算法选举 Leader Sentinel
 * - 只要大多数 Sentinel 正常，就能进行故障转移
 * - 推荐部署奇数个 Sentinel（3、5、7 等）
 */

// 部署 3 个 Sentinel 节点
// Sentinel 1: 127.0.0.1:26379
// Sentinel 2: 127.0.0.1:26380
// Sentinel 3: 127.0.0.1:26381

// 每个 Sentinel 的配置都相同
sentinel monitor mymaster 127.0.0.1 6379 2

/**
 * Redis Cluster 使用 16384 个槽来分片数据
 * 
 * 槽的计算方式：
 * slot = CRC16(key) % 16384
 * 
 * 其中：
 * - CRC16：循环冗余校验算法
 * - key：Redis 的键
 * - 16384：槽的总数（2^14）
 * 
 * 槽的分配：
 * - 每个主节点负责一部分槽
 * - 所有 16384 个槽都必须被分配
 * - 槽的分配可以动态调整
 */

// 计算 key 所在的槽
// 示例：key = "user:1000"
// slot = CRC16("user:1000") % 16384 = 5474

// 在 Redis 中查询 key 所在的槽
redis-cli cluster keyslot "user:1000"
// 输出：5474

/**
 * MOVED 重定向：当客户端请求的 key 不在当前节点时
 * 
 * 场景：
 * 1. 客户端连接到 Node A
 * 2. 请求 key "user:1000"
 * 3. 计算 slot = 5474
 * 4. Node A 发现 slot 5474 不在自己负责的范围内
 * 5. Node A 返回 MOVED 5474 127.0.0.1:6380
 *    表示：这个 slot 在 127.0.0.1:6380 上
 * 6. 客户端重定向到 Node B (127.0.0.1:6380)
 * 7. Node B 处理请求，返回结果
 * 
 * MOVED 表示：槽的位置已经确定，客户端应该更新路由表
 */

// 客户端收到 MOVED 响应后的处理
// 1. 更新本地的槽位映射表
// 2. 下次请求直接连接到正确的节点
// 3. 避免不必要的重定向

/**
 * ASK 重定向：当槽正在迁移时
 * 
 * 场景：
 * 1. 集群正在进行槽迁移
 * 2. slot 5474 从 Node A 迁移到 Node B
 * 3. 迁移过程中，slot 5474 既在 Node A，也在 Node B
 * 4. 客户端请求 key "user:1000"（slot 5474）
 * 5. Node A 发现 slot 5474 正在迁移
 * 6. Node A 返回 ASK 5474 127.0.0.1:6380
 *    表示：这个 slot 正在迁移，请尝试连接到新节点
 * 7. 客户端发送 ASKING 命令，然后重试请求
 * 8. Node B 处理请求，返回结果
 * 
 * ASK 表示：槽的位置临时改变，客户端应该尝试新节点
 * 与 MOVED 不同，ASK 不会更新路由表
 */

// MOVED vs ASK
// MOVED：槽的位置已经确定，更新路由表
// ASK：槽正在迁移，临时重定向，不更新路由表

/**
 * Redis Cluster 客户端路由实现
 * 
 * 1. 初始化：获取集群的槽位映射表
 *    CLUSTER SLOTS 命令返回所有槽的分配情况
 * 
 * 2. 请求处理：
 *    a. 计算 key 所在的槽
 *    b. 查询本地槽位映射表，找到对应的节点
 *    c. 连接到该节点，发送请求
 * 
 * 3. 处理重定向：
 *    a. 如果收到 MOVED 响应，更新槽位映射表
 *    b. 如果收到 ASK 响应，发送 ASKING 命令，重试请求
 * 
 * 4. 定期更新：
 *    a. 定期执行 CLUSTER SLOTS 命令
 *    b. 更新本地的槽位映射表
 *    c. 应对集群拓扑变化
 */

// 获取集群的槽位映射表
redis-cli cluster slots

// 输出示例：
// 1) 1) (integer) 0
//    2) (integer) 5460
//    3) 1) "127.0.0.1"
//       2) (integer) 6379
//    4) 1) "127.0.0.1"
//       2) (integer) 6382
// 2) 1) (integer) 5461
//    2) (integer) 10922
//    3) 1) "127.0.0.1"
//       2) (integer) 6380
//    4) 1) "127.0.0.1"
//       2) (integer) 6383
// ...

/**
 * 数据重平衡：当添加或删除节点时，重新分配槽
 * 
 * 场景1：添加新节点
 * 1. 新节点加入集群，初始没有槽
 * 2. 使用 redis-cli --cluster reshard 命令
 * 3. 从其他节点迁移一些槽到新节点
 * 4. 新节点开始处理请求
 * 
 * 场景2：删除节点
 * 1. 使用 redis-cli --cluster reshard 命令
 * 2. 将该节点的所有槽迁移到其他节点
 * 3. 节点变为空，可以安全删除
 */

// 添加新节点到集群
redis-cli --cluster add-node 127.0.0.1:6385 127.0.0.1:6379

// 重新分配槽（交互式）
redis-cli --cluster reshard 127.0.0.1:6379

// 删除节点
redis-cli --cluster del-node 127.0.0.1:6379 <node_id>

/**
 * Cluster 的故障转移机制
 * 
 * 1. 故障检测：
 *    - 每个节点定期向其他节点发送 PING 命令
 *    - 如果没有收到 PONG 响应，认为节点可能故障
 * 
 * 2. 故障确认：
 *    - 多个节点都认为某个节点故障
 *    - 该节点被标记为 FAIL 状态
 * 
 * 3. 从节点升级：
 *    - 故障主节点的从节点发起选举
 *    - 使用 Raft 算法选出新主节点
 * 
 * 4. 配置更新：
 *    - 新主节点接管故障主节点的槽
 *    - 集群配置更新，通知所有节点
 */

// 查看集群节点状态
redis-cli cluster nodes

// 输出示例：
// 07c37dfeb235213a872192d90877d0cd55635b91 127.0.0.1:6379@16379 master - 0 1234567890 1 connected 0-5460
// 0f8c9b8c8c8c8c8c8c8c8c8c8c8c8c8c8c8c8c8c 127.0.0.1:6380@16380 master - 0 1234567890 2 connected 5461-10922
// ...

// 节点状态说明：
// - connected：节点正常
// - disconnected：节点断开连接
// - fail：节点故障
// - handshake：节点握手中

/**
 * Cluster 从节点选举算法
 * 
 * 1. 选举条件：
 *    - 主节点被标记为 FAIL
 *    - 从节点的主节点是 FAIL 状态
 * 
 * 2. 选举过程：
 *    - 从节点向其他节点发送 CLUSTER FAILOVER 命令
 *    - 其他节点投票
 *    - 获得大多数投票的从节点成为新主节点
 * 
 * 3. 选举优先级：
 *    - 复制偏移量最大（数据最新）
 *    - 从节点优先级（replica-priority）
 *    - 运行 ID 最小
 */

// 手动触发故障转移（测试用）
redis-cli cluster failover

// 强制故障转移（不等待主节点确认）
redis-cli cluster failover force

// 故障转移超时配置
// cluster-node-timeout 15000  # 毫秒

/**
 * Gossip 协议：集群节点之间的通信协议
 * 
 * 特点：
 * 1. 去中心化：没有中心节点，所有节点平等
 * 2. 最终一致性：信息最终会传播到所有节点
 * 3. 容错性强：即使某些节点故障，集群仍能正常工作
 * 
 * 工作原理：
 * 1. 每个节点定期选择随机节点发送 PING 消息
 * 2. 接收节点返回 PONG 消息
 * 3. 消息中包含节点的状态信息
 * 4. 节点根据收到的信息更新自己的状态
 * 5. 信息逐步传播到所有节点
 */

// Gossip 消息的内容：
// 1. 节点 ID
// 2. 节点地址和端口
// 3. 节点状态（master/slave/fail）
// 4. 节点的槽位分配
// 5. 节点的配置版本
// 6. 其他节点的信息

import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;
import java.util.HashSet;
import java.util.Set;

public class JedisClusterExample {
    
    public static void main(String[] args) {
        // 1. 创建集群节点集合
        Set<HostAndPort> nodes = new HashSet<>();
        nodes.add(new HostAndPort("127.0.0.1", 6379));
        nodes.add(new HostAndPort("127.0.0.1", 6380));
        nodes.add(new HostAndPort("127.0.0.1", 6381));
        
        // 2. 创建 JedisCluster 实例
        JedisCluster cluster = new JedisCluster(nodes);
        
        // 3. 使用集群
        cluster.set("key1", "value1");
        String value = cluster.get("key1");
        System.out.println(value);  // 输出：value1
        
        // 4. 关闭连接
        cluster.close();
    }
}

import io.lettuce.core.RedisURI;
import io.lettuce.core.cluster.RedisClusterClient;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands;

public class LettuceClusterExample {
    
    public static void main(String[] args) {
        // 1. 创建集群 URI
        RedisURI uri1 = RedisURI.create("redis://127.0.0.1:6379");
        RedisURI uri2 = RedisURI.create("redis://127.0.0.1:6380");
        RedisURI uri3 = RedisURI.create("redis://127.0.0.1:6381");
        
        // 2. 创建集群客户端
        RedisClusterClient client = RedisClusterClient.create(
            Arrays.asList(uri1, uri2, uri3)
        );
        
        // 3. 获取连接
        StatefulRedisClusterConnection<String, String> connection = 
            client.connect();
        
        // 4. 获取同步命令接口
        RedisAdvancedClusterCommands<String, String> commands = 
            connection.sync();
        
        // 5. 使用集群
        commands.set("key1", "value1");
        String value = commands.get("key1");
        System.out.println(value);  // 输出：value1
        
        // 6. 关闭连接
        connection.close();
        client.shutdown();
    }
}

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisClusterConfiguration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

@Configuration
public class RedisClusterConfig {
    
    @Bean
    public RedisClusterConfiguration redisClusterConfiguration() {
        RedisClusterConfiguration config = new RedisClusterConfiguration();
        config.clusterNode("127.0.0.1", 6379);
        config.clusterNode("127.0.0.1", 6380);
        config.clusterNode("127.0.0.1", 6381);
        return config;
    }
    
    @Bean
    public LettuceConnectionFactory connectionFactory(
            RedisClusterConfiguration config) {
        return new LettuceConnectionFactory(config);
    }
    
    @Bean
    public RedisTemplate<String, String> redisTemplate(
            LettuceConnectionFactory factory) {
        RedisTemplate<String, String> template = new RedisTemplate<>();
        template.setConnectionFactory(factory);
        
        // 设置序列化器
        StringRedisSerializer serializer = new StringRedisSerializer();
        template.setKeySerializer(serializer);
        template.setValueSerializer(serializer);
        
        return template;
    }
}

// 使用示例
@Service
public class RedisService {
    
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    
    public void set(String key, String value) {
        redisTemplate.opsForValue().set(key, value);
    }
    
    public String get(String key) {
        return redisTemplate.opsForValue().get(key);
    }
}

/**
 * 问题1：写操作丢失
 * 
 * 场景：
 * T1: 客户端写入数据到主节点
 * T2: 主节点返回成功
 * T3: 主节点还没来得及发送给从节点
 * T4: 主节点故障
 * T5: 从节点升级为主节点
 * T6: 这条写操作永久丢失
 * 
 * 解决方案：
 * - 使用 min-slaves-to-write 和 min-slaves-max-lag
 * - 主节点必须至少有 N 个从节点确认，才能接受写操作
 * - 如果从节点数量不足或延迟过大，主节点拒绝写操作
 */

// redis.conf
min-slaves-to-write 1      # 至少 1 个从节点
min-slaves-max-lag 10      # 从节点延迟不超过 10 秒

// 如果从节点数量不足或延迟过大，主节点拒绝写操作
// 这样可以保证数据不丢失

/**
 * 问题2：读操作返回旧数据
 * 
 * 场景：
 * T1: 客户端从从节点读取数据
 * T2: 从节点返回旧数据（还没同步最新数据）
 * T3: 主节点故障，从节点升级为主节点
 * T4: 客户端读取的是旧数据
 * 
 * 解决方案：
 * - 对于强一致性要求的数据，必须从主节点读取
 * - 使用 WAIT 命令等待从节点确认
 * - 接受最终一致性，允许短期数据不一致
 */

// 方案1：从主节点读取
String value = redisTemplate.opsForValue().get(key);

// 方案2：使用 WAIT 命令
redis.set("key", "value");
redis.wait(1, 1000);  // 等待至少 1 个从节点确认，超时 1000ms

// 方案3：接受最终一致性
// 从节点可能返回旧数据，但最终会一致

/**
 * 问题3：脑裂导致数据不一致
 * 
 * 场景：
 * T1: 主节点和 Sentinel 之间网络中断
 * T2: Sentinel 认为主节点故障，触发故障转移
 * T3: 从节点升级为新主节点
 * T4: 网络恢复，原主节点仍然接受写操作
 * T5: 现在有两个主节点，数据不一致
 * 
 * 解决方案：
 * - 使用 min-slaves-to-write 和 min-slaves-max-lag
 * - 主节点必须至少有 N 个从节点确认，才能接受写操作
 * - 如果从节点数量不足或延迟过大，主节点拒绝写操作
 * - 这样可以防止脑裂时的数据不一致
 */

// redis.conf
min-slaves-to-write 1      # 至少 1 个从节点
min-slaves-max-lag 10      # 从节点延迟不超过 10 秒

// 脑裂时，原主节点无法连接到从节点
// 因此拒绝写操作，防止数据不一致

/**
 * 数据冗余：每个主节点都有从节点
 * 
 * 优势：
 * 1. 主节点故障，从节点可以升级为主节点
 * 2. 数据不会丢失
 * 3. 服务不会中断
 * 
 * 配置：
 * - 每个主节点至少有 1 个从节点
 * - 可以配置多个从节点，提高可用性
 */

// 查看集群节点信息
redis-cli cluster nodes

// 输出示例：
// 07c37dfeb235213a872192d90877d0cd55635b91 127.0.0.1:6379@16379 master - 0 1234567890 1 connected 0-5460
// 0f8c9b8c8c8c8c8c8c8c8c8c8c8c8c8c8c8c8c8c 127.0.0.1:6382@16382 slave 07c37dfeb235213a872192d90877d0cd55635b91 0 1234567890 1 connected

/**
 * 故障检测和转移：
 * 
 * 1. 故障检测：
 *    - 每个节点定期向其他节点发送 PING 命令
 *    - 如果没有收到 PONG 响应，认为节点可能故障
 * 
 * 2. 故障确认：
 *    - 多个节点都认为某个节点故障
 *    - 该节点被标记为 FAIL 状态
 * 
 * 3. 从节点升级：
 *    - 故障主节点的从节点发起选举
 *    - 使用 Raft 算法选出新主节点
 * 
 * 4. 配置更新：
 *    - 新主节点接管故障主节点的槽
 *    - 集群配置更新，通知所有节点
 */

// 配置参数
// cluster-node-timeout 15000  # 节点超时时间（毫秒）
// cluster-failover-timeout 180000  # 故障转移超时时间（毫秒）

/**
 * 槽的完整性：所有 16384 个槽都必须被分配
 * 
 * 如果某个槽没有被分配，集群会进入 FAIL 状态
 * 所有请求都会被拒绝
 * 
 * 场景：
 * 1. 节点故障，该节点的槽无法访问
 * 2. 槽迁移过程中，槽暂时无法访问
 * 3. 集群配置错误，某个槽没有被分配
 * 
 * 解决方案：
 * - 确保每个主节点都有从节点
 * - 故障转移时，从节点升级为主节点
 * - 槽迁移完成后，更新集群配置
 */

// 查看集群状态
redis-cli cluster info

// 输出示例：
// cluster_state:ok  # 集群状态正常
// cluster_slots_assigned:16384  # 已分配的槽数
// cluster_slots_ok:16384  # 正常的槽数
// cluster_slots_pfail:0  # 可能故障的槽数
// cluster_slots_fail:0  # 故障的槽数

import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisPoolConfig;
import java.util.HashSet;
import java.util.Set;

public class JedisClusterExample {
    
    public static void main(String[] args) {
        // 1. 配置连接池
        JedisPoolConfig config = new JedisPoolConfig();
        config.setMaxTotal(100);  // 最大连接数
        config.setMaxIdle(50);    // 最大空闲连接数
        config.setMinIdle(10);    // 最小空闲连接数
        config.setTestOnBorrow(true);  // 获取连接时测试
        
        // 2. 创建集群节点集合
        Set<HostAndPort> nodes = new HashSet<>();
        nodes.add(new HostAndPort("127.0.0.1", 6379));
        nodes.add(new HostAndPort("127.0.0.1", 6380));
        nodes.add(new HostAndPort("127.0.0.1", 6381));
        
        // 3. 创建 JedisCluster 实例
        JedisCluster cluster = new JedisCluster(nodes, config);
        
        // 4. 使用集群
        cluster.set("key1", "value1");
        String value = cluster.get("key1");
        System.out.println(value);  // 输出：value1
        
        // 5. 关闭连接
        cluster.close();
    }
}