Redis Cluster槽位迁移原理与脑裂故障恢复实战

小编 • • 数据库

Redis Cluster采用一致性哈希槽（Hash Slot）机制将数据分布在多个节点上，16384个槽位分配到集群中的Master节点。当集群扩缩容或节点故障时，槽位迁移过程涉及数据在节点间的转移。数据库运维中，Redis Cluster的脑裂（Split-Brain）问题可能导致数据丢失，理解槽位迁移机制和故障恢复流程对于构建高可用Redis集群至关重要。

Redis Cluster槽位分配与一致性哈希原理

Redis Cluster没有使用一致性哈希环，而是使用固定16384个槽位的哈希分片。每个key通过CRC16计算后对16384取模确定所属槽位，每个Master节点负责一部分槽位范围。

# Redis Cluster初始化（6节点: 3主3从）
redis-cli --cluster create \
  192.168.1.101:7000 192.168.1.102:7000 192.168.1.103:7000 \
  192.168.1.101:7001 192.168.1.102:7001 192.168.1.103:7001 \
  --cluster-replicas 1

# 查看集群信息
redis-cli -c -p 7000 cluster nodes
# 输出:
# 7a1b... 192.168.1.101:7000@17000 master - 0 ... 1 connected 0-5460
# 8c2d... 192.168.1.102:7000@17000 master - 0 ... 2 connected 5461-10922
# 9e3f... 192.168.1.103:7000@17000 master - 0 ... 3 connected 10923-16383
# a4b5... 192.168.1.101:7001@17001 slave 7a1b... 0 ... 4 connected
# b6c7... 192.168.1.102:7001@17001 slave 8c2d... 0 ... 5 connected
# c8d9... 192.168.1.103:7001@17001 slave 9e3f... 0 ... 6 connected

# 查看槽位分配
redis-cli -c -p 7000 cluster slots
# 1) 1) (integer) 0
#    2) (integer) 5460
#    3) 1) "192.168.1.101"
#       2) (integer) 7000
#       3) "7a1b..."
#    4) 1) "192.168.1.101"
#       2) (integer) 7001
#       3) "a4b5..."

# 验证key的槽位
redis-cli -c -p 7000 cluster keyslot user:1001
# (integer) 5473

# hash tag机制
redis-cli -c -p 7000 cluster keyslot {user}:profile
# (integer) 8134  # 与{user}:settings相同槽位

hash tag机制让大括号内的内容决定槽位：{user}:profile、{user}:settings、{user}:avatar会分配到同一槽位，可以在同一节点执行MGET和事务操作。

集群扩容槽位迁移流程与在线数据迁移

向集群添加新节点时，需要从现有Master节点迁移部分槽位到新节点。迁移过程对业务在线进行，通过ASK/MOVED重定向保证客户端正常访问。

# 1. 添加新Master节点
redis-cli --cluster add-node 192.168.1.104:7000 192.168.1.101:7000

# 2. 为新Master添加Slave
redis-cli --cluster add-node 192.168.1.104:7001 192.168.1.101:7000 \
  --cluster-slave --cluster-master-id 

# 3. 迁移槽位（从三个Master各迁移一部分到新Master）
redis-cli --cluster reshard 192.168.1.101:7000 \
  --cluster-from 7a1b...,8c2d...,9e3f... \
  --cluster-to  \
  --cluster-slots 4096 \
  --cluster-yes

# 4. 手动单槽迁移过程（了解底层机制）
# 步骤1: 在源节点标记槽位为MIGRATING
redis-cli -p 7000 cluster setslot 4096 migrating 

# 步骤2: 在目标节点标记槽位为IMPORTING
redis-cli -p 7004 cluster setslot 4096 importing 

# 步骤3: 迁移槽中的key
redis-cli -p 7000 cluster getkeysinslot 4096 100
# 对每个key执行MIGRATE
redis-cli -p 7000 migrate 192.168.1.104 7000 "" 0 5000 KEYS key1 key2

# 步骤4: 迁移完成后通知所有节点
redis-cli -p 7000 cluster setslot 4096 node 
redis-cli -p 7001 cluster setslot 4096 node 
# ... 对集群中每个节点执行

# 5. 验证迁移结果
redis-cli -c -p 7000 cluster nodes

Python客户端处理槽位迁移时的ASK重定向：

import redis
from redis.cluster import RedisCluster, ClusterNode

startup_nodes = [
    ClusterNode("192.168.1.101", 7000),
    ClusterNode("192.168.1.102", 7000),
    ClusterNode("192.168.1.103", 7000),
]

rc = RedisCluster(startup_nodes=startup_nodes, decode_responses=True)

# 客户端自动处理MOVED重定向
rc.set("user:1001", "Alice")
print(rc.get("user:1001"))

# 原始协议层面的重定向处理（理解原理）
def smart_get(node_client, key):
    try:
        return node_client.get(key)
    except redis.exceptions.ResponseError as e:
        error_msg = str(e)
        if error_msg.startswith("MOVED"):
            _, slot, host_port = error_msg.split()
            host, port = host_port.split(":")
            new_client = redis.Redis(host=host, port=int(port))
            return new_client.get(key)
        elif error_msg.startswith("ASK"):
            _, slot, host_port = error_msg.split()
            host, port = host_port.split(":")
            redirect_client = redis.Redis(host=host, port=int(port))
            redirect_client.execute_command("ASKING")
            return redirect_client.get(key)
        raise

脑裂故障场景分析与min-replicas配置

Redis Cluster的脑裂发生在网络分区时，Master与Slave分属不同分区。如果Master所在分区与少数节点保持连接，Master继续接受写入，而另一个分区的Slave被提升为新Master，网络恢复后旧Master降为Slave并丢弃分区期间的写入数据。

# min-replicas配置防止脑裂数据丢失
# 在redis.conf中配置

# Master至少需要N个Slave确认写入才能接受新写入
min-replicas-to-write 1
min-replicas-max-lag 10

# 集群超时判定（节点失联多久触发故障转移）
cluster-node-timeout 15000

# 故障转移相关配置
cluster-require-full-coverage yes
cluster-allow-reads-when-down no

# 模拟脑裂场景
# 步骤1: 隔离Master的网络
iptables -A OUTPUT -d 192.168.1.101 -j DROP
iptables -A INPUT -s 192.168.1.101 -j DROP

# 步骤2: 等待cluster-node-timeout（15秒）
# 集群判定Master A下线，触发故障转移
# Slave B被提升为新Master

# 步骤3: 在被隔离的旧Master A上尝试写入
redis-cli -p 7000 set test_key "written_during_split"
# 如果配置了min-replicas-to-write, 写入会被拒绝:
# (error) NOREPLICAS Not enough good replicas to write
# 如果未配置min-replicas, 写入会成功（脑裂数据）

# 步骤4: 恢复网络
iptables -D OUTPUT -d 192.168.1.101 -j DROP
iptables -D INPUT -d 192.168.1.101 -j DROP

# 步骤5: 旧Master A重新加入集群，降级为Slave
# 旧Master A在分区期间接受的写入数据被新Master B的数据覆盖

集群故障检测与自动故障转移恢复

#!/bin/bash
# cluster_health_check.sh - 定时检测集群健康

CLUSTER_OK=0
CLUSTER_FAIL=0

for port in 7000 7001 7002 7003 7004 7005; do
    info=$(redis-cli -p $port cluster info 2>/dev/null)
    state=$(echo "$info" | grep "cluster_state:" | cut -d: -f2 | tr -d '\r')
    slots=$(echo "$info" | grep "cluster_slots_ok:" | cut -d: -f2 | tr -d '\r')
    size=$(echo "$info" | grep "cluster_known_nodes:" | cut -d: -f2 | tr -d '\r')
    
    if [ "$state" = "ok" ] && [ "$slots" = "16384" ]; then
        echo "[$port] OK - State:$state Slots:$slots Nodes:$size"
        ((CLUSTER_OK++))
    else
        echo "[$port] FAIL - State:$state Slots:$slots"
        ((CLUSTER_FAIL++))
    fi
done

if [ $CLUSTER_FAIL -gt 0 ]; then
    echo "WARNING: $CLUSTER_FAIL nodes report cluster issues"
fi

# 手动故障转移（在Slave上执行）
redis-cli -p 7001 cluster failover

# 强制故障转移（不等Master确认）
redis-cli -p 7001 cluster failover FORCE

# 接管故障Master（手动指定新Master）
redis-cli -p 7001 cluster failover TAKEOVER

# 故障节点恢复后重新加入集群
redis-cli -p 7000 flushall
redis-cli -p 7000 cluster reset
redis-cli --cluster add-node 192.168.1.101:7000 192.168.1.102:7000 \
  --cluster-slave --cluster-master-id

槽位迁移性能优化与大批量数据迁移方案

大规模集群扩容时，单槽迁移效率低。使用redis-cli的–cluster reshard批量迁移，或编写自定义迁移脚本控制迁移速率。

# 批量槽位迁移（Python脚本）
import redis
import time

class SlotMigrator:
    def __init__(self, source_host, source_port, target_host, target_port):
        self.source = redis.Redis(host=source_host, port=source_port)
        self.target = redis.Redis(host=target_host, port=target_port)
        self.source_id = self.source.cluster("myid")
        self.target_id = self.target.cluster("myid")
        self.batch_size = 100
        self.migrate_timeout = 10000
        
        nodes = self.source.cluster("nodes")
        self.all_nodes = []
        for line in nodes.split("\n"):
            if line:
                parts = line.split()
                addr = parts[1].split("@")[0]
                host, port = addr.split(":")
                self.all_nodes.append(redis.Redis(host=host, port=int(port)))
    
    def migrate_slot(self, slot):
        for node in self.all_nodes:
            node.cluster("setslot", slot, "migrating", self.target_id)
        
        while True:
            keys = self.source.cluster("getkeysinslot", slot, self.batch_size)
            if not keys:
                break
            
            for key in keys:
                try:
                    self.source.migrate(
                        self.target.connection_pool.connection_kwargs["host"],
                        int(self.target.connection_pool.connection_kwargs["port"]),
                        key, 0, self.migrate_timeout, replace=True
                    )
                except redis.ResponseError as e:
                    if "BUSYKEY" in str(e):
                        self.source.migrate(
                            self.target.connection_pool.connection_kwargs["host"],
                            int(self.target.connection_pool.connection_kwargs["port"]),
                            key, 0, self.migrate_timeout, replace=True
                        )
                    elif "NOKEY" in str(e):
                        continue
                    else:
                        raise
            time.sleep(0.1)
        
        for node in self.all_nodes:
            node.cluster("setslot", slot, "node", self.target_id)
    
    def migrate_range(self, start_slot, end_slot, rate_limit=10):
        total = end_slot - start_slot + 1
        interval = 1.0 / rate_limit
        
        for slot in range(start_slot, end_slot + 1):
            start_time = time.time()
            self.migrate_slot(slot)
            print(f"Migrated slot {slot} ({slot - start_slot + 1}/{total})")
            
            elapsed = time.time() - start_time
            if elapsed < interval:
                time.sleep(interval - elapsed)

# 使用示例
migrator = SlotMigrator(
    source_host="192.168.1.101", source_port=7000,
    target_host="192.168.1.104", target_port=7000
)

# 从槽0迁移到槽4095，每秒迁移10个槽
migrator.migrate_range(0, 4095, rate_limit=10)

在生产环境执行槽位迁移时，建议在业务低峰期进行，迁移速率控制在每秒5-20个槽位。每个槽位的迁移时间取决于该槽位中key的数量和大小，单槽包含约6000个key（16384总槽位、1亿key的场景）时，单槽迁移时间约2-5秒。整个4096槽位的迁移可在20-60分钟内完成。迁移过程中客户端通过ASK重定向访问正在迁移的槽位，业务无感知。迁移完成后建议执行redis-check-rdb验证数据完整性，并对比源节点和目标节点的dbsize确认key数量一致。

原创文章，作者：小编，如若转载，请注明出处：https://www.yunthe.com/rediscluster-cao-wei-qian-yi-yuan-li-yu-nao-lie-gu-zhang/

赞 (0)
小编小编
0
上一篇 6小时前
下一篇 6小时前

相关推荐