Redis 集群实战

SunnyFan大约 15 分钟约 4497 字

Redis 集群实战

简介

Redis 集群是 Redis 官方提供的分布式解决方案，支持数据自动分片、高可用和水平扩展。在生产环境中，单机 Redis 无法满足大规模并发和数据存储需求，通过集群模式可以有效解决单点瓶颈和数据容量限制。本文将深入讲解 Redis Cluster 模式、哨兵模式、槽位分配机制以及故障转移的配置与实践。

特点

自动数据分片，支持 16384 个哈希槽分布到多个节点
内置高可用机制，支持自动故障检测和故障转移
支持在线扩容和缩容，无需停止服务
去中心化架构，所有节点相互通信，无单点故障

Cluster 模式配置

Redis Cluster 采用去中心化的 Gossip 协议进行节点间通信，将数据划分为 16384 个槽位分布在多个节点上。

节点配置

# redis-7001.conf 节点1配置
port 7001
bind 0.0.0.0
daemonize yes
cluster-enabled yes
cluster-config-file nodes-7001.conf
cluster-node-timeout 15000
cluster-announce-ip 192.168.1.100
cluster-announce-port 7001
cluster-announce-bus-port 17001
appendonly yes
appendfsync everysec
save 900 1
save 300 10
save 60 10000
maxmemory 4gb
maxmemory-policy allkeys-lru

创建集群

# 启动所有节点（7001-7006，3主3从）
for port in 7001 7002 7003 7004 7005 7006; do
    redis-server redis-${port}.conf
done

# 创建集群（Redis 5+ 使用 redis-cli）
redis-cli --cluster create \
    192.168.1.100:7001 \
    192.168.1.100:7002 \
    192.168.1.101:7003 \
    192.168.1.101:7004 \
    192.168.1.102:7005 \
    192.168.1.102:7006 \
    --cluster-replicas 1

# 检查集群状态
redis-cli -p 7001 cluster info
redis-cli -p 7001 cluster nodes

Java 连接集群示例

import redis.clients.jedis.*;
import java.util.HashSet;
import java.util.Set;

public class RedisClusterDemo {
    public static void main(String[] args) {
        Set<HostAndPort> nodes = new HashSet<>();
        nodes.add(new HostAndPort("192.168.1.100", 7001));
        nodes.add(new HostAndPort("192.168.1.100", 7002));
        nodes.add(new HostAndPort("192.168.1.101", 7003));
        nodes.add(new HostAndPort("192.168.1.101", 7004));
        nodes.add(new HostAndPort("192.168.1.102", 7005));
        nodes.add(new HostAndPort("192.168.1.102", 7006));

        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(200);
        poolConfig.setMaxIdle(50);
        poolConfig.setMinIdle(10);
        poolConfig.setMaxWaitMillis(3000);

        try (JedisCluster jedisCluster = new JedisCluster(nodes, 3000, 3000, 5, poolConfig)) {
            // 写入数据
            jedisCluster.set("user:1001:name", "张三");
            jedisCluster.set("user:1001:age", "28");
            jedisCluster.hset("user:1001", "email", "zhangsan@example.com");
            jedisCluster.hset("user:1001", "phone", "13800138000");

            // 读取数据
            String name = jedisCluster.get("user:1001:name");
            System.out.println("用户名: " + name);

            // 使用 Pipeline 批量操作（同一槽位）
            jedisCluster.pipelined();
        }
    }
}

哨兵模式

Redis Sentinel 是官方推荐的高可用方案，用于监控 Redis 主从复制架构，并在主节点故障时自动进行故障转移。

Sentinel 配置

# sentinel-26379.conf
port 26379
daemonize yes
sentinel monitor mymaster 192.168.1.100 6379 2
sentinel down-after-milliseconds mymaster 10000
sentinel failover-timeout mymaster 60000
sentinel parallel-syncs mymaster 1
sentinel auth-pass mymaster YourStrongPassword

# 多个 Sentinel 节点相互发现
sentinel known-sentinel mymaster 192.168.1.101 26379
sentinel known-sentinel mymaster 192.168.1.102 26379

Sentinel 管理命令

# 启动 Sentinel
redis-sentinel sentinel-26379.conf

# 查看主节点状态
redis-cli -p 26379 sentinel master mymaster

# 查看从节点列表
redis-cli -p 26379 sentinel slaves mymaster

# 查看 Sentinel 节点列表
redis-cli -p 26379 sentinel sentinels mymaster

# 获取当前主节点地址
redis-cli -p 26379 sentinel get-master-addr-by-name mymaster

# 手动故障转移
redis-cli -p 26379 sentinel failover mymaster

# 重置指定主节点
redis-cli -p 26379 sentinel reset mymaster

Sentinel 关键参数说明

参数	默认值	说明
`down-after-milliseconds`	30000	判定主观下线的超时时间
`failover-timeout`	180000	故障转移超时时间
`parallel-syncs`	1	故障转移后同时向新主节点发起复制的从节点数
`quorum`	2	判定客观下线所需的 Sentinel 同意数

槽位分配与管理

Redis Cluster 将所有数据映射到 16384 个哈希槽中，每个主节点负责一部分槽位。

槽位计算与分配

# 查看键对应的槽位
redis-cli -p 7001 cluster keyslot "user:1001"
# 输出: 12569

# 查看槽位范围分布
redis-cli -p 7001 cluster slots
# 输出: 每个节点负责的槽位范围

# 查看指定槽位中的键数量
redis-cli -p 7001 cluster countkeysinslot 12569

# 获取指定槽位中的键
redis-cli -p 7001 cluster getkeysinslot 12569 10

重新分配槽位（扩容）

# 添加新节点到集群
redis-cli --cluster add-node 192.168.1.103:7007 192.168.1.100:7001

# 重新分配槽位
redis-cli --cluster reshard 192.168.1.100:7001 \
    --cluster-from all \
    --cluster-to <new-node-id> \
    --cluster-slots 2048 \
    --cluster-yes

# 查看重新分配后的状态
redis-cli --cluster check 192.168.1.100:7001

# 为新节点添加从节点
redis-cli --cluster add-node \
    192.168.1.103:7008 192.168.1.100:7001 \
    --cluster-slave --cluster-master-id <new-node-id>

使用 Hash Tag 优化多键操作

# 使用 Hash Tag 确保相关键在同一槽位
SET user:{1001}:name "张三"
SET user:{1001}:age "28"
SET user:{1001}:email "zhangsan@example.com"

# 这样可以在同一槽位执行多键操作
MGET user:{1001}:name user:{1001}:age user:{1001}:email

故障转移

Redis Cluster 内置了完善的故障检测和自动转移机制，确保在节点故障时服务仍然可用。

故障检测与转移流程

# 模拟主节点故障（停止节点）
redis-cli -p 7001 DEBUG SLEEP 30

# 监控集群状态变化
watch -n 1 "redis-cli -p 7002 cluster nodes"

# 手动触发故障转移（在从节点上执行）
redis-cli -p 7004 cluster failover

# 强制故障转移（不等待主节点确认）
redis-cli -p 7004 cluster failover takeover

故障转移监控脚本

import redis
import time
import smtplib
from datetime import datetime

class RedisClusterMonitor:
    def __init__(self, nodes):
        self.nodes = nodes
        self.alert_recipients = ["admin@example.com"]

    def check_cluster_health(self):
        """检查集群健康状态"""
        for host, port in self.nodes:
            try:
                client = redis.Redis(host=host, port=port, decode_responses=True)
                info = client.cluster("info")

                cluster_state = info.get("cluster_state")
                slots_assigned = info.get("cluster_slots_assigned")
                slots_ok = info.get("cluster_slots_ok")

                if cluster_state != "ok":
                    self.send_alert(
                        f"Redis 集群异常",
                        f"节点 {host}:{port} 报告集群状态为 {cluster_state}\n"
                        f"时间: {datetime.now()}"
                    )

                if slots_assigned != slots_ok:
                    self.send_alert(
                        f"Redis 槽位异常",
                        f"节点 {host}:{port} 槽位不完整\n"
                        f"已分配: {slots_assigned}, 正常: {slots_ok}"
                    )

                # 检查各节点连接状态
                nodes_info = client.cluster("nodes")
                for line in nodes_info.split("\n"):
                    if "fail" in line and "myself" not in line:
                        self.send_alert(
                            f"Redis 节点故障",
                            f"发现故障节点: {line[:100]}"
                        )

                client.close()

            except redis.ConnectionError as e:
                self.send_alert(
                    f"Redis 连接失败",
                    f"无法连接到 {host}:{port}\n错误: {str(e)}"
                )

    def send_alert(self, subject, message):
        """发送告警通知"""
        print(f"[告警] {subject}: {message}")

    def run(self, interval=10):
        """持续监控"""
        while True:
            self.check_cluster_health()
            time.sleep(interval)

if __name__ == "__main__":
    monitor = RedisClusterMonitor([
        ("192.168.1.100", 7001),
        ("192.168.1.100", 7002),
        ("192.168.1.101", 7003),
    ])
    monitor.run()

Cluster 与 Sentinel 模式对比

特性	Cluster 模式	Sentinel 模式
数据分片	原生支持	不支持
最大数据量	多节点总和	单机内存限制
最小节点数	6（3主3从）	3 Sentinel + 1主1从
多键操作	仅限同槽位	无限制
扩容	支持在线扩容	需手动调整
运维复杂度	较高	较低
适用场景	大数据量、高并发	中小规模高可用

优点

原生支持数据分片，可水平扩展存储容量和吞吐量
自动故障转移机制成熟，无需外部依赖即可实现高可用
去中心化架构避免单点故障，整体可靠性高
支持在线扩缩容，业务无感知完成节点变更

Hash Tag 机制详解

跨槽位操作解决方案

# Hash Tag 机制：使用 {} 包裹的部分参与 CRC16 计算
# 同一个 Hash Tag 下的 Key 会被分配到同一个槽位

# 用户相关数据放到同一个槽位
SET user:1001:profile '{"name":"张三","age":25}'
SET user:1001:settings '{"theme":"dark","lang":"zh"}'
SET user:1001:orders '{"order_count":5,"last_order":"2026-04-14"}'

# 这三个 Key 中的 1001 被包含在 {} 中
# 因此它们会被分配到同一个槽位（同一个节点）
# 可以使用 MGET、MSET、Pipeline 等批量操作

# 批量操作
MGET user:1001:profile user:1001:settings user:1001:orders

# 跨槽位操作（不支持）
# MSET user:1001:profile "data1" user:2002:profile "data2"
# 错误：CROSSSLOT Keys in request don't hash to the same slot

# 正确做法：使用 Hash Tag 或分别操作
MSET user:{1001}:profile "data1" user:{1001}:settings "data2"
# 或者
SET user:1001:profile "data1"
SET user:2002:profile "data2"

# Hash Tag 使用场景：
# 1. 用户维度数据聚合
# 2. 订单相关数据（订单 + 明细）
# 3. 会话数据（session + preferences）
# 4. 需要事务操作的多个 Key

# 注意事项：
# 1. 不要滥用 Hash Tag，可能导致数据倾斜
# 2. Hash Tag 部分应该是高分散的（如用户 ID）
# 3. 不要用固定值作为 Hash Tag（如 {global}）

集群数据迁移

从单机迁移到集群

# 方案一：使用 redis-cli --cluster import
redis-cli --cluster import 192.168.1.100:7001 \
  --cluster-from 192.168.1.50:6379 \
  --cluster-copy \
  --cluster-replace

# --cluster-copy — 保留源数据（不删除）
# --cluster-replace — 覆盖已存在的 Key

# 方案二：使用 redis-shake（阿里开源）
# 配置文件 redis-shake.conf
source.type = standalone
source.address = 192.168.1.50:6379
target.type = cluster
target.address = 192.168.1.100:7001
target.password = target_pwd

# 启动迁移
./redis-shake -conf redis-shake.conf

# 方案三：双写方案（应用层）
# 1. 应用同时写入单机和集群
# 2. 数据校验确认一致后，切换读流量到集群
# 3. 确认无问题后，停止写单机

# 数据校验
redis-cli --cluster check 192.168.1.100:7001
# 检查所有节点状态、槽位分配、数据一致性

# 全量数据校验
# 使用 redis-full-check（阿里开源）
./redis-full-check -s 192.168.1.50:6379 -t 192.168.1.100:7001

集群扩容实战

# 场景：从 3 主 3 从 扩容到 5 主 5 从

# 步骤 1：启动新节点
redis-server /etc/redis/redis-7007.conf
redis-server /etc/redis/redis-7008.conf
redis-server /etc/redis/redis-7009.conf
redis-server /etc/redis/redis-7010.conf

# 步骤 2：添加新节点到集群
redis-cli --cluster add-node 192.168.1.102:7007 192.168.1.100:7001
redis-cli --cluster add-node 192.168.1.102:7008 192.168.1.100:7002
redis-cli --cluster add-node 192.168.1.102:7009 192.168.1.101:7003
redis-cli --cluster add-node 192.168.1.102:7010 192.168.1.101:7004

# 步骤 3：重新分配槽位
redis-cli --cluster reshard 192.168.1.100:7001
# 提示输入：
# How many slots do you want to move? 4096  (16384 / 5 ≈ 3277，每个主节点分出约 655 个槽位)
# What is the receiving node ID? <新节点 ID>
# Source node #1: all  (从所有现有主节点迁移)

# 步骤 4：为新主节点添加从节点
redis-cli --cluster add-node \
  --cluster-slave \
  --cluster-master-id <新主节点ID> \
  192.168.1.102:7010 \
  192.168.1.100:7001

# 步骤 5：验证集群状态
redis-cli --cluster info 192.168.1.100:7001
redis-cli --cluster check 192.168.1.100:7001

# 注意事项：
# 1. 迁移过程中集群可用，但有短暂延迟
# 2. 大量数据迁移时建议在低峰期进行
# 3. 迁移完成后观察各节点内存使用是否均衡

.NET 集群集成

StackExchange.Redis 集群配置

// 基本集群连接
var config = new ConfigurationOptions
{
    EndPoints = {
        { "192.168.1.100", 7001 },
        { "192.168.1.100", 7002 },
        { "192.168.1.101", 7003 },
    },
    // 集群配置
    AbortOnConnectFail = false,
    ConnectTimeout = 5000,
    SyncTimeout = 5000,
    AsyncTimeout = 5000,
    AllowAdmin = true,
    // 连接池
    // Redis 6+ 支持 RESP3 协议
    // DefaultVersion = new Version(6, 0),
};

var connection = ConnectionMultiplexer.Connect(config);
var db = connection.GetDatabase();

// 批量操作（集群模式）
// 方案一：使用 Hash Tag
var batch = db.CreateBatch();
batch.StringSetAsync("user:{1001}:name", "张三");
batch.StringSetAsync("user:{1001}:age", "25");
batch.Execute();

// 方案二：使用 Lua 脚本（保证原子性）
var script = @"
    redis.call('SET', KEYS[1], ARGV[1])
    redis.call('SET', KEYS[2], ARGV[2])
    redis.call('SET', KEYS[3], ARGV[3])
    return 'OK'";
await db.ScriptEvaluateAsync(script,
    new RedisKey[] { "user:{1001}:name", "user:{1001}:age", "user:{1001}:email" },
    new RedisValue[] { "张三", "25", "zhangsan@example.com" });

// 集群信息查询
var endpoints = connection.GetEndPoints(true);
foreach (var endpoint in endpoints)
{
    var server = connection.GetServer(endpoint);
    if (server.IsConnected)
    {
        var clusterInfo = server.ClusterInfo();
        var nodes = server.ClusterNodes();
        Console.WriteLine($"节点 {endpoint}: {clusterInfo}");
    }
}

集群模式下的缓存策略

/// <summary>
/// 集群模式下的分布式缓存服务
/// </summary>
public class ClusterCacheService
{
    private readonly IDatabase _redis;

    public ClusterCacheService(IConnectionMultiplexer redis)
    {
        _redis = redis.GetDatabase();
    }

    // 使用 Hash Tag 的缓存 Key 设计
    // 格式: cache:{业务模块}:{实体ID}:{字段}
    private static string CacheKey(string module, string entityId, string field)
        => $"cache:{{{module}}}:{entityId}:{field}";

    // 设置缓存（带 Hash Tag）
    public async Task SetAsync(string module, string entityId, Dictionary<string, string> fields, TimeSpan? expiry = null)
    {
        var batch = _redis.CreateBatch();
        foreach (var kv in fields)
        {
            var key = CacheKey(module, entityId, kv.Key);
            _ = batch.StringSetAsync(key, kv.Value, expiry);
        }
        batch.Execute();
    }

    // 批量获取（同模块同实体的字段，保证在同一槽位）
    public async Task<Dictionary<string, string?>> GetAsync(string module, string entityId, string[] fields)
    {
        var keys = fields.Select(f => (RedisKey)CacheKey(module, entityId, f)).ToArray();
        var values = await _redis.StringGetAsync(keys);
        return fields.Zip(values, (f, v) => new { Field = f, Value = v })
            .ToDictionary(x => x.Field, x => x.Value.HasValue ? x.Value.ToString() : null);
    }

    // 分布式锁（集群模式）
    public async Task<IDisposable?> AcquireLockAsync(string resource, TimeSpan expiry)
    {
        var lockKey = $"lock:{{{resource}}}";
        var lockValue = Guid.NewGuid().ToString();
        var acquired = await _redis.StringSetAsync(lockKey, lockValue, expiry, When.NotExists);

        if (!acquired) return null;

        return new RedisLockReleaser(_redis, lockKey, lockValue);
    }

    private class RedisLockReleaser : IDisposable
    {
        private readonly IDatabase _redis;
        private readonly string _lockKey;
        private readonly string _lockValue;

        public RedisLockReleaser(IDatabase redis, string lockKey, string lockValue)
        {
            _redis = redis;
            _lockKey = lockKey;
            _lockValue = lockValue;
        }

        public void Dispose()
        {
            var script = @"
                if redis.call('get', KEYS[1]) == ARGV[1] then
                    return redis.call('del', KEYS[1])
                else return 0 end";
            _redis.ScriptEvaluate(script,
                new RedisKey[] { _lockKey },
                new RedisValue[] { _lockValue });
        }
    }
}

故障恢复与数据安全

节点故障处理

# 主节点故障（自动故障转移）
# 1. 从节点检测到主节点失联
# 2. 从节点发起故障转移
# 3. 从节点升级为主节点
# 4. 其他从节点重新指向新主节点
# 5. 集群恢复可用

# 查看故障转移日志
redis-cli -p 7001 CLUSTER INFO
# cluster_state:ok
# cluster_stats_messages_sent:xxx
# cluster_stats_messages_received:xxx

# 手动故障转移（维护时使用）
redis-cli -p 7002 CLUSTER FAILOVER
# 将从节点 7002 提升为主节点（主节点 7001 降级）

# 强制故障转移（主节点确实宕机无法通信时）
redis-cli -p 7002 CLUSTER FAILOVER FORCE

# 节点恢复后重新加入集群
redis-cli --cluster add-node 192.168.1.100:7001 192.168.1.102:7007
redis-cli --cluster replicate <主节点ID> 192.168.1.100:7001

备份与恢复

# 集群备份（每个节点都需要备份）
for port in 7001 7002 7003 7004 7005 7006; do
    redis-cli -p $port BGSAVE
done

# 等待 RDB 文件生成
for port in 7001 7002 7003 7004 7005 7006; do
    until redis-cli -p $port LASTSAVE | grep -q $(redis-cli -p $port LASTSAVE); do
        sleep 1
    done
    cp /var/lib/redis/dump.rdb /backup/dump_${port}_$(date +%Y%m%d).rdb
done

# 集群恢复
# 1. 停止所有节点
# 2. 将备份的 dump.rdb 复制回各节点目录
# 3. 按顺序启动节点
# 4. 验证集群状态

# AOF 备份
for port in 7001 7002 7003 7004 7005 7006; do
    cp /var/lib/redis/appendonly.aof /backup/aof_${port}_$(date +%Y%m%d).aof
done

# 注意事项：
# 1. 集群模式下每个节点只保存部分数据
# 2. 恢复时必须保证所有节点的备份是同一时间点的
# 3. BGSAVE 期间可能影响性能，建议低峰期执行

性能调优

集群性能优化

# 1. 网络优化
# 节点间通信使用内网（低延迟）
# bind 0.0.0.0
# cluster-announce-ip 192.168.1.100

# 2. 内存优化
maxmemory 4gb
maxmemory-policy allkeys-lru
maxmemory-samples 10

# 3. 持久化优化
# 纯缓存场景：关闭持久化
save ""
appendonly no

# 需要持久化：使用 AOF everysec
appendonly yes
appendfsync everysec
auto-aof-rewrite-percentage 100
auto-aof-rewrite-min-size 64mb

# 4. 连接优化
tcp-backlog 511
timeout 300
tcp-keepalive 60

# 5. 集群通信优化
# 节点超时时间（超过此时间认为节点故障）
cluster-node-timeout 15000  # 15 秒

# 6. Pipeline 批量操作
# 集群模式下 Pipeline 只能操作同一槽位的 Key
# 使用 Hash Tag 确保批量 Key 在同一槽位

# 性能基准测试
redis-benchmark -h 192.168.1.100 -p 7001 -c 100 -n 100000 -t set,get
redis-benchmark --cluster -h 192.168.1.100 -p 7001 -c 100 -n 100000

# 集群性能监控
redis-cli --cluster info 192.168.1.100:7001
# 查看各节点内存使用、Key 数量、连接数等

缺点

多键操作受限于同一槽位，跨槽位事务和聚合操作实现复杂
客户端需要支持集群协议，部分老旧客户端不兼容
批量操作（如 KEYS、FLUSHALL）在集群模式下效率较低
运维门槛较高，槽位迁移和故障排查需要深入理解集群机制

Redis 集群方案的选择取决于业务规模和需求复杂度。对于数据量大、需要水平扩展的场景，Cluster 模式是最佳选择，它提供了完善的数据分片和高可用机制；对于数据量适中但需要高可用的场景，Sentinel 模式更简单易维护。在实际部署中，建议使用至少 3 主 3 从的 Cluster 架构或 3 个 Sentinel 节点的哨兵架构，并配合完善的监控告警体系。同时需要注意合理设计键名以利用 Hash Tag 机制，避免跨槽位操作带来的性能损耗。