Redis 是后端开发中使用最广泛的内存数据库，据 DB-Engines 2025 年统计，Redis 连续 8 年位列键值数据库排名第一。但大多数开发者对 Redis 的使用停留在 get/set 层面，对其数据结构选型、缓存策略设计和生产环境的坑知之甚少。本文将从实战角度出发，帮你真正掌握 Redis 在高并发场景下的正确用法。
🔧 一、数据结构选型：选对结构比优化代码更重要
Redis 提供了 String、Hash、List、Set、Sorted Set、Stream、Bitmap、HyperLogLog 等 8 种数据结构。选错结构不仅浪费内存，还会影响查询性能。
1.1 String vs Hash：存储对象时的选择
很多开发者习惯把对象序列化成 JSON 字符串存入 String，但这在需要部分更新字段时会带来额外开销。
// ❌ 错误写法：序列化整个对象存 String，更新一个字段需要先取再存
String userJson = redisTemplate.opsForValue().get("user:1001");
User user = objectMapper.readValue(userJson, User.class);
user.setLastLogin(now);
redisTemplate.opsForValue().set("user:1001", objectMapper.writeValueAsString(user));
// ✅ 正确写法：用 Hash 存储，直接更新单个字段
redisTemplate.opsForHash().put("user:1001", "lastLogin", String.valueOf(now));
// 只更新一个字段，无需序列化/反序列化整个对象
redisTemplate.opsForHash().put("user:1001", "name", "张三");
内存对比实测数据（存储 10000 个用户对象，每个对象 8 个字段）：
| 存储方式 | 内存占用 | 部分更新耗时 | 适用场景 |
|---|---|---|---|
| String + JSON | ~12.5 MB | 2 次 IO（读+写） | 对象整体读取，极少部分更新 |
| Hash | ~6.8 MB | 1 次 IO | 频繁读写单个字段 |
| Hash (ziplist 编码) | ~4.2 MB | 1 次 IO | 字段数 < 128，值 < 64 字节 |
⚠️ 警告： 当 Hash 的字段数超过
hash-max-ziplist-entries（默认 128）或值超过hash-max-ziplist-value（默认 64 字节）时，Redis 会将编码从 ziplist 转为 hashtable，内存占用会骤增 3-5 倍。
1.2 Sorted Set：排行榜的唯一正确选择
做排行榜功能时，很多人用 List + 排序，这在数据量大时性能极差。
// ❌ 错误写法：用 List 存储排行榜，每次取排行需要全量读取 + 排序
await redis.lPush('leaderboard', JSON.stringify({ userId: 'u1', score: 95 }));
const all = await redis.lRange('leaderboard', 0, -1);
const sorted = all.map(JSON.parse).sort((a, b) => b.score - a.score);
// ✅ 正确写法：用 Sorted Set，O(logN) 插入，O(logN+M) 范围查询
await redis.zAdd('leaderboard', [{ score: 95, value: 'u1' }]);
await redis.zAdd('leaderboard', [{ score: 87, value: 'u2' }]);
await redis.zAdd('leaderboard', [{ score: 100, value: 'u3' }]);
// 获取 Top 10（按分数从高到低）—— O(logN + 10)
const top10 = await redis.zRangeWithScores('leaderboard', 0, 9, { REV: true });
// 获取某用户排名 —— O(logN)
const rank = await redis.zRevRank('leaderboard', 'u1'); // 返回排名（0-based）
const score = await redis.zScore('leaderboard', 'u1'); // 返回分数
💡 提示： Sorted Set 的
ZADD操作时间复杂度为 O(logN)，ZRANGE为 O(logN+M)，ZRANK为 O(logN)。即使百万级数据，响应时间也在毫秒级。
1.3 HyperLogLog：UV 统计的内存杀手锏
统计页面独立访客（UV），如果用 Set 存储每个用户 ID，1 亿用户会占用约 4GB 内存。HyperLogLog 只需 12KB 就能统计 2^64 个不同元素，误差率仅 0.81%。
# Python + redis-py 示例：统计每日 UV
import redis
from datetime import date
r = redis.Redis(host='localhost', port=6379, db=0)
def record_visit(page: str, user_id: str):
"""记录一次页面访问"""
today = date.today().isoformat()
key = f"uv:{page}:{today}"
# PFADD 添加元素，如果 key 不存在会自动创建
r.pfadd(key, user_id)
# 设置过期时间为 3 天，自动清理旧数据
r.expire(key, 86400 * 3)
def get_uv(page: str, date_str: str = None) -> int:
"""获取指定页面的 UV 数"""
if date_str is None:
date_str = date.today().isoformat()
key = f"uv:{page}:{date_str}"
# PFCOUNT 返回近似计数值
return r.pfcount(key)
def merge_uv(page: str, start_date: str, end_date: str) -> int:
"""合并多天的 UV 统计（去重）"""
from datetime import datetime, timedelta
keys = []
current = datetime.strptime(start_date, '%Y-%m-%d')
end = datetime.strptime(end_date, '%Y-%m-%d')
while current <= end:
keys.append(f"uv:{page}:{current.strftime('%Y-%m-%d')}")
current += timedelta(days=1)
# PFMERGE 合并多个 HyperLogLog
dest = f"uv:{page}:{start_date}_to_{end_date}"
r.pfmerge(dest, *keys)
return r.pfcount(dest)
🛡️ 二、缓存三大问题：穿透、击穿、雪崩
缓存用不好，比不用更危险。缓存穿透、击穿、雪崩是生产环境中最常见的三大问题，也是面试高频考点——但更重要的是，它们真的会搞崩你的服务。
2.1 缓存穿透：查不存在的数据
攻击者用大量不存在的 key（如 user:-1）频繁请求，每次都穿透缓存直达数据库。
// ✅ 方案一：缓存空值（简单有效，适用于 key 数量有限的场景）
async function getUserById(userId) {
const cacheKey = `user:${userId}`;
const cached = await redis.get(cacheKey);
if (cached === 'NULL') return null; // 命中空值缓存
if (cached) return JSON.parse(cached); // 命中正常缓存
const user = await db.query('SELECT * FROM users WHERE id = ?', [userId]);
if (user) {
await redis.set(cacheKey, JSON.stringify(user), 'EX', 3600);
} else {
// ⚠️ 空值缓存时间要短，防止大量无效 key 占用内存
await redis.set(cacheKey, 'NULL', 'EX', 60);
}
return user;
}
// ✅ 方案二：布隆过滤器（适用于 key 空间极大的场景）
// 使用 Redis 的 RedisBloom 模块
async function checkUserExists(userId) {
// BF.EXISTS 检查元素是否可能存在
// 返回 0 = 一定不存在，1 = 可能存在（有 0.1% 误判率）
const exists = await redis.call('BF.EXISTS', 'users_bloom', `user:${userId}`);
return exists === 1;
}
// 初始化布隆过滤器（只需执行一次）
// BF.RESERVE key error_rate capacity
await redis.call('BF.RESERVE', 'users_bloom', 0.001, 1000000);
⚠️ 警告： 布隆过滤器不支持删除元素。如果需要删除，使用 Cuckoo Filter（
CF.RESERVE/CF.DEL），但内存占用会更大。
2.2 缓存击穿：热点 key 过期瞬间
一个热点 key（如秒杀商品信息）过期的瞬间，大量并发请求同时查数据库，造成数据库瞬时压力飙升。
// ✅ 解决方案：互斥锁（Mutex Lock）+ 逻辑过期
async function getHotProduct(productId) {
const cacheKey = `product:${productId}`;
const cached = await redis.get(cacheKey);
if (cached) {
const data = JSON.parse(cached);
// 检查逻辑过期时间
if (data.expireTime > Date.now()) {
return data.value; // 未过期，直接返回
}
// 已过期，尝试获取锁异步更新
const lockKey = `lock:${cacheKey}`;
const locked = await redis.set(lockKey, '1', 'NX', 'EX', 10);
if (locked) {
// 获取到锁，异步更新缓存
updateCacheAsync(productId, cacheKey, lockKey);
}
// 无论是否获取到锁，都先返回旧数据（stale data）
// 这是缓存击穿的核心：宁可返回旧数据，也不能让所有请求打到数据库
return data.value;
}
// 缓存完全不存在，走数据库查询
const product = await db.query('SELECT * FROM products WHERE id = ?', [productId]);
await setCacheWithLogicExpire(cacheKey, product, 300); // 逻辑过期 5 分钟
return product;
}
async function updateCacheAsync(productId, cacheKey, lockKey) {
try {
const product = await db.query('SELECT * FROM products WHERE id = ?', [productId]);
await setCacheWithLogicExpire(cacheKey, product, 300);
} finally {
await redis.del(lockKey); // 释放锁
}
}
2.3 缓存雪崩：大面积 key 同时过期
缓存雪崩是穿透和击穿的"放大版"——大量 key 在同一时间过期，或者 Redis 节点宕机，导致请求全部打到数据库。
预防策略：
| 策略 | 实现方式 | 适用场景 |
|---|---|---|
| 过期时间加随机值 | TTL = base_ttl + random(0, 300) |
防止同时过期 |
| 多级缓存 | 本地缓存（Caffeine）+ Redis + DB | 高可用要求高 |
| 熔断降级 | Hystrix / Sentinel | Redis 不可用时 |
| Redis Cluster | 主从 + 哨兵 / Cluster 模式 | 防止单点故障 |
// 过期时间加随机扰动，防止雪崩
function setWithRandomTTL(key, value, baseTTL) {
// 在基础 TTL 上加 0-10% 的随机偏移
const randomOffset = Math.floor(baseTTL * 0.1 * Math.random());
const ttl = baseTTL + randomOffset;
return redis.set(key, JSON.stringify(value), 'EX', ttl);
}
🔐 三、分布式锁：Redis 最容易用错的功能
分布式锁是 Redis 最常见的应用场景之一，但也是最容易写错的。一个漏洞百出的分布式锁，比没有锁更危险。
3.1 错误的分布式锁实现
// ❌ 错误写法：存在严重问题
async function acquireLock(key, ttl) {
const result = await redis.set(key, 'locked', 'EX', ttl, 'NX');
return result === 'OK';
}
async function releaseLock(key) {
await redis.del(key); // ❌ 可能释放了别人的锁！
}
⚠️ 警告： 上面的代码存在两个致命问题：(1) 释放锁时没有验证锁的归属，可能释放其他客户端的锁；(2) 如果业务执行时间超过 TTL，锁自动过期后仍会执行删除操作。
3.2 正确的分布式锁实现
// ✅ 正确写法：使用唯一标识 + Lua 脚本保证原子性
const { v4: uuidv4 } = require('uuid');
class RedisLock {
constructor(redis, key, ttl = 10000) {
this.redis = redis;
this.key = `lock:${key}`;
this.ttl = ttl;
this.token = null;
}
async acquire(retryCount = 3, retryDelay = 200) {
this.token = uuidv4();
for (let i = 0; i < retryCount; i++) {
const result = await this.redis.set(
this.key, this.token, 'PX', this.ttl, 'NX'
);
if (result === 'OK') return true;
// 等待后重试
await new Promise(r => setTimeout(r, retryDelay));
}
return false;
}
async release() {
// Lua 脚本：原子性地检查 + 删除
// 只有当锁的值等于自己的 token 时才删除
const luaScript = `
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end
`;
const result = await this.redis.eval(
luaScript, 1, this.key, this.token
);
return result === 1;
}
}
// 使用示例
async function transferMoney(fromId, toId, amount) {
const lock = new RedisLock(redis, `transfer:${fromId}:${toId}`, 5000);
try {
const acquired = await lock.acquire();
if (!acquired) {
throw new Error('获取锁失败，请重试');
}
// 执行转账业务逻辑
const balance = await db.getBalance(fromId);
if (balance < amount) throw new Error('余额不足');
await db.transfer(fromId, toId, amount);
} finally {
await lock.release(); // 必须在 finally 中释放
}
}
📌 记住： 生产环境建议直接使用 Redisson（Java）或 Redlock 算法的成熟实现，不要自己造轮子。Redis 的作者 Antirez 提出的 Redlock 算法虽然有争议（Martin Kleppmann 曾发文质疑），但在大多数业务场景下已经足够安全。
3.3 Redlock vs 单节点锁
| 特性 | 单节点 + Lua | Redlock（多节点） |
|---|---|---|
| 复杂度 | 低 | 高 |
| 可用性 | 依赖单节点 | 节点故障仍可用 |
| 安全性 | 一般（时钟漂移风险） | 较高 |
| 性能 | 1 次网络往返 | N 次网络往返 |
| 适用场景 | 大部分业务场景 | 强一致性要求的金融场景 |
| 推荐程度 | ✅ 推荐 | ⚠️ 仅在必要时使用 |
💡 四、生产环境避坑清单
经过多个项目的踩坑经验，总结以下高频问题：
1. 大 Key 问题
# Redis 大 key 定义：
# String 类型：值 > 10 KB
# Hash/Set/ZSet：元素数 > 5000 或总大小 > 10 MB
大 key 会导致：阻塞其他请求、主从同步延迟、内存不均衡。使用 redis-cli --bigkeys 扫描，用 UNLINK（异步删除）替代 DEL。
2. 热点 Key 问题
单个 key QPS 过高时，单个分片承受所有压力。解决方案：
- 本地缓存（Caffeine/Guava）做一级缓存
- 读写分离，读请求走从节点
- key 分片：将
hot_key拆成hot_key:0~hot_key:N，随机读取
3. Pipeline 批量操作
// ❌ 循环中逐条执行，N 次网络往返
for (const id of userIds) {
await redis.get(`user:${id}`);
}
// ✅ 使用 Pipeline，1 次网络往返
const pipeline = redis.pipeline();
for (const id of userIds) {
pipeline.get(`user:${id}`);
}
const results = await pipeline.exec();
4. 连接池配置
# 推荐的 Redis 连接池配置（Spring Boot application.yml）
spring:
data:
redis:
lettuce:
pool:
max-active: 16 # 最大连接数（建议 CPU 核数 * 2）
max-idle: 8 # 最大空闲连接
min-idle: 2 # 最小空闲连接
max-wait: 2000ms # 获取连接最大等待时间
timeout: 3000ms # 命令超时时间
✅ 总结
Redis 的价值不在于它"快"，而在于它提供了丰富的数据结构来解决特定场景的问题。选对数据结构、理解缓存策略、规避常见陷阱，才能在生产环境中发挥 Redis 的真正威力。
工具推荐：
- 🔧 Redis Insight — 官方可视化管理工具
- 🔧 redis-cli — 命令行调试利器
- 🔧 jsjson.com JSON 工具 — 序列化/反序列化 JSON 数据，配合 Redis 调试
- 📖 Redis 命令参考 — 官方文档
⚡ 关键结论： 缓存的核心不是"把数据放内存"，而是"在一致性、可用性和性能之间找到适合业务的平衡点"。没有银弹，只有 trade-off。