Redis 集群完全指南：从原理到实战

slot = CRC16(key) % 16384

# 键 "user:1000" 的存储位置
CRC16("user:1000") = 52143
52143 % 16384 = 2799
→ 槽位 2799 在 Master1 (0-5460)
→ 数据存储在 Master1

# 键 "order:2000" 的存储位置
CRC16("order:2000") = 31324
31324 % 16384 = 14940
→ 槽位 14940 在 Master3 (10923-16383)
→ 数据存储在 Master3

# 用户相关数据
SET user:{10086}:profile "张三"
SET user:{10086}:balance "1000"
SET user:{10086}:vip "true"

# 订单相关数据
SET order:{20240101}:total "999"
SET order:{20240101}:items "商品列表"

# 好处：可以使用事务或 Lua 脚本
MULTI
SET user:{10086}:balance "900"
SET order:{20240101}:paid "true"
EXEC

- Redis 版本：≥ 5.0（推荐 7.x）
- 服务器：至少 3 台（推荐 6 台）
- 网络：节点间互相访问
- 端口：6379 和 16379

version: '3.8'

services:
  redis-1:
    image: redis:7.2-alpine
    container_name: redis-1
    command: redis-server --cluster-enabled yes --cluster-config-file nodes.conf --cluster-node-timeout 5000 --appendonly yes --port 6379
    ports:
      - "6379:6379"
      - "16379:16379"
    volumes:
      - redis-1-data:/data

  redis-2:
    image: redis:7.2-alpine
    container_name: redis-2
    command: redis-server --cluster-enabled yes --cluster-config-file nodes.conf --cluster-node-timeout 5000 --appendonly yes --port 6380
    ports:
      - "6380:6380"
      - "16380:16380"
    volumes:
      - redis-2-data:/data

  redis-3:
    image: redis:7.2-alpine
    container_name: redis-3
    command: redis-server --cluster-enabled yes --cluster-config-file nodes.conf --cluster-node-timeout 5000 --appendonly yes --port 6381
    ports:
      - "6381:6381"
      - "16381:16381"
    volumes:
      - redis-3-data:/data

  redis-4:
    image: redis:7.2-alpine
    container_name: redis-4
    command: redis-server --cluster-enabled yes --cluster-config-file nodes.conf --cluster-node-timeout 5000 --appendonly yes --port 6382
    ports:
      - "6382:6382"
      - "16382:16382"
    volumes:
      - redis-4-data:/data

  redis-5:
    image: redis:7.2-alpine
    container_name: redis-5
    command: redis-server --cluster-enabled yes --cluster-config-file nodes.conf --cluster-node-timeout 5000 --appendonly yes --port 6383
    ports:
      - "6383:6383"
      - "16383:16383"
    volumes:
      - redis-5-data:/data

  redis-6:
    image: redis:7.2-alpine
    container_name: redis-6
    command: redis-server --cluster-enabled yes --cluster-config-file nodes.conf --cluster-node-timeout 5000 --appendonly yes --port 6384
    ports:
      - "6384:6384"
      - "16384:16384"
    volumes:
      - redis-6-data:/data

volumes:
  redis-1-data:
  redis-2-data:
  redis-3-data:
  redis-4-data:
  redis-5-data:
  redis-6-data:

# 1. 启动所有节点
docker-compose up -d

# 2. 查看容器状态
docker-compose ps

# 3. 创建集群
docker exec -it redis-1 redis-cli --cluster create \
  172.17.0.2:6379 \
  172.17.0.3:6380 \
  172.17.0.4:6381 \
  172.17.0.5:6382 \
  172.17.0.6:6383 \
  172.17.0.7:6384 \
  --cluster-replicas 1

# 4. 验证集群
docker exec -it redis-1 redis-cli -c -p 6379 cluster nodes

# CentOS/RHEL
sudo yum install redis

# Ubuntu/Debian
sudo apt-get install redis-server

# 编译安装
wget https://download.redis.io/redis-stable.tar.gz
tar xzf redis-stable.tar.gz
cd redis-stable
make
sudo make install

mkdir -p /etc/redis/cluster/{7000,7001,7002,7003,7004,7005}

for port in 7001 7002 7003 7004 7005; do
  sed "s/7000/${port}/g" /etc/redis/cluster/7000/redis.conf > /etc/redis/cluster/${port}/redis.conf
done

# 创建数据目录
for port in 7000 7001 7002 7003 7004 7005; do
  mkdir -p /var/lib/redis/${port}
done

# 启动所有节点
for port in 7000 7001 7002 7003 7004 7005; do
  redis-server /etc/redis/cluster/${port}/redis.conf
done

# 检查进程
ps aux | grep redis-server

redis-cli --cluster create \
  127.0.0.1:7000 \
  127.0.0.1:7001 \
  127.0.0.1:7002 \
  127.0.0.1:7003 \
  127.0.0.1:7004 \
  127.0.0.1:7005 \
  --cluster-replicas 1

# 或者先添加 Bitnami 仓库
helm repo add bitnami https://charts.bitnami.com/bitnami
helm repo update

# 创建 values-redis-cluster.yaml 文件
cat > values-redis-cluster.yaml <<EOF
# 设置集群节点数量
cluster:
  nodes: 6  # 默认 6 个节点（3主3从）
  replicas: 1  # 每个主节点有 1 个从节点

# 配置持久化存储
persistence:
  enabled: true
  size: 8Gi
  storageClass: "local-path"  # 使用默认 StorageClass

# 设置 Redis 密码
password: "your-secure-password"

# 资源限制
resources:
  limits:
    cpu: 250m
    memory: 256Mi
  requests:
    cpu: 100m
    memory: 128Mi
EOF

# 部署 Redis 集群
helm upgrade -i redis bitnami/redis-cluster --version 13.0.4 -f values-redis-cluster.yaml

# 实现动态扩容节点

cat > values-expand-cluster.yaml <<EOF
cluster:
  init: true

  ## 扩容方案：从 6 节点扩到 12 节点
  ## 原配置：3 master + 1 replica (nodes=6, replicas=1)
  ## 新配置：6 master + 1 replica (nodes=12, replicas=1)
  ##
  nodes: 12
  replicas: 1

password: "your-secure-password"

# 配置持久化存储
persistence:
  enabled: true
  size: 8Gi
  storageClass: "local-path"  # 使用默认 StorageClass

# 资源限制
resources:
  limits:
    cpu: 250m
    memory: 256Mi
  requests:
    cpu: 100m
    memory: 128Mi

  ## 关键：扩容时需要启用 update 配置
  update:
    ## 设置为 true，启用节点添加功能
    addNodes: true

    ## 当前部署的节点数（扩容前的数量）
    currentNumberOfNodes: 6

    ## 当前部署的副本数（扩容前的数量）
    currentNumberOfReplicas: 1
EOF

# 扩容节点
helm upgrade -i redis bitnami/redis-cluster --version 13.0.4 -f values-expand-cluster.yaml

redis-cli -c -p 7000 cluster info

redis-cli -c -p 7000 cluster nodes

# 查看所有槽位分配
redis-cli -c -p 7000 cluster slots

# 计算键的槽位
redis-cli -c -p 7000 cluster keyslot "user:1000"
# Output: (integer) 2799

# 统计槽位中的键数量
redis-cli -c -p 7000 cluster countkeysinslot 2799
# Output: (integer) 5

# 获取槽位中的键（最多返回 count 个）
redis-cli -c -p 7000 cluster getkeysinslot 2799 10

# 将槽位 100-200 分配给指定节点
redis-cli -c -p 7000 cluster addslots 100 101 102 ... 200

# 批量分配
redis-cli -c -p 7000 cluster addslots {100..200}

# 删除槽位分配
redis-cli -c -p 7000 cluster delslots 100 101 102

# 添加新的 Master 节点
redis-cli --cluster add-node 127.0.0.1:7006 127.0.0.1:7000

# 添加新的 Slave 节点
redis-cli --cluster add-node 127.0.0.1:7007 127.0.0.1:7000 \
  --cluster-slave \
  --cluster-master-id <master节点ID>

# 或者先添加节点，再设置为从节点
redis-cli -c -p 7007 cluster replicate <master节点ID>

# 1. 如果是 Master，先迁移槽位（见下文）
# 2. 删除节点
redis-cli --cluster del-node 127.0.0.1:7000 <节点ID>

# 注意：节点ID 可以从 CLUSTER NODES 获取

# 在集群的每个节点上执行
redis-cli -c -p 7000 cluster forget <节点ID>

# 在 Slave 节点执行
redis-cli -c -p 7004 cluster failover

# 强制故障转移（即使数据不同步）
redis-cli -c -p 7004 cluster failover force

# 接管故障转移（Master 已下线）
redis-cli -c -p 7004 cluster failover takeover

# 软重置：保留节点ID和配置纪元
redis-cli -c -p 7000 cluster reset soft

# 硬重置：完全清空集群信息
redis-cli -c -p 7000 cluster reset hard

from rediscluster import RedisCluster

# 连接集群
startup_nodes = [
    {"host": "127.0.0.1", "port": "7000"},
    {"host": "127.0.0.1", "port": "7001"},
    {"host": "127.0.0.1", "port": "7002"}
]

rc = RedisCluster(
    startup_nodes=startup_nodes,
    decode_responses=True,
    skip_full_coverage_check=True  # 允许部分槽位不可用
)

# 客户端会自动维护槽位映射
rc.set("user:1000", "张三")
print(rc.get("user:1000"))

# 批量操作（使用 pipeline）
pipe = rc.pipeline()
pipe.set("user:1001", "李四")
pipe.set("user:1002", "王五")
pipe.execute()

import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;

Set<HostAndPort> nodes = new HashSet<>();
nodes.add(new HostAndPort("127.0.0.1", 7000));
nodes.add(new HostAndPort("127.0.0.1", 7001));
nodes.add(new HostAndPort("127.0.0.1", 7002));

JedisCluster jedis = new JedisCluster(nodes);

// 自动路由
jedis.set("user:1000", "张三");
System.out.println(jedis.get("user:1000"));

# 1. 计算槽位
redis-cli -c cluster keyslot "user:1000"
# Output: 2799

# 2. 查找槽位所在节点
redis-cli -c -p 7000 cluster nodes | grep "0-5460"

# 3. 连接对应节点
redis-cli -c -p 7000 get "user:1000"

# 同一用户的所有会话数据存储在一起
SET session:{user:10086}:token "abc123"
SET session:{user:10086}:expire "1640000000"
SET session:{user:10086}:device "iPhone"

# 使用事务
MULTI
SET session:{user:10086}:login_time "2024-01-01"
EXPIRE session:{user:10086}:login_time 3600
EXEC

# 同一资源的锁信息存储在一起
SET lock:{resource:order_12345}:owner "worker-1"
SET lock:{resource:order_12345}:expire "1640000000"

# 使用 Lua 脚本原子操作
EVAL "..." 2 lock:{resource:order_12345}:owner lock:{resource:order_12345}:expire

# 同一商品的库存信息
SET product:{SKU:12345}:stock "100"
SET product:{SKU:12345}:reserved "20"
SET product:{SKU:12345}:sold "80"

# 扣减库存（原子操作）
EVAL "
  local stock = redis.call('GET', KEYS[1])
  if tonumber(stock) >= tonumber(ARGV[1]) then
    redis.call('DECRBY', KEYS[1], ARGV[1])
    redis.call('INCRBY', KEYS[2], ARGV[1])
    return 1
  else
    return 0
  end
" 2 product:{SKU:12345}:stock product:{SKU:12345}:sold 5

# ❌ 错误：键在不同槽位
redis-cli -c -p 7000
> MGET user:1000 user:2000 user:3000
(error) CROSSSLOT Keys in request don't hash to the same slot

> MSET user:1000 "张三" user:2000 "李四"
(error) CROSSSLOT Keys in request don't hash to the same slot

# ✅ 正确：使用相同标签
MGET user:{batch1}:1000 user:{batch1}:2000 user:{batch1}:3000
MSET user:{batch1}:1000 "张三" user:{batch1}:2000 "李四"

# Python 示例
pipe = rc.pipeline()
pipe.get("user:1000")
pipe.get("user:2000")
pipe.get("user:3000")
results = pipe.execute()  # 自动路由到不同节点

# 手动聚合多个节点的数据
def mget_cluster(keys):
    results = {}
    for key in keys:
        results[key] = rc.get(key)
    return results

values = mget_cluster(["user:1000", "user:2000", "user:3000"])

# 模拟 Master 故障
redis-cli -c -p 7000 DEBUG SLEEP 30

# 其他节点日志
[7001] Node 7000 marked as PFAIL
[7001] Asking other nodes about 7000
[7001] Received enough PFAIL reports, marking 7000 as FAIL

# 自动执行的操作
1. 新 Master 接管原 Master 的槽位
2. 向集群广播配置变更
3. 其他 Slave 切换复制目标
4. 客户端更新路由表

# 旧 Master 重新上线后
1. 发现集群配置变更
2. 自动降级为 Slave
3. 开始复制新 Master 的数据

# 1. 查看当前集群状态
redis-cli -c -p 7000 cluster nodes

# 输出示例
a1b2c3... 127.0.0.1:7000@17000 master - 0 1640000000 1 connected 0-5460
e5f6a1... 127.0.0.1:7004@17004 slave a1b2c3... 0 1640000000 1 connected

# 2. 停止 Master (7000)
docker stop redis-1
# 或
redis-cli -c -p 7000 shutdown

# 3. 观察故障转移（15-30秒内）
watch -n 1 'redis-cli -c -p 7001 cluster nodes'

# 4. 验证 Slave 提升
redis-cli -c -p 7004 cluster nodes

# 输出：7004 已变为 master
e5f6a1... 127.0.0.1:7004@17004 master - 0 1640000030 7 connected 0-5460

# 5. 重启旧 Master
docker start redis-1

# 6. 观察降级为 Slave
redis-cli -c -p 7000 cluster nodes
a1b2... 127.0.0.1:7000@17000 slave e5f6a1... 0 1640000060 7 connected

# 1. 模拟网络分区（使用 iptables）
# 隔离节点 7000
sudo iptables -A INPUT -p tcp --dport 7000 -j DROP
sudo iptables -A INPUT -p tcp --dport 17000 -j DROP

# 2. 观察集群行为
# 节点 7000 被标记为 FAIL
redis-cli -c -p 7001 cluster nodes

# 3. 恢复网络
sudo iptables -F

# 4. 验证节点重新加入
redis-cli -c -p 7000 cluster nodes

# 步骤 1：启动新节点
redis-server --port 7006 \
  --cluster-enabled yes \
  --cluster-config-file nodes-7006.conf \
  --daemonize yes

# 步骤 2：加入集群（作为 Master）
redis-cli --cluster add-node 127.0.0.1:7006 127.0.0.1:7000

# 步骤 3：检查节点状态
redis-cli -c -p 7000 cluster nodes
# 输出：新节点没有分配槽位
f1g2h3... 127.0.0.1:7006@17006 master - 0 1640000000 0 connected

# 步骤 4：重新分配槽位
redis-cli --cluster reshard 127.0.0.1:7000

# 交互式提示
How many slots do you want to move? 4096  # 从3个节点分别迁移
What is the receiving node ID? f1g2h3...   # 新节点ID
Source node #1: all                        # 从所有节点迁移
Do you want to proceed? yes

# 步骤 5：验证槽位分配
redis-cli -c -p 7000 cluster nodes

#!/bin/bash
# add_master.sh

NEW_PORT=$1
CLUSTER_NODE="127.0.0.1:7000"

echo "启动新节点: $NEW_PORT"
redis-server --port $NEW_PORT \
  --cluster-enabled yes \
  --cluster-config-file nodes-${NEW_PORT}.conf \
  --daemonize yes

sleep 2

echo "加入集群"
redis-cli --cluster add-node 127.0.0.1:$NEW_PORT $CLUSTER_NODE

echo "自动分配槽位"
redis-cli --cluster rebalance $CLUSTER_NODE \
  --cluster-use-empty-masters

echo "完成！"
redis-cli -c -p 7000 cluster nodes

# 方式 1：添加时指定 Master
redis-cli --cluster add-node 127.0.0.1:7007 127.0.0.1:7000 \
  --cluster-slave \
  --cluster-master-id a1b2c3d4e5f6...

# 方式 2：先添加后指定
redis-cli --cluster add-node 127.0.0.1:7007 127.0.0.1:7000
redis-cli -c -p 7007 cluster replicate a1b2c3d4e5f6...

# 方式 3：自动分配（给 Slave 最少的 Master）
redis-cli --cluster add-node 127.0.0.1:7007 127.0.0.1:7000 \
  --cluster-slave

# 直接删除（简单）
redis-cli --cluster del-node 127.0.0.1:7000 <slave节点ID>

# 步骤 1：迁移槽位到其他节点
redis-cli --cluster reshard 127.0.0.1:7000

# 交互式提示
How many slots? 5461                    # 该节点的槽位总数
What is the receiving node ID? b2c3... # 目标节点
Source node #1: a1b2...                # 源节点（要删除的节点）
Source node #2: done
Do you want to proceed? yes

# 步骤 2：验证槽位已清空
redis-cli -c -p 7000 cluster nodes | grep a1b2...
# 应显示：connected (无槽位范围)

# 步骤 3：删除节点
redis-cli --cluster del-node 127.0.0.1:7000 a1b2c3d4e5f6...

# 步骤 4：停止节点进程
redis-cli -p 7000 shutdown

# 1. 设置源节点为 MIGRATING 状态
redis-cli -c -p 7000 cluster setslot 1000 migrating b2c3d4...

# 2. 设置目标节点为 IMPORTING 状态
redis-cli -c -p 7001 cluster setslot 1000 importing a1b2c3...

# 3. 获取槽位中的键
keys=$(redis-cli -c -p 7000 cluster getkeysinslot 1000 100)

# 4. 迁移每个键
for key in $keys; do
  redis-cli -c -p 7000 migrate 127.0.0.1 7001 "$key" 0 5000
done

# 5. 确认目标节点拥有槽位
redis-cli -c -p 7001 cluster setslot 1000 node b2c3d4...

# 6. 确认源节点释放槽位
redis-cli -c -p 7000 cluster setslot 1000 node b2c3d4...

# 计算需要的节点数
当前数据量 = 100GB
单节点容量 = 32GB
安全系数 = 0.7

需要节点数 = ceil(100 / (32 * 0.7)) = 5 个 Master

# 不要一次性添加太多节点
# 推荐：每次添加 1-2 个节点

# 第一批：添加 1 个节点
./add_master.sh 7006

# 等待数据迁移完成（监控 cluster info）
redis-cli -c -p 7000 cluster info | grep cluster_state

# 第二批：添加第 2 个节点
./add_master.sh 7007

# 监控脚本
#!/bin/bash
while true; do
  echo "=== $(date) ==="
  redis-cli -c -p 7000 info replication | grep master_repl_offset
  redis-cli -c -p 7000 cluster info | grep -E "cluster_state|cluster_slots"
  sleep 5
done

# /etc/sysctl.conf
net.core.somaxconn = 511
net.ipv4.tcp_max_syn_backlog = 511
vm.overcommit_memory = 1

# 透明大页
echo never > /sys/kernel/mm/transparent_hugepage/enabled

# 应用配置
sysctl -p

from rediscluster import RedisCluster

rc = RedisCluster(
    startup_nodes=nodes,
    max_connections=50,           # 每个节点最大连接数
    max_connections_per_node=True,
    socket_timeout=5,
    socket_connect_timeout=5,
    socket_keepalive=True,
    decode_responses=True
)

JedisPoolConfig config = new JedisPoolConfig();
config.setMaxTotal(100);        // 最大连接数
config.setMaxIdle(20);          // 最大空闲连接
config.setMinIdle(10);          // 最小空闲连接
config.setMaxWaitMillis(3000);  // 获取连接超时
config.setTestOnBorrow(true);   // 获取时测试连接

JedisCluster jedis = new JedisCluster(nodes, 3000, 3000, 5, password, config);

# ❌ 低效：逐个操作
for i in range(1000):
    rc.set(f"key:{i}", f"value:{i}")
# 耗时：~2000ms (每次 RTT ~2ms)

# ✅ 高效：使用 Pipeline
pipe = rc.pipeline()
for i in range(1000):
    pipe.set(f"key:{i}", f"value:{i}")
pipe.execute()
# 耗时：~50ms (批量发送)

# 场景：存储用户信息

# 方式 1：多个 String（不推荐）
SET user:10086:name "张三"
SET user:10086:age "25"
SET user:10086:email "zhang@example.com"
# 内存：~300 bytes

# 方式 2：单个 Hash（推荐）
HSET user:10086 name "张三" age "25" email "zhang@example.com"
# 内存：~100 bytes (节省 66%)

# 场景：好友列表

# 方式 1：Set（无序）
SADD user:10086:friends 10087 10088 10089
SISMEMBER user:10086:friends 10087  # O(1)

# 方式 2：Sorted Set（需要排序）
ZADD user:10086:friends 1640000001 10087 1640000002 10088
ZRANGE user:10086:friends 0 -1  # 按时间排序

# ✅ 推荐格式
业务:对象:ID:属性

# 示例
user:profile:10086:name
order:info:20240101:total
product:stock:SKU12345:count
cache:homepage:banner:list

# ❌ 不推荐
user_10086_name              # 难以分类
very_long_key_name_is_bad    # 浪费内存
中文键名                      # 编码问题

# 不同数据类型的 TTL 建议
TTL_CONFIG = {
    'cache:homepage': 300,        # 5分钟
    'cache:product': 3600,        # 1小时
    'session:user': 86400,        # 1天
    'lock:resource': 30,          # 30秒
    'temp:verification': 300,     # 5分钟
}

# 设置过期时间
rc.setex('cache:homepage:banner', 300, json.dumps(data))

# 1. 性能指标
redis-cli -c -p 7000 info stats | grep -E "ops_per_sec|hit_rate"

# 2. 内存指标
redis-cli -c -p 7000 info memory | grep -E "used_memory|fragmentation"

# 3. 持久化指标
redis-cli -c -p 7000 info persistence | grep -E "rdb_last|aof_last"

# 4. 复制指标
redis-cli -c -p 7000 info replication | grep -E "master_repl|slave"

# 5. 集群指标
redis-cli -c -p 7000 cluster info

# 1. 内存指标
- used_memory              # 已用内存
- used_memory_rss          # 物理内存
- mem_fragmentation_ratio  # 碎片率（建议<1.5）

# 2. 性能指标
- instantaneous_ops_per_sec  # QPS
- keyspace_hits              # 命中次数
- keyspace_misses            # 未命中次数
- hit_rate = hits / (hits + misses)  # 命中率（建议>90%）

# 3. 持久化指标
- rdb_last_save_time         # 最后RDB时间
- aof_current_size           # AOF文件大小
- aof_last_rewrite_time_sec  # AOF重写耗时

# 4. 集群指标
- cluster_state              # 集群状态（ok/fail）
- cluster_slots_assigned     # 已分配槽位（应为16384）
- cluster_known_nodes        # 已知节点数

# 5. 连接指标
- connected_clients          # 当前连接数
- blocked_clients            # 阻塞客户端数
- rejected_connections       # 拒绝连接数

# redis-exporter 配置
scrape_configs:
  - job_name: 'redis-cluster'
    static_configs:
      - targets:
        - redis-exporter:9121
    params:
      target:
        - redis://127.0.0.1:7000
        - redis://127.0.0.1:7001
        - redis://127.0.0.1:7002

# ServiceMonitor示例
apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: redis-exporter
spec:
  selector:
    matchLabels:
      app: redis-exporter
  endpoints:
  - port: metrics
    interval: 30s

groups:
- name: redis-alerts
  rules:
  # 内存使用率告警
  - alert: RedisMemoryHigh
    expr: redis_memory_used_bytes / redis_memory_max_bytes > 0.9
    for: 5m
    annotations:
      summary: "Redis内存使用率超过90%"

  # 命中率告警
  - alert: RedisLowHitRate
    expr: redis_keyspace_hits_total / (redis_keyspace_hits_total + redis_keyspace_misses_total) < 0.8
    for: 10m
    annotations:
      summary: "Redis命中率低于80%"

  # 集群状态告警
  - alert: RedisClusterDown
    expr: redis_cluster_state != 1
    for: 1m
    annotations:
      summary: "Redis集群状态异常"

# 原因：节点无法相互通信

# 排查步骤
# 1. 检查端口是否监听
netstat -tlnp | grep redis

# 2. 检查防火墙
sudo iptables -L -n | grep 6379
sudo iptables -L -n | grep 16379

# 3. 测试连通性
redis-cli -h 127.0.0.1 -p 7000 ping
redis-cli -h 127.0.0.1 -p 7001 ping

# 解决方案
# 开放端口
sudo firewall-cmd --permanent --add-port=6379/tcp
sudo firewall-cmd --permanent --add-port=16379/tcp
sudo firewall-cmd --reload

# 原因：节点已有数据或集群配置

# 解决方案
redis-cli -c -p 7000 cluster reset hard
redis-cli -c -p 7000 flushall

# 原因：部分槽位未分配或节点下线

# 排查
redis-cli -c -p 7000 cluster nodes
redis-cli -c -p 7000 cluster slots

# 解决方案 1：修复槽位分配
redis-cli --cluster fix 127.0.0.1:7000

# 解决方案 2：允许部分槽位不可用
redis-cli -c -p 7000 config set cluster-require-full-coverage no

# 检查复制偏移量
redis-cli -c -p 7000 info replication

# Master
master_repl_offset:1234567

# Slave
redis-cli -c -p 7004 info replication
slave_repl_offset:1234500  # 延迟 67 字节

# 解决方案
# 1. 检查网络延迟
ping 从节点IP

# 2. 检查主节点负载
redis-cli -c -p 7000 info stats

# 3. 调整复制参数
repl-backlog-size 64mb
repl-timeout 60

# 查看慢查询
redis-cli -c -p 7000 slowlog get 10

# 输出示例
1) 1) (integer) 3        # 日志ID
   2) (integer) 1640000000  # 时间戳
   3) (integer) 12000    # 执行时间(微秒)
   4) 1) "KEYS"          # 命令
      2) "user:*"

# 优化建议
# ❌ 避免使用 KEYS
KEYS user:*

# ✅ 使用 SCAN
SCAN 0 MATCH user:* COUNT 100

# 场景：3主3从集群，Master1 彻底损坏

# 步骤 1：确认故障范围
redis-cli -c -p 7001 cluster nodes
# Master1 (7000) 和 Slave1 (7004) 都 FAIL

# 步骤 2：评估数据影响
redis-cli -c -p 7001 cluster slots
# 槽位 0-5460 不可用

# 步骤 3：启动新节点替换
redis-server --port 7008 \
  --cluster-enabled yes \
  --cluster-config-file nodes-7008.conf

# 步骤 4：加入集群
redis-cli --cluster add-node 127.0.0.1:7008 127.0.0.1:7001

# 步骤 5：分配槽位
redis-cli -c -p 7008 cluster addslots {0..5460}

# 步骤 6：从备份恢复数据（如果有）
redis-cli -c -p 7008 < backup.rdb

# 步骤 7：验证集群
redis-cli --cluster check 127.0.0.1:7001

-- ❌ 错误：访问多个槽位的键
redis.call('SET', 'user:1000', 'value1')  -- 槽位 A
redis.call('SET', 'user:2000', 'value2')  -- 槽位 B
-- Error: Script attempted to access keys that don't hash to the same slot

-- ✅ 正确：使用哈希标签
redis.call('SET', 'user:{batch1}:1000', 'value1')
redis.call('SET', 'user:{batch1}:2000', 'value2')

-- rate_limiter.lua
-- 使用滑动窗口实现限流
local key = KEYS[1]              -- 限流键，如 "rate_limit:{user:10086}"
local limit = tonumber(ARGV[1])  -- 限流阈值，如 100
local window = tonumber(ARGV[2]) -- 时间窗口(秒)，如 60
local current_time = tonumber(ARGV[3])

-- 清理过期记录
redis.call('ZREMRANGEBYSCORE', key, 0, current_time - window)

-- 获取当前窗口内的请求数
local current_count = redis.call('ZCARD', key)

if current_count < limit then
    -- 允许请求，记录当前时间戳
    redis.call('ZADD', key, current_time, current_time .. math.random())
    redis.call('EXPIRE', key, window)
    return 1
else
    -- 拒绝请求
    return 0
end

import time
import hashlib

def rate_limit(user_id, limit=100, window=60):
    # 使用哈希标签确保同一用户的限流数据在同一槽位
    key = f"rate_limit:{{user:{user_id}}}"
    current_time = int(time.time())

    # 执行 Lua 脚本
    with open('rate_limiter.lua', 'r') as f:
        script = f.read()

    result = rc.eval(script, 1, key, limit, window, current_time)
    return result == 1

# 测试
if rate_limit(10086):
    print("允许访问")
else:
    print("请求过于频繁，请稍后再试")

# 发布者
def publish_message(channel, message):
    # 集群模式下，消息会广播到所有节点
    rc.publish(channel, message)

# 订阅者
from redis import Redis

# 需要连接到每个节点
nodes = [
    Redis(host='127.0.0.1', port=7000),
    Redis(host='127.0.0.1', port=7001),
    Redis(host='127.0.0.1', port=7002)
]

def subscribe_channel(channel):
    for node in nodes:
        pubsub = node.pubsub()
        pubsub.subscribe(channel)

        for message in pubsub.listen():
            if message['type'] == 'message':
                print(f"收到消息: {message['data']}")

# 场景：订单状态变更通知

# 1. 订单服务发布消息
def order_status_changed(order_id, status):
    channel = f"order:{{notify:{order_id % 10}}}"  # 分片
    message = json.dumps({
        'order_id': order_id,
        'status': status,
        'timestamp': time.time()
    })
    rc.publish(channel, message)

# 2. 通知服务订阅消息
def start_notification_worker(shard_id):
    channel = f"order:{{notify:{shard_id}}}"
    pubsub = rc.pubsub()
    pubsub.subscribe(channel)

    for message in pubsub.listen():
        if message['type'] == 'message':
            data = json.loads(message['data'])
            send_notification(data['order_id'], data['status'])

# 3. 启动多个 Worker 处理不同分片
from multiprocessing import Process

workers = []
for i in range(10):
    p = Process(target=start_notification_worker, args=(i,))
    p.start()
    workers.append(p)

# 生产消息
XADD order:stream * order_id 12345 amount 999 status pending
# 返回：1640000000000-0

# 创建消费组
XGROUP CREATE order:stream order-processor 0

# 消费消息
XREADGROUP GROUP order-processor consumer1 COUNT 10 STREAMS order:stream >

# 确认消息
XACK order:stream order-processor 1640000000000-0

# 查看待处理消息
XPENDING order:stream order-processor

class RedisStreamQueue:
    def __init__(self, stream_key, group_name):
        self.stream_key = f"{{queue:{stream_key}}}"  # 哈希标签
        self.group_name = group_name

        # 创建消费组（忽略已存在错误）
        try:
            rc.xgroup_create(self.stream_key, self.group_name, id='0', mkstream=True)
        except:
            pass

    def produce(self, data):
        """生产消息"""
        return rc.xadd(self.stream_key, data)

    def consume(self, consumer_name, count=10, block=5000):
        """消费消息"""
        messages = rc.xreadgroup(
            groupname=self.group_name,
            consumername=consumer_name,
            streams={self.stream_key: '>'},
            count=count,
            block=block
        )

        result = []
        if messages:
            for stream, msgs in messages:
                for msg_id, data in msgs:
                    result.append((msg_id, data))
        return result

    def ack(self, *msg_ids):
        """确认消息"""
        return rc.xack(self.stream_key, self.group_name, *msg_ids)

    def claim_stale_messages(self, consumer_name, min_idle_time=60000):
        """认领超时消息"""
        pending = rc.xpending_range(
            self.stream_key,
            self.group_name,
            min='-',
            max='+',
            count=100
        )

        stale_ids = [
            msg['message_id']
            for msg in pending
            if msg['time_since_delivered'] > min_idle_time
        ]

        if stale_ids:
            return rc.xclaim(
                self.stream_key,
                self.group_name,
                consumer_name,
                min_idle_time,
                stale_ids
            )
        return []

# 使用示例
queue = RedisStreamQueue('order:processing', 'order-workers')

# 生产者
def produce_orders():
    for i in range(100):
        queue.produce({
            'order_id': i,
            'amount': 999,
            'created_at': time.time()
        })

# 消费者
def process_orders(worker_id):
    while True:
        # 消费新消息
        messages = queue.consume(f'worker-{worker_id}', count=10)

        for msg_id, data in messages:
            try:
                # 处理订单
                process_order(data)
                # 确认消息
                queue.ack(msg_id)
            except Exception as e:
                logger.error(f"处理失败: {e}")

        # 认领超时消息
        stale = queue.claim_stale_messages(f'worker-{worker_id}')
        for msg_id, data in stale:
            try:
                process_order(data)
                queue.ack(msg_id)
            except Exception as e:
                logger.error(f"重试失败: {e}")

import time
import uuid

class RedisDistributedLock:
    def __init__(self, resource_name, ttl=30000):
        # 使用哈希标签确保锁在同一槽位
        self.resource = f"lock:{{resource:{resource_name}}}"
        self.ttl = ttl  # 毫秒
        self.token = str(uuid.uuid4())

    def acquire(self):
        """获取锁"""
        # SET resource token NX PX ttl
        result = rc.set(
            self.resource,
            self.token,
            nx=True,
            px=self.ttl
        )
        return result is not None

    def release(self):
        """释放锁（使用 Lua 保证原子性）"""
        lua_script = """
        if redis.call("GET", KEYS[1]) == ARGV[1] then
            return redis.call("DEL", KEYS[1])
        else
            return 0
        end
        """
        return rc.eval(lua_script, 1, self.resource, self.token)

    def extend(self, additional_time=30000):
        """延长锁的有效期"""
        lua_script = """
        if redis.call("GET", KEYS[1]) == ARGV[1] then
            return redis.call("PEXPIRE", KEYS[1], ARGV[2])
        else
            return 0
        end
        """
        return rc.eval(lua_script, 1, self.resource, self.token, additional_time)

    def __enter__(self):
        retry_count = 0
        max_retries = 3

        while retry_count < max_retries:
            if self.acquire():
                return self
            retry_count += 1
            time.sleep(0.1)

        raise Exception(f"无法获取锁: {self.resource}")

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.release()

# 使用示例
def process_order(order_id):
    lock_key = f"order:{order_id}"

    with RedisDistributedLock(lock_key, ttl=5000):
        # 执行业务逻辑
        order = get_order(order_id)
        order['status'] = 'processing'
        update_order(order)

        # 长时间操作可以延长锁
        # lock.extend(5000)

def submit_order(user_id, order_data):
    """防止订单重复提交"""
    # 使用用户ID和订单数据生成唯一标识
    order_hash = hashlib.md5(
        json.dumps(order_data, sort_keys=True).encode()
    ).hexdigest()

    lock_key = f"submit:{user_id}:{order_hash}"

    try:
        with RedisDistributedLock(lock_key, ttl=10000):
            # 检查是否已提交
            submitted_key = f"submitted:{{user:{user_id}}}:{order_hash}"
            if rc.exists(submitted_key):
                return {"success": False, "message": "订单已提交，请勿重复操作"}

            # 创建订单
            order_id = create_order(user_id, order_data)

            # 标记已提交（24小时过期）
            rc.setex(submitted_key, 86400, order_id)

            return {"success": True, "order_id": order_id}
    except Exception as e:
        return {"success": False, "message": "系统繁忙，请稍后再试"}

#!/usr/bin/env python3
# migrate_to_cluster.py

import redis
from rediscluster import RedisCluster

# 源：单机 Redis
source = redis.Redis(host='old-redis', port=6379, db=0)

# 目标：Redis 集群
target = RedisCluster(
    startup_nodes=[{"host": "127.0.0.1", "port": "7000"}],
    decode_responses=True
)

def migrate_keys(pattern='*', batch_size=1000):
    """迁移键"""
    cursor = 0
    migrated = 0

    while True:
        cursor, keys = source.scan(cursor, match=pattern, count=batch_size)

        for key in keys:
            try:
                # 获取键类型
                key_type = source.type(key)
                ttl = source.ttl(key)

                # 根据类型迁移
                if key_type == b'string':
                    value = source.get(key)
                    target.set(key, value)

                elif key_type == b'hash':
                    value = source.hgetall(key)
                    target.hmset(key, value)

                elif key_type == b'list':
                    value = source.lrange(key, 0, -1)
                    if value:
                        target.rpush(key, *value)

                elif key_type == b'set':
                    value = source.smembers(key)
                    if value:
                        target.sadd(key, *value)

                elif key_type == b'zset':
                    value = source.zrange(key, 0, -1, withscores=True)
                    if value:
                        target.zadd(key, dict(value))

                # 设置过期时间
                if ttl > 0:
                    target.expire(key, ttl)

                migrated += 1
                if migrated % 1000 == 0:
                    print(f"已迁移 {migrated} 个键")

            except Exception as e:
                print(f"迁移失败 {key}: {e}")

        if cursor == 0:
            break

    print(f"迁移完成！总计 {migrated} 个键")
    return migrated

# 执行迁移
if __name__ == '__main__':
    migrate_keys()

class DualWriteRedis:
    """双写 Redis：同时写入旧集群和新集群"""

    def __init__(self, old_redis, new_cluster):
        self.old = old_redis
        self.new = new_cluster

    def set(self, key, value, ex=None):
        # 先写新集群
        self.new.set(key, value, ex=ex)
        # 再写旧集群（失败不影响）
        try:
            self.old.set(key, value, ex=ex)
        except:
            pass

    def get(self, key):
        # 优先从新集群读取
        value = self.new.get(key)
        if value is not None:
            return value

        # 新集群没有，从旧集群读取并回填
        value = self.old.get(key)
        if value is not None:
            try:
                self.new.set(key, value)
            except:
                pass
        return value

    def delete(self, *keys):
        self.new.delete(*keys)
        try:
            self.old.delete(*keys)
        except:
            pass

# 使用示例
old_redis = redis.Redis(host='old', port=6379)
new_cluster = RedisCluster(startup_nodes=[...])

# 应用程序使用双写客户端
rc = DualWriteRedis(old_redis, new_cluster)

# 业务代码无需修改
rc.set('user:1000', 'data')
value = rc.get('user:1000')

# 基础测试
redis-benchmark -h 127.0.0.1 -p 7000 -c 50 -n 100000

# 测试特定命令
redis-benchmark -h 127.0.0.1 -p 7000 -t set,get -n 100000 -q

# 集群模式测试
redis-benchmark -h 127.0.0.1 -p 7000 --cluster -c 50 -n 100000

# 测试不同数据大小
redis-benchmark -h 127.0.0.1 -p 7000 -t set -n 100000 -d 1024

# Pipeline 测试
redis-benchmark -h 127.0.0.1 -p 7000 -t set -n 100000 -P 16

import time
import threading
from concurrent.futures import ThreadPoolExecutor
import random

class RedisBenchmark:
    def __init__(self, redis_client, num_threads=10, num_requests=10000):
        self.rc = redis_client
        self.num_threads = num_threads
        self.num_requests = num_requests
        self.results = []

    def test_set(self, thread_id, count):
        """测试 SET 操作"""
        latencies = []

        for i in range(count):
            key = f"bench:{{thread:{thread_id}}}:key:{i}"
            value = f"value_{i}"

            start = time.time()
            self.rc.set(key, value)
            latency = (time.time() - start) * 1000  # 毫秒
            latencies.append(latency)

        return latencies

    def test_get(self, thread_id, count):
        """测试 GET 操作"""
        # 先写入数据
        keys = []
        for i in range(count):
            key = f"bench:{{thread:{thread_id}}}:key:{i}"
            self.rc.set(key, f"value_{i}")
            keys.append(key)

        # 测试读取
        latencies = []
        for key in keys:
            start = time.time()
            self.rc.get(key)
            latency = (time.time() - start) * 1000
            latencies.append(latency)

        return latencies

    def run(self, test_func):
        """运行测试"""
        requests_per_thread = self.num_requests // self.num_threads

        start_time = time.time()

        with ThreadPoolExecutor(max_workers=self.num_threads) as executor:
            futures = []
            for i in range(self.num_threads):
                future = executor.submit(test_func, i, requests_per_thread)
                futures.append(future)

            for future in futures:
                self.results.extend(future.result())

        duration = time.time() - start_time

        # 统计结果
        self.print_stats(duration)

    def print_stats(self, duration):
        """打印统计信息"""
        total_requests = len(self.results)
        qps = total_requests / duration

        sorted_latencies = sorted(self.results)
        p50 = sorted_latencies[int(len(sorted_latencies) * 0.50)]
        p95 = sorted_latencies[int(len(sorted_latencies) * 0.95)]
        p99 = sorted_latencies[int(len(sorted_latencies) * 0.99)]
        avg = sum(sorted_latencies) / len(sorted_latencies)

        print(f"""
基准测试结果
=====================================
总请求数：{total_requests}
耗时：{duration:.2f} 秒
QPS：{qps:.2f}

延迟统计（毫秒）：
  平均：{avg:.2f}
  P50：{p50:.2f}
  P95：{p95:.2f}
  P99：{p99:.2f}
  最大：{max(sorted_latencies):.2f}
        """)

# 使用示例
benchmark = RedisBenchmark(rc, num_threads=50, num_requests=100000)
print("测试 SET 操作...")
benchmark.run(benchmark.test_set)
benchmark.results = []
print("\n测试 GET 操作...")
benchmark.run(benchmark.test_get)

# 1. 启用密码认证
requirepass YourStrongPassword123!
masterauth YourStrongPassword123!

# 2. 配置 ACL
aclfile /etc/redis/users.acl

# 3. 禁用危险命令
rename-command FLUSHALL ""
rename-command FLUSHDB ""
rename-command CONFIG "CONFIG-$(openssl rand -hex 8)"
rename-command KEYS ""

# 4. 网络安全
bind 0.0.0.0  # 生产环境改为内网 IP
protected-mode yes
port 6379
requirepass strong-password

# 5. 限制客户端连接
maxclients 10000
timeout 300

# 6. 关闭 Lua 脚本调试
lua-time-limit 5000
# 不要开启 lua-debugger

# 7. 日志审计
logfile /var/log/redis/redis.log
loglevel notice

# 8. 禁用 DEBUG 和 MONITOR（生产环境）
# 通过 ACL 控制

# 9. TLS 加密（可选）
tls-port 6380
tls-cert-file /path/to/redis.crt
tls-key-file /path/to/redis.key
tls-ca-cert-file /path/to/ca.crt

#!/bin/bash
# security_check.sh

echo "Redis 集群安全检查"
echo "===================="

# 检查密码强度
echo "1. 检查认证配置..."
redis-cli -c -p 7000 ping 2>&1 | grep -q "NOAUTH" && echo "⚠️  未配置密码" || echo "✅ 已配置密码"

# 检查危险命令
echo "2. 检查危险命令..."
for cmd in FLUSHALL FLUSHDB KEYS; do
    redis-cli -a "$REDIS_PASS" -p 7000 $cmd 2>&1 | grep -q "unknown command" && \
        echo "✅ $cmd 已禁用" || echo "⚠️  $cmd 仍可用"
done

# 检查 ACL
echo "3. 检查 ACL 配置..."
redis-cli -a "$REDIS_PASS" -p 7000 ACL LIST | grep -q "user default" && \
    echo "✅ ACL 已配置" || echo "⚠️  未配置 ACL"

# 检查网络绑定
echo "4. 检查网络配置..."
netstat -tlnp | grep redis-server | grep "0.0.0.0" && \
    echo "⚠️  绑定到所有网卡" || echo "✅ 只绑定特定 IP"

# 检查慢查询
echo "5. 检查慢查询..."
slow_count=$(redis-cli -a "$REDIS_PASS" -p 7000 SLOWLOG LEN)
echo "慢查询数量: $slow_count"

echo "===================="
echo "安全检查完成"