Redis 进阶实战
哨兵集群
Sentinel 架构
Redis Sentinel（哨兵）是 Redis 官方提供的高可用（HA）解决方案，其核心架构由一个或多个 Sentinel 节点与 Redis 主从节点共同组成。Sentinel 的主要职责包括：
- 监控（Monitoring）：持续检查主节点和从节点是否正常运行。
- 通知（Notification）：当被监控的 Redis 实例出现问题时，通过 API 通知系统管理员或其他程序。
- 自动故障转移（Automatic Failover）：当主节点不可用时，自动将一个从节点升级为新的主节点，并通知其他从节点复制新的主节点。
- 配置提供（Configuration Provider）：客户端连接 Sentinel 获取当前主节点的地址。
一个典型的 Sentinel 部署架构如下：
+-----------+
| Client |---+
+-----------+ |
v
+---------------+
| Sentinel 1 |---+
+---------------+ |
v
+---------------+ |
| Sentinel 2 |---+
+---------------+ |
v
+---------------+ |
| Sentinel 3 |
+---------------+
+-------+ +-------+
|Master |---------| Slave |
+-------+ +-------+
|
v
+-------+
| Slave |
+-------+推荐至少部署 3 个奇数个 Sentinel 节点，以保证 Leader 选举能够正常达成多数派。
主观下线与客观下线
Sentinel 通过两个阶段判定节点是否宕机：
主观下线（Subjectively Down，SDOWN）
单个 Sentinel 节点通过定期发送 PING 命令检测节点状态。若在 down-after-milliseconds 时间内未收到有效回复，该 Sentinel 将该节点标记为主观下线。
sentinel monitor mymaster 127.0.0.1 6379 2
sentinel down-after-milliseconds mymaster 30000客观下线（Objectively Down，ODOWN）
当 Sentinel 将一个主节点标记为 SDOWN 后，它会通过 sentinel is-master-down-by-addr 命令向其他 Sentinel 节点询问对该主节点的状态判断。当收到足够数量的确认（超过 quorum 值）后，该主节点被标记为客观下线（ODOWN），此时触发故障转移流程。
Leader 选举
当主节点被标记为 ODOWN 后，Sentinel 集群需要通过 Raft 算法 选举出一个 Leader 来执行故障转移操作：
- 发起选举：每个检测到 ODOWN 的 Sentinel 都会向其他节点发送竞选请求，增加自己的
epoch（任期号）。 - 投票：每个 Sentinel 在每个 epoch 内只能投一票，先到先得。获得超过半数（
n/2 + 1）投票的 Sentinel 成为 Leader。 - 执行故障转移：Leader 选出一个健康的从节点（优先级高、数据最新、运行稳定的从节点），执行
slaveof no one将其提升为主节点。 - 重新配置：Leader 通知其他从节点复制新的主节点，并更新 Sentinel 集群中的主节点信息。
bash
# Sentinel 日志示例（故障转移过程）
+ sdown master mymaster 127.0.0.1 6379
+ odown master mymaster 127.0.0.1 6379 #quorum 2/2
+ new-epoch 1
+ try-failover master mymaster 127.0.0.1 6379
+ vote-for-leader cb7818a7006f7aa76fbf58a9c2b9e8f75e6b2b5f 1
+ elected-leader master mymaster 127.0.0.1 6379
+ failover-state-select-slave master mymaster 127.0.0.1 6379
+ selected-slave slave 127.0.0.1:6380 127.0.0.1 6380 @ mymaster 127.0.0.1 6379
+ promoted-slave slave 127.0.0.1:6380 127.0.0.1 6380 @ mymaster 127.0.0.1 6379
+ switch-master mymaster 127.0.0.1 6379 127.0.0.1 6380客户端连接
客户端不应直接连接 Redis 主节点地址，而是通过 Sentinel 获取当前主节点信息：
bash
# 使用 redis-cli 通过 Sentinel 获取主节点
redis-cli -p 26379 SENTINEL get-master-addr-by-name mymaster
1) "127.0.0.1"
2) "6380"Java 客户端（Lettuce/Jedis）均已内置 Sentinel 支持：
java
// Jedis Sentinel 连接示例
Set<String> sentinels = new HashSet<>();
sentinels.add("127.0.0.1:26379");
sentinels.add("127.0.0.1:26380");
sentinels.add("127.0.0.1:26381");
JedisSentinelPool pool = new JedisSentinelPool("mymaster", sentinels);
try (Jedis jedis = pool.getResource()) {
jedis.set("key", "value");
String value = jedis.get("key");
}Spring Boot 配置
在 Spring Boot 中集成 Redis Sentinel 极为简便：
yaml
# application.yml
spring:
redis:
sentinel:
master: mymaster
nodes:
- 127.0.0.1:26379
- 127.0.0.1:26380
- 127.0.0.1:26381
# 可选：连接池配置
lettuce:
pool:
max-active: 16
max-idle: 8
min-idle: 4java
@Configuration
public class RedisConfig {
@Bean
public RedisTemplate<String, Object> redisTemplate(
RedisConnectionFactory connectionFactory) {
RedisTemplate<String, Object> template = new RedisTemplate<>();
template.setConnectionFactory(connectionFactory);
template.setKeySerializer(new StringRedisSerializer());
template.setValueSerializer(new GenericJackson2JsonRedisSerializer());
return template;
}
}Spring Boot 自动通过 Sentinel 获取主节点地址，并在故障转移发生后自动更新连接。
缓存穿透
定义
缓存穿透是指客户端查询一个数据库中也不存在的数据，导致请求绕过缓存，直接打到数据库层。由于缓存中永远不会有该数据的副本，每次请求都会穿透到数据库，在高并发场景下可能压垮数据库。
原因
- 恶意攻击：攻击者构造大量不存在的 Key 发起请求。
- 业务 Bug：查询参数校验不严，导致非法的 Key 进入查询流程。
- 数据未同步：数据已被删除，但缓存未及时清理，且数据库中也不存在。
正常情况下，请求会先查缓存 → 缓存命中则返回，否则查询数据库 → 数据库存在则回写缓存。但在缓存穿透的情况下，数据库也不存在该记录，因此缓存中一直没有数据，每次请求都直接查询数据库。
解决方案一：布隆过滤器
布隆过滤器（Bloom Filter）是一种空间效率极高的概率型数据结构，用于判断一个元素是否在集合中。它有以下特点：
- 判断不存在时 100% 准确：如果布隆过滤器说元素不存在，则元素一定不存在。
- 判断存在时有误判率：如果布隆过滤器说元素存在，可能不存在（假阳性）。
- 无法删除元素：标准布隆过滤器不支持删除操作。
原理：使用一个很长的二进制位数组和多个哈希函数。插入元素时，用多个哈希函数计算出多个位索引，将这些位设为 1。查询时，同样计算多个位索引，如果所有位都是 1，则认为可能存在；如果任意一位为 0，则一定不存在。
java
import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnels;
import java.nio.charset.StandardCharsets;
public class BloomFilterExample {
// 预计插入 100 万条数据，误判率 1%
private static final BloomFilter<String> bloomFilter =
BloomFilter.create(
Funnels.stringFunnel(StandardCharsets.UTF_8),
1_000_000,
0.01
);
static {
// 预热：将存在的 ID 初始化到布隆过滤器中
bloomFilter.put("user:1001");
bloomFilter.put("user:1002");
bloomFilter.put("user:1003");
}
public User getUserById(String userId) {
// 1. 布隆过滤器前置校验
if (!bloomFilter.mightContain(userId)) {
return null; // 一定不存在，直接返回
}
// 2. 查缓存
User user = cache.get(userId);
if (user != null) return user;
// 3. 查数据库
user = database.get(userId);
if (user != null) {
cache.put(userId, user);
}
return user;
}
}大型系统中的布隆过滤器实践：
java
@Component
public class RedisBloomFilter {
@Autowired
private RedisTemplate<String, String> redisTemplate;
/**
* 使用 Redis BitMap 实现简易布隆过滤器
*/
public boolean mightContain(String key, String element) {
int[] offsets = hash(element);
for (int offset : offsets) {
if (!redisTemplate.opsForValue()
.getBit(key, offset)) {
return false;
}
}
return true;
}
public void put(String key, String element) {
int[] offsets = hash(element);
for (int offset : offsets) {
redisTemplate.opsForValue()
.setBit(key, offset, true);
}
}
private int[] hash(String element) {
// 使用多个哈希函数（如 MurmurHash、FNV 等）
return new int[] {
Math.abs(element.hashCode() % (1 << 20)),
Math.abs(element.hashCode() * 31 % (1 << 20))
};
}
}解决方案二：缓存空对象
当数据库查询结果为空时，仍然将该空结果（null 或特殊标记）写入缓存，并设置一个较短的过期时间（如 30~60 秒）。这样后续相同请求可以直接返回空值，避免穿透到数据库。
java
public User getUserById(String userId) {
// 1. 查缓存
Object cacheResult = cache.get(userId);
if (cacheResult != null) {
// 缓存命中
if ("NULL".equals(cacheResult)) {
return null; // 空对象标记
}
return (User) cacheResult;
}
// 2. 查数据库
User user = database.get(userId);
if (user == null) {
// 3. 缓存空对象，有效期 30 秒
cache.set(userId, "NULL", 30);
return null;
}
// 4. 缓存真实数据
cache.set(userId, user, 3600);
return user;
}| 对比项 | 布隆过滤器 | 缓存空对象 |
|---|---|---|
| 额外内存 | 较低（位数组） | 较高（大量空 Key） |
| 实现复杂度 | 较高 | 较低 |
| 穿透保护 | 零穿透（不存在的一定拦截） | 短时间窗口内有效 |
| 适用场景 | 数据量大、Key 固定范围 | 数据量小、Key 不固定 |
| 弊端 | 有误判率、不支持删除 | 空 Key 占用内存、过期后仍有穿透风险 |
缓存击穿
定义
缓存击穿是指一个热点 Key 在缓存过期的瞬间，大量并发请求同时涌入，由于缓存中没有数据，所有请求都落到数据库上，导致数据库压力激增甚至崩溃。
区别于缓存穿透，缓存击穿中的 Key 在数据库中是真实存在的，问题仅发生在缓存过期的那个瞬间。
原因
- 热点 Key 过期：某个高并发访问的热点数据缓存到期。
- 集中失效：多个热点 Key 在同一时间过期。
解决方案一：互斥锁（Mutex Lock）
利用分布式锁，保证只有一个线程去查询数据库并重建缓存，其他线程等待该线程完成后从缓存获取数据。
java
@Autowired
private RedisTemplate<String, Object> redisTemplate;
private static final String LOCK_PREFIX = "lock:";
private static final long LOCK_EXPIRE = 10; // 锁过期时间，秒
public String getData(String key) {
// 1. 尝试从缓存获取数据
String data = (String) redisTemplate.opsForValue().get(key);
if (data != null) {
return data;
}
// 2. 缓存未命中，尝试获取分布式锁
String lockKey = LOCK_PREFIX + key;
String requestId = UUID.randomUUID().toString();
Boolean locked = redisTemplate.opsForValue()
.setIfAbsent(lockKey, requestId, LOCK_EXPIRE, TimeUnit.SECONDS);
if (Boolean.TRUE.equals(locked)) {
// 3. 成功获取锁，查询数据库（双重检查）
try {
// 双重检查：获取锁后再次检查缓存
data = (String) redisTemplate.opsForValue().get(key);
if (data != null) {
return data;
}
// 查询数据库
data = queryFromDatabase(key);
// 回写缓存，设置过期时间
redisTemplate.opsForValue().set(key, data, 3600, TimeUnit.SECONDS);
return data;
} finally {
// 4. 释放锁（使用 Lua 脚本保证原子性）
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then " +
"return redis.call('del', KEYS[1]) " +
"else return 0 end";
redisTemplate.execute(
new DefaultRedisScript<>(script, Long.class),
List.of(lockKey),
requestId
);
}
} else {
// 5. 未获取到锁，等待重试
try {
Thread.sleep(50);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
// 递归重试
return getData(key);
}
}解决方案二：逻辑过期
不设置物理过期时间，而是在缓存数据中嵌入一个逻辑过期时间字段。当读取时发现逻辑过期，由单个线程异步更新缓存，其他线程仍然返回旧数据。
java
@Data
@AllArgsConstructor
@NoArgsConstructor
public class CacheData<T> {
private T data;
private long expireTime; // 逻辑过期时间戳
}
@Component
public class LogicalExpireCache {
@Autowired
private RedisTemplate<String, CacheData<Object>> redisTemplate;
// 异步更新线程池
private static final ExecutorService UPDATE_POOL =
Executors.newFixedThreadPool(10);
private static final String LOCK_PREFIX = "lock:";
@SuppressWarnings("unchecked")
public <T> T getWithLogicalExpire(String key,
Class<T> type,
long expireSeconds,
Supplier<T> dbLoader) {
// 1. 获取缓存数据
CacheData<Object> cacheData =
redisTemplate.opsForValue().get(key);
if (cacheData == null) {
// 2. 数据不存在，加锁同步加载
String lockKey = LOCK_PREFIX + key;
Boolean locked = redisTemplate.opsForValue()
.setIfAbsent(lockKey, "1", 10, TimeUnit.SECONDS);
if (Boolean.TRUE.equals(locked)) {
try {
// 双重检查
cacheData = redisTemplate.opsForValue().get(key);
if (cacheData != null) {
return (T) cacheData.getData();
}
// 加载数据库
T data = dbLoader.get();
cacheData = new CacheData<>(
data, System.currentTimeMillis() + expireSeconds * 1000
);
redisTemplate.opsForValue().set(key, cacheData);
return data;
} finally {
redisTemplate.delete(lockKey);
}
} else {
// 等待后重试
try {
Thread.sleep(50);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return getWithLogicalExpire(key, type, expireSeconds, dbLoader);
}
}
// 3. 数据存在，检查是否逻辑过期
T data = (T) cacheData.getData();
if (cacheData.getExpireTime() > System.currentTimeMillis()) {
// 未过期，直接返回
return data;
}
// 4. 逻辑过期，尝试获取锁异步更新
String lockKey = LOCK_PREFIX + key;
Boolean locked = redisTemplate.opsForValue()
.setIfAbsent(lockKey, "1", 10, TimeUnit.SECONDS);
if (Boolean.TRUE.equals(locked)) {
UPDATE_POOL.submit(() -> {
try {
// 异步加载数据库
T newData = dbLoader.get();
CacheData<Object> newCacheData = new CacheData<>(
newData,
System.currentTimeMillis() + expireSeconds * 1000
);
redisTemplate.opsForValue().set(key, newCacheData);
} finally {
redisTemplate.delete(lockKey);
}
});
}
// 5. 返回旧数据
return data;
}
}| 对比项 | 互斥锁 | 逻辑过期 |
|---|---|---|
| 一致性 | 强一致（等待最新数据） | 弱一致（可能读到旧数据） |
| 性能 | 较差（有等待） | 较好（无需等待） |
| 实现复杂度 | 中等 | 较高 |
| 适用场景 | 对一致性要求高的场景 | 容忍短暂不一致的高并发场景 |
缓存雪崩
定义
缓存雪崩是指在同一时间段大量缓存 Key 同时失效，或者缓存节点宕机，导致所有请求直接打到数据库层，造成数据库压力陡增甚至宕机，进而引发系统级联故障。
原因
- 大量 Key 设置了相同的过期时间（如每天 0 点统一刷新）。
- Redis 节点宕机，整个缓存层不可用。
- 缓存服务重启后，缓存中无数据，大量请求穿透到数据库。
解决方案一：随机过期时间
为每个 Key 的过期时间增加一个随机偏移量，避免 Key 集中失效。
java
public void setRandomExpire(String key, String value, long baseExpire) {
// 在基础过期时间上增加 ±300 秒的随机偏移
long randomOffset = ThreadLocalRandom.current().nextLong(-300, 301);
long expireTime = Math.max(1, baseExpire + randomOffset);
redisTemplate.opsForValue().set(key, value, expireTime, TimeUnit.SECONDS);
}批量操作的改进：
java
public void batchCacheWithRandomExpiry(Map<String, String> dataMap,
long baseExpire) {
for (Map.Entry<String, String> entry : dataMap.entrySet()) {
setRandomExpire(entry.getKey(), entry.getValue(), baseExpire);
}
}解决方案二：多级缓存
引入本地缓存（如 Caffeine、Guava Cache）作为一级缓存，Redis 作为二级缓存，数据库作为三级存储，形成多层防护。
java
@Component
public class MultiLevelCache {
// 一级缓存：Caffeine 本地缓存
private final Cache<String, Object> localCache = Caffeine.newBuilder()
.maximumSize(10_000)
.expireAfterWrite(5, TimeUnit.MINUTES)
.recordStats()
.build();
@Autowired
private RedisTemplate<String, Object> redisTemplate;
public Object get(String key) {
// 1. 查本地缓存
Object localResult = localCache.getIfPresent(key);
if (localResult != null) {
return localResult;
}
// 2. 查 Redis 缓存
Object redisResult = redisTemplate.opsForValue().get(key);
if (redisResult != null) {
localCache.put(key, redisResult); // 回填本地缓存
return redisResult;
}
// 3. 查数据库
Object dbResult = queryDatabase(key);
if (dbResult != null) {
redisTemplate.opsForValue()
.set(key, dbResult, 1, TimeUnit.HOURS);
localCache.put(key, dbResult);
}
return dbResult;
}
}解决方案三：限流降级
使用限流组件（如 Sentinel、Hystrix、Resilience4j）保护数据库层，当缓存大量失效时，限制对数据库的并发访问量。
java
@Component
public class DegradedCache {
@Autowired
private RedisTemplate<String, Object> redisTemplate;
private final RateLimiter dbRateLimiter =
RateLimiter.create(200); // 每秒最多 200 个数据库请求
public Object getWithDegrade(String key) {
// 查缓存
Object result = redisTemplate.opsForValue().get(key);
if (result != null) {
return result;
}
// 缓存未命中，尝试获取限流许可
if (dbRateLimiter.tryAcquire(100, TimeUnit.MILLISECONDS)) {
// 有限流许可，查询数据库
Object dbResult = queryDatabase(key);
if (dbResult != null) {
redisTemplate.opsForValue()
.set(key, dbResult, 1, TimeUnit.HOURS);
}
return dbResult;
}
// 降级处理：返回默认值或错误提示
return getDefaultValue(key);
}
}完整的多层防御策略：
请求 → 限流降级 → 本地缓存 → Redis 缓存 → 数据库
↓ ↓ ↓ ↓ ↓
返回 拒绝/降级 命中返回 命中返回 查询后回填Redisson 高级
Redisson 是 Redis 官方推荐的 Java 客户端，提供了丰富的分布式数据结构和锁实现。以下介绍其核心高级特性。
看门狗机制
Redisson 的看门狗（Watchdog）机制解决了分布式锁的自动续期问题。当一个持有锁的线程还未完成业务操作但锁即将过期时，看门狗会自动为锁续期，避免业务未完成锁就被释放。
工作原理：
- 客户端获取锁时，默认锁的过期时间为 30 秒。
- Redisson 启动一个后台定时任务（看门狗），每隔 10 秒检查一次锁是否仍被当前线程持有。
- 如果锁仍然持有，看门狗自动将锁的过期时间重置为 30 秒。
- 如果持有锁的线程宕机，锁在 30 秒后自动释放，不会造成死锁。
java
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import java.util.concurrent.TimeUnit;
public class WatchdogExample {
public static void main(String[] args) {
Config config = new Config();
config.useSingleServer()
.setAddress("redis://127.0.0.1:6379");
RedissonClient redisson = Redisson.create(config);
RLock lock = redisson.getLock("myLock");
// 看门狗模式：
// 不指定 leaseTime 时，看门狗自动生效
lock.lock();
try {
// 模拟长时间业务操作（>30 秒）
Thread.sleep(60_000);
// 看门狗会自动续期，确保锁不会在业务执行期间过期
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
lock.unlock();
}
}
}参数控制：
java
// 1. 使用看门狗（默认 leaseTime = -1，触发看门狗）
lock.lock();
// 2. 手动指定过期时间（不触发看门狗）
lock.lock(10, TimeUnit.SECONDS);
// 3. 尝试获取锁，带等待时间 + 看门狗
boolean acquired = lock.tryLock(5, -1, TimeUnit.SECONDS);
// 4. 尝试获取锁，手动指定过期时间（不看门狗）
boolean acquired = lock.tryLock(5, 10, TimeUnit.SECONDS);可重入锁
Redisson 的锁是可重入的，即同一个线程可以多次获取同一把锁而不会死锁。每次获取锁时内部计数器加 1，每次解锁时减 1，当计数器归零时锁才真正释放。
java
public void reentrantExample() {
RLock lock = redisson.getLock("reentrantLock");
lock.lock();
try {
System.out.println("第一次获取锁");
// 同一线程再次获取同一把锁（可重入）
lock.lock();
try {
System.out.println("第二次获取锁（可重入）");
// 业务逻辑
} finally {
lock.unlock(); // 计数器减为 1
}
} finally {
lock.unlock(); // 计数器归零，锁真正释放
}
}可重入锁的存储结构（Redis Hash）：
Key: "reentrantLock"
Field: "UUID:threadId"
Value: 3（重入次数）读写锁
Redisson 的 RReadWriteLock 遵循经典的读写锁语义：
- 读锁 + 读锁：可以共存（共享锁），多个线程可以同时加读锁。
- 读锁 + 写锁：互斥（有写锁时，读锁必须等待）。
- 写锁 + 写锁：互斥。
java
public void readWriteLockExample() {
RReadWriteLock rwLock = redisson.getReadWriteLock("inventoryLock");
RLock readLock = rwLock.readLock();
RLock writeLock = rwLock.writeLock();
// 读操作：多个线程可以同时读取
readLock.lock();
try {
String inventory = redisTemplate.opsForValue()
.get("inventory").toString();
System.out.println("读取库存：" + inventory);
} finally {
readLock.unlock();
}
// 写操作：独占
writeLock.lock();
try {
// 更新库存
redisTemplate.opsForValue()
.set("inventory", "100");
System.out.println("更新库存成功");
} finally {
writeLock.unlock();
}
}信号量
Redisson 的 RSemaphore 基于 Redis 实现了分布式信号量，用于控制多个线程对共享资源的并发访问数量。
java
public void semaphoreExample() throws InterruptedException {
// 创建信号量，设置 3 个许可
RSemaphore semaphore = redisson.getSemaphore("mySemaphore");
// 初始化许可数量（只在第一次时设置）
semaphore.trySetPermits(3);
// 获取一个许可（阻塞直到有可用许可）
semaphore.acquire();
try {
System.out.println("获取到信号量许可，执行业务...");
Thread.sleep(1000);
} finally {
// 释放许可
semaphore.release();
}
// 非阻塞尝试获取许可
boolean acquired = semaphore.tryAcquire(1, TimeUnit.SECONDS);
if (acquired) {
try {
System.out.println("在 1 秒内获取到许可");
} finally {
semaphore.release();
}
}
}典型应用场景：控制同时访问某个 API 或资源的并发数、限流、资源池管理。
闭锁
Redisson 的 RCountDownLatch 对应 Java 的 CountDownLatch，但在分布式场景下可跨多个节点协调任务进度。
java
public void countDownLatchExample() throws InterruptedException {
String latchName = "distributedLatch";
RCountDownLatch latch = redisson.getCountDownLatch(latchName);
// 设置计数器为 3
latch.trySetCount(3);
// 启动三个工作线程
for (int i = 0; i < 3; i++) {
int taskId = i;
new Thread(() -> {
try {
// 模拟任务执行
Thread.sleep((long) (Math.random() * 3000));
System.out.println("任务 " + taskId + " 完成");
latch.countDown(); // 计数器减 1
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}).start();
}
System.out.println("等待所有任务完成...");
// 阻塞等待计数器归零
boolean completed = latch.await(10, TimeUnit.SECONDS);
if (completed) {
System.out.println("所有任务均已完成！");
} else {
System.out.println("等待超时，部分任务未完成");
}
}Redis 管道（Pipeline）
原理
Redis Pipeline（管道）是一种网络优化技术，允许客户端将多个命令一次性批量发送到服务端，而不需要等待每个命令的响应。传统模式下，每个命令都需要一次网络往返（RTT，Round Trip Time），而在 Pipeline 模式下，多个命令共享一次网络往返，大幅减少了网络延迟的影响。
传统模式（无 Pipeline）：
Client: SET key1 val1
Server: +OK
Client: GET key1
Server: val1
Client: SET key2 val2
Server: +OKPipeline 模式：
Client: SET key1 val1 GET key1 SET key2 val2
Server: +OK val1 +OK性能提升
性能提升量与以下因素相关：
- 网络延迟（RTT）：RTT 越高，Pipeline 的收益越大。
- 命令数量：批量的命令数量越多，收益越明显。
- Pipelining 批次大小：一次 Pipeline 发送的命令数（通常推荐 100~1000 条）。
假设 RTT = 10ms，单次命令执行时间 = 0.1ms：
| 命令数 | 无 Pipeline | Pipeline（100条/批） | 提升倍数 |
|---|---|---|---|
| 100 | 1,010 ms | 20 ms | 约 50 倍 |
| 1,000 | 10,100 ms | 110 ms | 约 92 倍 |
| 10,000 | 101,000 ms | 1,010 ms | 约 100 倍 |
代码示例
Jedis Pipeline：
java
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;
import redis.clients.jedis.Response;
import java.util.HashMap;
import java.util.Map;
public class JedisPipelineExample {
public void batchInsert() {
try (Jedis jedis = new Jedis("127.0.0.1", 6379)) {
Pipeline pipeline = jedis.pipelined();
// 批量写入
for (int i = 0; i < 10_000; i++) {
pipeline.set("batch:key:" + i, "value:" + i);
}
// 同步提交并获取响应
pipeline.sync();
}
}
public void batchReadWrite() {
try (Jedis jedis = new Jedis("127.0.0.1", 6379)) {
Pipeline pipeline = jedis.pipelined();
// 存储 Response 对象以便后续获取结果
Map<String, Response<String>> responses = new HashMap<>();
for (int i = 0; i < 100; i++) {
String key = "batch:key:" + i;
responses.put(key, pipeline.get(key));
}
// 同步执行
pipeline.sync();
// 获取结果
for (Map.Entry<String, Response<String>> entry : responses.entrySet()) {
String value = entry.getValue().get();
System.out.println(entry.getKey() + " = " + value);
}
}
}
}Spring RedisTemplate Pipeline：
java
@Autowired
private RedisTemplate<String, Object> redisTemplate;
public void batchOperations() {
List<Object> results = redisTemplate.executePipelined(
(RedisCallback<Object>) connection -> {
// 在这个回调内执行的所有命令都会被自动 Pipeline
for (int i = 0; i < 1000; i++) {
byte[] key = ("pipeline:key:" + i).getBytes();
byte[] value = ("value:" + i).getBytes();
connection.set(key, value);
}
// 注意：Pipeline 模式下不需要返回值
return null;
}
);
}
public List<String> batchGet(List<String> keys) {
List<Object> results = redisTemplate.executePipelined(
(RedisCallback<Object>) connection -> {
for (String key : keys) {
connection.get(key.getBytes());
}
return null;
}
);
// 转换结果
List<String> values = new ArrayList<>();
for (Object result : results) {
if (result instanceof byte[]) {
values.add(new String((byte[]) result));
} else {
values.add(null);
}
}
return values;
}注意事项：
- Pipeline 是非原子性的——执行过程中如果服务端返回错误，后续命令会继续执行。
- Pipeline 会占用服务端内存来缓存命令响应，一次批量的命令数量不宜过大（建议 100~1000 条）。
- Pipeline 适用于对一致性要求不高的批量操作场景。
Redis Lua 脚本
原子性
Redis 使用 Lua 脚本的核心优势在于原子性。Redis 服务器采用单线程执行 Lua 脚本，脚本运行期间不会执行其他任何命令，因此 Lua 脚本天然具备事务性——要么全部执行，要么全部不执行。
适用场景：
- 需要执行多个 Redis 命令并保证原子性。
- 需要读取当前值并根据条件作出修改（CAS 操作）。
- 需要实现 Redis 原生不直接支持的复杂逻辑。
EVAL / EVALSHA
EVAL：直接执行 Lua 脚本。
EVAL script numkeys key [key ...] arg [arg ...]bash
# 简单示例：SET 并返回旧值
EVAL "local old = redis.call('GET', KEYS[1]); redis.call('SET', KEYS[1], ARGV[1]); return old" 1 mykey newvalue
# 使用 Lua 实现原子性的「扣减库存并检查」
EVAL "
local stock = redis.call('GET', KEYS[1])
if not stock then
return -1 -- key 不存在
end
if tonumber(stock) < tonumber(ARGV[1]) then
return 0 -- 库存不足
end
redis.call('DECRBY', KEYS[1], ARGV[1])
return 1 -- 扣减成功
" 1 stock:1001 1EVALSHA：通过脚本的 SHA1 校验和来执行脚本，减少网络传输。
bash
# 1. 先使用 SCRIPT LOAD 加载脚本，获取 SHA
SCRIPT LOAD "return redis.call('GET', KEYS[1])"
> "4e6d5c0a8b1c3f2e7a9b8c0d1e2f3a4b5c6d7e8f"
# 2. 使用 EVALSHA 执行
EVALSHA 4e6d5c0a8b1c3f2e7a9b8c0d1e2f3a4b5c6d7e8f 1 mykey建议策略：先尝试 EVALSHA，如果脚本不存在（返回 NOSCRIPT 错误），再回退到 EVAL。
常见 Lua 脚本示例
1. 原子性扣减库存：
lua
-- 脚本：deduct_stock.lua
-- KEYS[1]: 库存 Key
-- ARGV[1]: 扣减数量
-- 返回值：-2 = 库存不足, -1 = Key 不存在, >=0 = 剩余库存
local stock = redis.call('GET', KEYS[1])
if not stock then
return -1
end
local remain = tonumber(stock) - tonumber(ARGV[1])
if remain < 0 then
return -2
end
redis.call('SET', KEYS[1], remain)
return remain2. 限流器（滑动窗口）：
lua
-- 脚本：rate_limiter.lua
-- KEYS[1]: 限流 Key
-- ARGV[1]: 窗口大小（毫秒）
-- ARGV[2]: 窗口内最大请求数
-- 返回值：1 = 允许, 0 = 限流
local window = tonumber(ARGV[1])
local maxRequests = tonumber(ARGV[2])
local now = redis.call('TIME')[1] * 1000 -- 当前毫秒时间戳
-- 移除窗口外的记录
redis.call('ZREMRANGEBYSCORE', KEYS[1], 0, now - window)
-- 统计当前窗口内的请求数
local current = redis.call('ZCARD', KEYS[1])
if current >= maxRequests then
return 0
end
-- 记录本次请求
redis.call('ZADD', KEYS[1], now, now)
redis.call('PEXPIRE', KEYS[1], window)
return 13. CAS（Compare And Swap）：
lua
-- 脚本：cas.lua
-- KEYS[1]: 目标 Key
-- ARGV[1]: 期望的旧值
-- ARGV[2]: 新值
-- 返回值：1 = 更新成功, 0 = 值不匹配
local current = redis.call('GET', KEYS[1])
if current == ARGV[1] then
redis.call('SET', KEYS[1], ARGV[2])
return 1
end
return 0Spring Redis 集成 Lua
java
@Component
public class LuaScriptExecutor {
@Autowired
private RedisTemplate<String, Object> redisTemplate;
/**
* 使用 Lua 脚本原子性扣减库存
*/
public long deductStock(String stockKey, int quantity) {
String script = """
local stock = redis.call('GET', KEYS[1])
if not stock then
return -1
end
local remain = tonumber(stock) - tonumber(ARGV[1])
if remain < 0 then
return -2
end
redis.call('SET', KEYS[1], remain)
return remain
""";
DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
redisScript.setScriptText(script);
redisScript.setResultType(Long.class);
Long result = redisTemplate.execute(
redisScript,
List.of(stockKey),
String.valueOf(quantity)
);
return result != null ? result : -1;
}
/**
* 从文件中加载 Lua 脚本
*/
public void loadScriptFromFile() {
DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
// 从 classpath 加载 lua 文件
redisScript.setLocation(new ClassPathResource("scripts/deduct_stock.lua"));
redisScript.setResultType(Long.class);
// 预先加载脚本以获得 SHA（后续可使用 EVALSHA）
String sha = redisTemplate.execute(
(RedisCallback<String>) connection ->
connection.scriptLoad(redisScript.getScriptAsString().getBytes())
);
System.out.println("Script SHA: " + sha);
}
/**
* 使用 EVALSHA 执行
*/
public Object executeWithSha(String sha, List<String> keys, Object... args) {
try {
return redisTemplate.execute(
(RedisCallback<Object>) connection ->
connection.evalSha(sha.getBytes(),
ReturnType.INTEGER,
keys.size(),
keys.toArray(new String[0]),
Arrays.stream(args)
.map(Object::toString)
.toArray(String[]::new)
)
);
} catch (Exception e) {
// NOSCRIPT 错误时回退到 EVAL
if (e.getMessage() != null &&
e.getMessage().contains("NOSCRIPT")) {
// 重新加载并执行
return null; // 实际生产环境应重新加载脚本
}
throw e;
}
}
}最佳实践：
- 将 Lua 脚本编译为
.lua文件放在资源目录，使用ClassPathResource加载。 - 使用
SCRIPT LOAD预加载脚本，后续使用EVALSHA减少带宽消耗。 - 脚本应尽量简短，避免长时间阻塞 Redis 单线程。
- 脚本中的 Key 应通过参数传递，不要硬编码在脚本中——这会影响 Redis 集群的哈希分片。
Redis 慢查询
SLOWLOG 配置
Redis 慢查询日志用于记录执行时间超过指定阈值的命令，帮助开发者定位性能瓶颈。
核心配置项：
bash
# 在 redis.conf 中配置
# 执行时间超过 10 毫秒的命令将被记录（单位：微秒）
slowlog-log-slower-than 10000
# 最多保存 128 条慢查询日志（队列长度）
slowlog-max-len 128运行时动态修改：
bash
# 设置慢查询阈值（微秒）
127.0.0.1:6379> CONFIG SET slowlog-log-slower-than 10000
OK
# 设置慢查询日志长度
127.0.0.1:6379> CONFIG SET slowlog-max-len 128
OK
# 查看当前配置
127.0.0.1:6379> CONFIG GET slowlog-log-slower-than
1) "slowlog-log-slower-than"
2) "10000"
127.0.0.1:6379> CONFIG GET slowlog-max-len
1) "slowlog-max-len"
2) "128"查看和分析慢查询：
bash
# 获取最新的 10 条慢查询日志
127.0.0.1:6379> SLOWLOG GET 10
1) 1) (integer) 3 # 日志唯一 ID
2) (integer) 1719487200 # 时间戳
3) (integer) 15230 # 执行耗时（微秒）
4) 1) "KEYS" # 执行的命令
2) "user:*" # 命令参数
# 获取慢查询日志数
127.0.0.1:6379> SLOWLOG LEN
(integer) 3
# 清空慢查询日志
127.0.0.1:6379> SLOWLOG RESET
OK分析与优化
慢查询日志输出字段说明：
| 字段 | 说明 |
|---|---|
| ID | 慢查询日志的唯一标识，自增 |
| 时间戳 | 命令执行时的 Unix 时间戳（秒） |
| 耗时 | 命令执行耗时，单位微秒（μs） |
| 命令及参数 | 执行的命令和完整的参数 |
| 客户端 IP:Port | 发送命令的客户端地址 |
| 客户端名称 | 客户端的名称（若有设置） |
常见慢查询原因及优化策略：
bash
# 1. 禁止在生产环境使用 KEYS 命令
# 改为使用 SCAN 命令代替
127.0.0.1:6379> SCAN 0 MATCH user:* COUNT 100
# 2. 大集合操作的优化
# 避免使用 SMEMBERS 获取大集合所有元素
# 改为使用 SSCAN
127.0.0.1:6379> SSCAN myset 0 COUNT 100
# 3. 大量数据的排序/聚合操作
# 将大数据量的排序操作拆分或异步执行
# 4. 复杂 Lua 脚本
# 确保 Lua 脚本执行时间短，避免长时间占用单线程
# 5. 使用批量操作减少网络 RTT
# Pipeline 或 mget/mset监控与分析脚本：
bash
#!/bin/bash
# 慢查询分析脚本
REDIS_CLI="redis-cli -h 127.0.0.1 -p 6379"
echo "=== Redis 慢查询分析 ==="
echo "慢查询阈值: $($REDIS_CLI CONFIG GET slowlog-log-slower-than)"
echo "慢查询队列长度: $($REDIS_CLI CONFIG GET slowlog-max-len)"
echo ""
echo "=== 最近的慢查询 ==="
$REDIS_CLI --raw SLOWLOG GET 20 | awk '
NR%4==1 {id=$0}
NR%4==2 {ts=$0; cmd=strftime("%Y-%m-%d %H:%M:%S", $0)}
NR%4==3 {cost_us=$0; cost_ms=$0/1000}
NR%4==0 {printf "[%s] %s (%.2f ms): %s\n", cmd, id, cost_ms, $0}
'Big Key 问题
定义
Big Key 是指 Redis 中某个 Key 的值占用内存过大或包含的元素数量过多，通常超过以下阈值：
| 数据类型 | Big Key 阈值 |
|---|---|
| String | 值大小 > 10 KB |
| Hash | 字段数量 > 10,000 |
| List | 元素数量 > 10,000 |
| Set | 元素数量 > 10,000 |
| Sorted Set | 元素数量 > 10,000 |
危害
- 阻塞网络：大 Key 的传输占用大量带宽，导致其他命令响应变慢。
GET/SET一个 10 MB 的 String 需要传输 10 MB 数据。 - 阻塞 Redis 单线程：操作大 Key 的命令（如
DEL、HGETALL、SMEMBERS）会阻塞 Redis 单线程，长时间无法处理其他请求。 - 内存不均衡：在 Redis 集群中，Big Key 会导致某个分片的内存使用远高于其他分片，造成内存分布不均。
- 数据迁移慢：集群扩缩容时，Big Key 需要更长的时间迁移，可能导致迁移超时。
bash
# Big Key 的典型危害示例：
# 对一个包含 500 万个元素的 List 执行 DEL 操作
127.0.0.1:6379> DEL biglist
# 这条命令会阻塞 Redis 数秒钟，期间所有其他请求都无法处理
# 获取包含 100 万元素的 Hash
127.0.0.1:6379> HGETALL hugehash
# 返回巨量数据，网络传输阻塞，占用大量带宽发现
方法一：redis-cli --bigkeys
使用 Redis 自带的分析工具扫描 Big Key：
bash
# 扫描所有 Key，输出最大的 Key 信息
redis-cli -h 127.0.0.1 -p 6379 --bigkeys
# 输出示例
# Scanning the entire keyspace to find biggest keys as well as
# average sizes per key type. You can use -i 0.1 to sleep 0.1 sec
# per 100 SCAN commands (not usually needed).
[00.00%] Biggest string found so far '"user:1001"' with 5 bytes
[12.50%] Biggest list found so far '"mylist"' with 100000 items
[25.00%] Biggest hash found so far '"myhash"' with 50000 fields
[37.50%] Biggest set found so far '"myset"' with 200000 members
-------- summary -------
Sampled 1000 keys in the keyspace!
Total key length in bytes is 12000 (avg len 12.00)
Biggest string found '"user:1001"' has 5 bytes
Biggest list found '"mylist"' has 100000 items
Biggest hash found '"myhash"' has 50000 fields
Biggest set found '"myset"' has 200000 members
1 largest strings with 5 bytes (0.00% of keys, avg size 5.00)
1 largest lists with 100000 items (0.10% of keys, avg size 500.00)
1 largest sets with 200000 members (0.10% of keys, avg size 500.00)方法二：SCAN + DEBUG OBJECT
使用 SCAN 渐进式遍历 Key，配合 DEBUG OBJECT 获取每个 Key 的详细信息，实现自定义的 Big Key 扫描。
bash
# 使用 Lua 脚本扫描 Big Key
EVAL "
local cursor = '0'
local big_keys = {}
repeat
local result = redis.call('SCAN', cursor, 'COUNT', 1000)
cursor = result[1]
for _, key in ipairs(result[2]) do
local info = redis.call('DEBUG OBJECT', key)
-- 解析 serializedlength
local len = tonumber(string.match(info, 'serializedlength:(%d+)'))
if len and len > 10240 then -- 大于 10 KB
table.insert(big_keys, key .. ' -> ' .. len .. ' bytes')
end
end
until cursor == '0'
return big_keys
" 0方法三：使用 MEMORY USAGE 命令（Redis 4.0+）
bash
# 查看 Key 的内存占用（单位：字节）
127.0.0.1:6379> MEMORY USAGE myhash
(integer) 1234567
# 批量检查
127.0.0.1:6379> MEMORY USAGE bigkey
(integer) 52428800 # 约 50 MB方法四：监控工具
- RedisInsight：Redis 官方 GUI 工具，可视化展示 Big Key。
- Prometheus + redis_exporter：监控 Key 的大小指标。
- RDB Tools：通过分析 RDB 文件发现 Big Key。
处理方案
方案一：拆分（Sharding）
将大 Key 拆分为多个小 Key，按照 Hash 或业务维度分片。
java
// 拆分前：一个 Hash 存储所有用户数据
// Key: "user:all" -> 100 万字段
// 拆分后：按用户 ID 哈希分片到 100 个 Hash
public class BigHashSplitter {
private static final int BUCKET_COUNT = 100;
public String getShardKey(String baseKey, String field) {
int bucket = Math.abs(field.hashCode()) % BUCKET_COUNT;
return baseKey + ":" + bucket;
}
public void setField(String baseKey, String field, String value) {
String shardKey = getShardKey(baseKey, field);
redisTemplate.opsForHash().put(shardKey, field, value);
}
public String getField(String baseKey, String field) {
String shardKey = getShardKey(baseKey, field);
return (String) redisTemplate.opsForHash().get(shardKey, field);
}
}方案二：冷热分离
将热数据保留在 Redis 中，冷数据迁移到其他存储（如 MySQL、MongoDB）。
java
public class HotColdSeparation {
public void writeData(String key, String hotData, String coldData) {
// 热数据保留在 Redis（设置过期时间）
redisTemplate.opsForValue()
.set(key + ":hot", hotData, 1, TimeUnit.HOURS);
// 冷数据写入数据库
database.insert(key + ":cold", coldData);
}
public String readData(String key) {
// 先读热数据
String hot = (String) redisTemplate.opsForValue()
.get(key + ":hot");
if (hot != null) return hot;
// 读冷数据
return database.query(key + ":cold");
}
}方案三：压缩存储
对于大 String，使用压缩算法减少内存占用。
java
public class CompressedCache {
public void setCompressed(String key, String value) {
try {
// 使用 GZIP 压缩
ByteArrayOutputStream bos = new ByteArrayOutputStream();
try (GZIPOutputStream gzip = new GZIPOutputStream(bos)) {
gzip.write(value.getBytes(StandardCharsets.UTF_8));
}
byte[] compressed = bos.toByteArray();
redisTemplate.opsForValue().set(key, compressed);
System.out.println("压缩比: " + (value.length() * 1.0 / compressed.length));
} catch (IOException e) {
throw new RuntimeException("压缩失败", e);
}
}
public String getCompressed(String key) {
byte[] compressed = (byte[]) redisTemplate.opsForValue().get(key);
if (compressed == null) return null;
try {
ByteArrayInputStream bis = new ByteArrayInputStream(compressed);
try (GZIPInputStream gzip = new GZIPInputStream(bis)) {
return new String(gzip.readAllBytes(), StandardCharsets.UTF_8);
}
} catch (IOException e) {
throw new RuntimeException("解压失败", e);
}
}
}方案四：渐进式删除
删除 Big Key 时使用 UNLINK 替代 DEL，避免阻塞 Redis 单线程。
bash
# 使用 UNLINK 异步删除（Redis 4.0+）
127.0.0.1:6379> UNLINK bigkey
(integer) 1
# UNLINK 在后台线程释放内存，立即返回
# 传统 DEL 会阻塞
127.0.0.1:6379> DEL bigkey
# 如果 bigkey 很大，会阻塞数秒Redis 内存淘汰策略
8 种淘汰策略对比
当 Redis 内存使用量达到 maxmemory 限制时，将根据配置的淘汰策略（maxmemory-policy）决定如何回收内存。
| 策略 | 名称 | 作用范围 | 说明 |
|---|---|---|---|
noeviction | 禁止淘汰 | 全部 | 内存满后写操作返回错误，读请求正常（默认策略） |
allkeys-lru | LRU（所有键） | 全部 Key | 淘汰最近最少使用的 Key |
allkeys-lfu | LFU（所有键） | 全部 Key | 淘汰最不经常使用的 Key（Redis 4.0+） |
allkeys-random | 随机（所有键） | 全部 Key | 随机淘汰 Key |
volatile-lru | LRU（带过期 Key） | 仅设置了 TTL 的 Key | 从设置了过期时间的 Key 中淘汰最近最少使用的 |
volatile-lfu | LFU（带过期 Key） | 仅设置了 TTL 的 Key | 从设置了过期时间的 Key 中淘汰最不经常使用的 |
volatile-random | 随机（带过期 Key） | 仅设置了 TTL 的 Key | 从设置了过期时间的 Key 中随机淘汰 |
volatile-ttl | TTL 优先 | 仅设置了 TTL 的 Key | 淘汰剩余存活时间最短的 Key |
bash
# 在 redis.conf 中配置
maxmemory 4gb
maxmemory-policy allkeys-lru
# 运行时修改
127.0.0.1:6379> CONFIG SET maxmemory 4gb
OK
127.0.0.1:6379> CONFIG SET maxmemory-policy allkeys-lru
OK
# 查看淘汰策略统计
127.0.0.1:6379> INFO stats
# ... evicted_keys:12345 # 被淘汰的 Key 数量LRU 近似算法
标准 LRU
标准 LRU（Least Recently Used）算法维护一个按访问时间排序的链表，每次淘汰时删除链表尾部的元素。但 Redis 并未采用标准 LRU，因为它需要维护一个全局链表，占用大量内存且移动操作成本高。
Redis 的近似 LRU
Redis 使用近似 LRU（Approximated LRU）算法，通过采样 + 淘汰来实现接近标准 LRU 的效果：
- 采样：从所有 Key 中随机取
maxmemory-samples个样本（默认 5 个）。 - 淘汰：淘汰样本中空闲时间（idle time）最长的 Key。
- 效率：时间复杂度 O(1)，内存开销极低。
bash
# 配置采样数量（值越大越接近标准 LRU，但 CPU 消耗更高）
maxmemory-samples 10Redis 近似 LRU vs 标准 LRU 对比：
准确度
标准 LRU ████████████████████ 100%
采样 5 ████████████████ 约 80%
采样 10 ██████████████████ 约 90%
采样 20 ███████████████████ 约 95%bash
# 通过 INFO 命令查看淘汰相关统计
127.0.0.1:6379> INFO stats
# Stats
evicted_keys:0 # 被淘汰的 Key 数量
keyspace_hits:1000 # 缓存命中次数
keyspace_misses:100 # 缓存未命中次数LFU 近似算法
LFU（Least Frequently Used，最不经常使用）是 Redis 4.0 引入的策略，通过统计 Key 的访问频次来决定淘汰对象。Redis 同样使用近似算法实现 LFU：
频率衰减机制：
- 每个 Key 的访问频率使用概率计数器（Morris Counter）近似统计，仅占用 3~4 位空间。
- 计数器会随时间对数衰减——长时间不访问的 Key 的计数逐渐降低，从而被优先淘汰。
bash
# LFU 相关配置（redis.conf）
# 计数器衰减速度（以分钟为单位）
# n = 1 表示每 1 分钟衰减一次
lfu-decay-time 1
# 计数器对数因子，影响计数器增长速率
# 默认 10，越大计数器增长越慢
lfu-log-factor 10LFU 工作原理示例：
Key A 在 1 小时内被访问 1000 次 → 计数器值 ≈ 50
Key B 在 1 小时内被访问 100 次 → 计数器值 ≈ 30
Key C 在 1 分钟内被访问 10 次 → 计数器值 ≈ 20
经过 1 小时后（假设所有 Key 未再被访问）：
Key A 衰减后 ≈ 40
Key B 衰减后 ≈ 20
Key C 衰减后 ≈ 10 ← 最不常用，将被优先淘汰策略选择建议
| 业务场景 | 推荐策略 | 理由 |
|---|---|---|
| 通用缓存（不区分热点） | allkeys-lru | 根据最近访问时间淘汰，简单有效 |
| 热点数据集中场景 | allkeys-lfu | 优先保留高频访问的 Key |
| 有明确过期时间的会话数据 | volatile-ttl 或 volatile-lru | 只淘汰带 TTL 的 Key，保留持久化数据 |
| 数据量可控，不愿丢失任何数据 | noeviction | 禁止淘汰，写满时报错（需配合业务告警） |
| 对淘汰无感，追求最大吞吐 | allkeys-random | 随机淘汰，实现代价最低 |
| 需要精确控制哪些 Key 可淘汰 | volatile-lru / volatile-lfu | 设置 TTL 的 Key 视为可淘汰，无 TTL 的保留 |
生产环境推荐配置：
bash
maxmemory 4gb
maxmemory-policy allkeys-lru
maxmemory-samples 10监控内存淘汰情况：
bash
# 实时监控被淘汰的 Key 数量
127.0.0.1:6379> INFO stats | grep evicted_keys
evicted_keys:0
# 如果 evicted_keys 持续快速增长，说明内存不足
# 需要扩容或优化缓存策略
# 查看内存使用率
127.0.0.1:6379> INFO memory
# Memory
used_memory:4294967296 # 已使用内存（字节）
used_memory_human:4.00G # 已使用内存（人类可读）
maxmemory:4294967296 # 最大内存限制
maxmemory_human:4.00G
used_memory_peak:5368709120 # 历史峰值（5 GB）