memo：2025 年 5 月 17 日，今天有球友发微信说拿到了一个国企子公司的 Java 后端开发和一个小米安卓的 offer，问我该怎么选择？

缓存设计

29.🌟什么是缓存击穿？

缓存击穿是指某个热点数据缓存过期时，大量请求就会穿透缓存直接访问数据库，导致数据库瞬间承受的压力巨大。

解决缓存击穿有两种常用的策略：

第一种是加互斥锁。当缓存失效时，第一个访问的线程先获取锁并负责重建缓存，其他线程等待或重试。

这种策略虽然会导致部分请求延迟，但实现起来相对简单。在技术派实战项目中，我们就使用了 Redisson 的分布式锁来确保只有一个服务实例能更新缓存。

String cacheKey = "product::" + productId;
RLock lock = redissonClient.getLock("lock::" + productId);
if (lock.tryLock(10, TimeUnit.SECONDS)) {
    try {
        String result = cache.get(cacheKey);
        if (result == null) {
            result = database.queryProductById(productId);
            cache.set(cacheKey, result, 60 * 1000); // 设置缓存
        }
    } finally {
        lock.unlock();
    }
}

第二种是永不过期策略。缓存项本身不设置过期时间，也就是永不过期，但在缓存值中维护一个逻辑过期时间。当缓存逻辑上过期时，返回旧值的同时，异步启动一个线程去更新缓存。

public String getData(String key) {
    CacheItem item = cache.get(key);
    
    if (item == null) {
        // 缓存不存在，同步加载
        String data = db.query(key);
        cache.set(key, new CacheItem(data, System.currentTimeMillis() + expireTime));
        return data;
    } else if (item.isLogicalExpired()) {
        // 逻辑过期，异步刷新
        asyncRefresh(key);
        // 返回旧数据
        return item.getData();
    }
    
    return item.getData();
}

// 异步刷新缓存
private void asyncRefresh(final String key) {
    threadPool.execute(() -> {
        // 重新查询数据库
        String newData = db.query(key);
        // 更新缓存
        cache.set(key, new CacheItem(newData, System.currentTimeMillis() + expireTime));
    });
}

memo：2025 年 5 月 18 日修改至此，今天给球友改简历时，碰到一个西北工业大学的球友，这又是一所 985 院校，希望这个社群能把所有的 985 院校集齐，也希望去帮助到更多院校的同学，希望大家都能拿到一个满意的 offer。

什么是缓存穿透？

缓存穿透是指查询的数据在缓存中没有命中，因为数据压根不存在，所以请求会直接落到数据库上。如果这种查询非常频繁，就会给数据库造成很大的压力。

缓存击穿是因为单个热点数据缓存失效导致的，而缓存穿透是因为查询的数据不存在，原因可能是自身的业务代码有问题，或者是恶意攻击造成的，比如爬虫。

常用的解决方案有两种：第一种是布隆过滤器，它是一种空间效率很高的数据结构，可以用来判断一个元素是否在集合中。

我们可以将所有可能存在的数据哈希到布隆过滤器中，查询时先检查布隆过滤器，如果布隆过滤器认为该数据不存在，就直接返回空；否则再去查询缓存，这样就可以避免无效的缓存查询。

代码示例：

public String getData(String key) {
    // 缓存中不存在该key
    String cacheResult = cache.get(key);
    if (cacheResult != null) {
        return cacheResult;
    }
    
    // 布隆过滤器判断key是否可能存在
    if (!bloomFilter.mightContain(key)) {
        return null; // 一定不存在，直接返回
    }
    
    // 可能存在，查询数据库
    String dbResult = db.query(key);
    
    // 将结果放入缓存，包括空值
    cache.set(key, dbResult != null ? dbResult : "", expireTime);
    
    return dbResult;
}

布隆过滤器存在误判，即可能会认为某个数据存在，但实际上并不存在。但绝不会漏判，即如果布隆过滤器认为某个数据不存在，那它一定不存在。因此它可以有效拦截不存在的数据查询，减轻数据库压力。

第二种是缓存空值。对于不存在的数据，我们将空值写入缓存，并设置一个合理的过期时间。这样下次相同的查询就能直接从缓存返回，而不再访问数据库。

代码示例：

public String getData(String key) {
    String cacheResult = cache.get(key);
    
    // 缓存命中，包括空值
    if (cacheResult != null) {
        // 特殊值表示空结果
        if (cacheResult.equals("")) {
            return null;
        }
        return cacheResult;
    }
    
    // 缓存未命中，查询数据库
    String dbResult = db.query(key);
    
    // 写入缓存，空值也缓存，但设置较短的过期时间
    int expireTime = dbResult == null ? EMPTY_EXPIRE_TIME : NORMAL_EXPIRE_TIME;
    cache.set(key, dbResult != null ? dbResult : "", expireTime);
    
    return dbResult;
}

缓存空值的方法实现起来比较简单，但需要给空值设置一个合理的过期时间，以免数据库中新增了这些数据后，缓存仍然返回空值。

在实际的项目当中，还需要在接口层面做一些处理，比如说对参数进行校验，拦截明显不合理的请求；或者对疑似攻击的 IP 进行限流和封禁。

memo：2025 年 5 月 19 日，今天有球友发微信说拿到了滴滴的测开实习 offer，目前还想继续找，问我该继续学点什么，我的回复说，暑期能拿到 offer，秋招继续就行了，加上滴滴的实习经历就很硬核了。大家在准备暑期和秋招的时候，也不要太焦虑，保持一个好的学习习惯，秋招没问题的。

什么是缓存雪崩？

缓存雪崩是指在某一时间段，大量缓存同时失效或者缓存服务突然宕机了，导致大量请求直接涌向数据库，导致数据库压力剧增，甚至引发系统崩溃的现象。

缓存击穿是单个热点数据失效导致的，缓存穿透是因为请求不存在的数据，而缓存雪崩是因为大范围的缓存失效。

缓存雪崩主要有三种成因和应对策略。

第一种，大量缓存同时过期，解决方法是添加随机过期时间。

public void setCache(String key, String value) {
    // 基础过期时间，例如30分钟
    int baseExpireSeconds = 1800;
    // 增加随机过期时间，范围0-300秒
    int randomSeconds = new Random().nextInt(300);
    // 最终过期时间为基础时间加随机时间
    cache.set(key, value, baseExpireSeconds + randomSeconds);
}

第二种，缓存服务崩溃，解决方法是使用高可用的缓存集群。

比如说使用 Redis Cluster 构建多节点集群，确保数据在多个节点上有备份，并且支持自动故障转移。

对于一些高频关键数据，可以配置本地缓存作为二级缓存，缓解 Redis 的压力。在技术派实战项目中，我们就采用了多级缓存的策略，其中就包括使用本地缓存 Caffeine 来作为二级缓存，当 Redis 出现问题时自动切换到本地缓存。

这个过程称为“缓存降级”，保证 Redis 发生故障时，系统能够继续提供服务。

LoadingCache<String, UserPermissions> permissionsCache = Caffeine.newBuilder()
    .maximumSize(1000)
    .expireAfterWrite(10, TimeUnit.MINUTES)
    .build(this::loadPermissionsFromRedis);

public UserPermissions loadPermissionsFromRedis(String userId) {
    try {
        return redisClient.getPermissions(userId);
    } catch (Exception ex) {
        // Redis 异常处理，尝试从本地缓存获取
        return permissionsCache.getIfPresent(userId);
    }
}

第三种，缓存服务正常但并发请求量超过了缓存服务的承载能力，这种情况下可以采用限流和降级措施。

public String getData(String key) {
    try {
        // 尝试从缓存获取数据
        return cache.get(key);
    } catch (Exception e) {
        // 缓存服务异常，触发熔断
        if (circuitBreaker.shouldTrip()) {
            // 直接从数据库获取，并进入降级模式
            circuitBreaker.trip();
            return getFromDbDirectly(key);
        }
        throw e;
    }
}

private String getFromDbDirectly(String key) {
    // 实施限流保护
    if (!rateLimit.tryAcquire()) {
        // 超过限流阈值，返回兜底数据或默认值
        return getDefaultValue(key);
    }
    
    // 限流通过，从数据库查询
    return db.query(key);
}

1. Java 面试指南（付费）收录的腾讯面经同学 22 暑期实习一面面试原题：缓存雪崩，如何解决
2. Java 面试指南（付费）收录的快手面经同学 7 Java 后端技术一面面试原题：说一下 缓存穿透、缓存击穿、缓存雪崩
3. Java 面试指南（付费）收录的字节跳动同学 7 Java 后端实习一面的原题：Redis 宕机会不会对权限系统有影响？
4. Java 面试指南（付费）收录的字节跳动同学 7 Java 后端实习一面的原题：说一下 Redis 雪崩、穿透、击穿等场景的解决方案
5. Java 面试指南（付费）收录的小米同学 F 面试原题：缓存常见问题和解决方案（引申到多级缓存），多级缓存（redis，nginx，本地缓存）的实现思路
6. Java 面试指南（付费）收录的TP联洲同学 5 Java 后端一面的原题：如何解决缓存穿透
7. Java 面试指南（付费）收录的理想汽车面经同学 2 一面面试原题：如何理解缓存雪崩、缓存击穿和缓存穿透？

memo：2025 年 5 月 20 日，今天有球友发微信说项目用的技术派，八股背的面渣，春招拿到了四个 offer，其中包括泰隆银行和交通银行，问我该怎么选择，说实话我看完后觉得挺难选的，😄不过还是值得恭喜一手。大家在准备春招的时候也不要着急，付出总会有回报的。

30.🌟能说说布隆过滤器吗？

布隆过滤器是一种空间效率极高的概率性数据结构，用于快速判断一个元素是否在一个集合中。它的特点是能够以极小的内存消耗，判断一个元素“一定不在集合中”或“可能在集合中”，常用来解决 Redis 缓存穿透的问题。

----这部分面试中可以不背start----

布隆过滤器的核心由一个很长的二进制向量和一系列哈希函数组成。

• 初始化的时候，创建一个长度为 m 的位数组，初始值全为 0，同时选择 k 个不同的哈希函数
• 当添加一个元素时，用 k 个哈希函数计算出 k 个哈希值，然后对 m 取模，得到 k 个位置，将这些位置的二进制位都设为 1
• 当需要判断一个元素是否在集合中时，同样用 k 个哈希函数计算出 k 个位置，如果这些位置的二进制位有任何一个为 0，该元素一定不在集合中；如果全部为 1，则该元素可能在集合中

public class BloomFilter<T> {
    private BitSet bitSet;
    private int bitSetSize;
    private int numberOfHashFunctions;
    
    public BloomFilter(double falsePositiveProbability, int expectedNumberOfElements) {
        // 根据预期元素数量和期望的误判率，计算最优的位数组大小和哈希函数个数
        this.bitSetSize = calculateOptimalBitSetSize(expectedNumberOfElements, falsePositiveProbability);
        this.numberOfHashFunctions = calculateOptimalNumberOfHashFunctions(expectedNumberOfElements, bitSetSize);
        this.bitSet = new BitSet(bitSetSize);
    }
    
    public void add(T element) {
        int[] hashes = createHashes(element);
        for (int hash : hashes) {
            bitSet.set(Math.abs(hash % bitSetSize), true);
        }
    }
    
    public boolean mightContain(T element) {
        int[] hashes = createHashes(element);
        for (int hash : hashes) {
            if (!bitSet.get(Math.abs(hash % bitSetSize))) {
                return false; // 如果任何一位为0，元素一定不存在
            }
        }
        return true; // 所有位都为1，元素可能存在
    }
    
    // 其他辅助方法，如计算哈希值，计算最优参数等
}

----这部分面试中可以不背end----

布隆过滤器存在误判吗？

是的，布隆过滤器存在误判。它可能会错误地认为某个元素在集合中，而元素实际上并不在集合中。

但如果布隆过滤器认为某个元素不存在于集合中，那么它一定不存在。

误判产生的原因是因为哈希冲突。在布隆过滤器中，多个不同的元素可能映射到相同的位置。随着向布隆过滤器中添加的元素越来越多，位数组中的 1 也越来越多，发生哈希冲突的概率随之增加，误判率也就随之上升。

误判率取决于以下 3 个因素：

1. 位数组的大小（m）：m 决定了可以存储的标志位数量。如果位数组过小，那么哈希碰撞的几率就会增加，从而导致更高的误判率。
2. 哈希函数的数量（k）：k 决定了每个元素在位数组中标记的位数。哈希函数越多，碰撞的概率也会相应变化。如果哈希函数太少，过滤器很快会变得不精确；如果太多，误判率也会升高，效率下降。
3. 存入的元素数量（n）：n 越多，哈希碰撞的几率越大，从而导致更高的误判率。

$$ f(k) = \left( 1 - e^{- \frac{kn}{m}} \right)^k $$

要降低误判率，可以增加位数组的大小或者减少插入的元素数量。

要彻底解决布隆过滤器的误判问题，可以在布隆过滤器返回"可能存在"时，再通过数据库进行二次确认。

布隆过滤器支持删除吗？

布隆过滤器并不支持删除操作，这是它的一个重要限制。

当我们添加一个元素时，会将位数组中的 k 个位置设置为 1。由于多个不同元素可能共享相同的位，如果我们尝试删除一个元素，将其对应的 k 个位重置为 0，可能会错误地影响到其他元素的判断结果。

例如，元素 A 和元素 B 都将位置 5 设为 1，如果删除元素 A 时将位置 5 重置为 0，那么对元素 B 的查询就会产生错误的"不存在"结果，这违背了布隆过滤器的基本特性。

如果想要实现删除操作，可以使用计数布隆过滤器，它在每个位置上存储一个计数器而不是单一的位。这样可以通过减少计数器的值来实现删除操作，但会增加内存开销。

public class CountingBloomFilter<T> {
    private int[] counters;
    private int size;
    private int hashFunctions;
    
    public CountingBloomFilter(int size, int hashFunctions) {
        this.size = size;
        this.hashFunctions = hashFunctions;
        this.counters = new int[size];
    }
    
    public void add(T element) {
        int[] positions = getHashPositions(element);
        for (int position : positions) {
            counters[position]++;
        }
    }
    
    public void remove(T element) {
        int[] positions = getHashPositions(element);
        for (int position : positions) {
            if (counters[position] > 0) {
                counters[position]--;
            }
        }
    }
    
    public boolean mightContain(T element) {
        int[] positions = getHashPositions(element);
        for (int position : positions) {
            if (counters[position] == 0) {
                return false;
            }
        }
        return true;
    }
    
    private int[] getHashPositions(T element) {
        // 计算哈希位置的代码
    }
}

为什么不能用哈希表而是用布隆过滤器？

布隆过滤器最突出的优势是内存效率。

假如我们要判断 10 亿个用户 ID 是否曾经访问过特定页面，使用哈希表至少需要 10G 内存（每个 ID 至少需要8字节），而使用布隆过滤器只需要 1.2G 内存。

m ≈ -n*ln(p)/ln(2)² ≈ -10⁹*ln(0.01)/ln(2)² ≈ 9.6 billion bits ≈ 1.2GB

1. Java 面试指南（付费）收录的字节跳动同学 7 Java 后端实习一面的原题：有了解过布隆过滤器吗？
2. Java 面试指南（付费）收录的TP联洲同学 5 Java 后端一面的原题：布隆过滤器原理，这种方式下5%的错误率可接受？
3. Java 面试指南（付费）收录的美团同学 9 一面面试原题：布隆过滤器？布隆过滤器优点？为什么不能用哈希表要用布隆过滤器？
4. Java 面试指南（付费）收录的理想汽车面经同学 2 一面面试原题：追问：说明一下布隆过滤器

memo：2025 年 5 月 20 日，今天有球友发贴说拿到了滴滴的暑期 offer，特意来感谢了一下面渣逆袭。

31.🌟如何保证缓存和数据库的数据⼀致性？

在技术派实战项目中，对于文章标签这种允许短暂不一致的数据，我会采用 Cache Aside + TTL 过期机制来保证缓存和数据库的一致性。

具体做法是读取时先查 Redis，未命中再查 MySQL，同时为缓存设置一个合理的过期时间；更新时先更新 MySQL，再删除 Redis。

这种方式简单有效，适用于读多写少的场景。TTL 过期时间也能够保证即使更新操作失败，未能及时删除缓存，过期时间也能确保数据最终一致。

// 读取逻辑
public UserInfo getUser(String userId) {
    // 先查缓存
    UserInfo user = cache.get("user:" + userId);
    if (user != null) {
        return user;
    }
    
    // 缓存未命中，查数据库
    user = database.selectUser(userId);
    if (user != null) {
        // 放入缓存，设置合理的过期时间
        cache.set("user:" + userId, user, 3600);
    }
    
    return user;
}

// 更新逻辑
public void updateUser(UserInfo user) {
    // 先更新数据库
    database.updateUser(user);
    
    // 删除缓存
    cache.delete("user:" + user.getId());
}

那再来说说为什么要删除缓存而不是更新缓存？

最初设计缓存策略时，我也考虑过直接更新缓存，但通过实践发现，删除缓存是更优的选择。

最主要的原因是在并发环境下，假设我们有两个并发的更新操作，如果采用更新缓存的策略，就可能出现这样的时序问题：

• 操作 A 和操作 B 同时发生，A 先更新 MySQL 将值改为 10，B 后更新 MySQL 将值改为 11。但在缓存更新时，可能 B 先执行将缓存设为 11，然后 A 才执行将缓存设为10。这样就会造成 MySQL 是 11 但 Redis 是 10 的不一致状态。

而采用删除策略，无论 A 和 B 谁先删除缓存，后续的读取操作都会从 MySQL 获取最新值。

另外，相对而言，删除缓存的速度比更新缓存的速度快得多。

因为删除操作只是简单的 DEL 命令，而更新可能需要重新序列化整个对象再写入缓存。

那再说说为什么要先更新数据库，再删除缓存？

这个操作顺序的选择也是我在实际项目中踩过坑才深刻理解的。假设我们采用先删缓存再更新数据库的策略，在高并发场景下就可能出现这样的问题：

• 线程 A 要更新用户信息，先删除了缓存
• 线程 B 恰好此时要读取该用户信息，发现缓存为空，于是查询数据库，此时还是旧值
• 线程 B 将查到的旧值重新放入缓存
• 线程 A 完成数据库更新

结果就是数据库是新的值，但缓存中还是旧值。

而采用先更新数据库再删缓存的策略，即使出现类似的并发情况，最坏的情况也只是短暂地从缓存中读取到了旧值，但缓存删除后的请求会直接从数据库中获取最新值。

另外，如果先删缓存再更新数据库，当数据库更新失败时，缓存已经被删除了。这会导致短期内所有读请求都会穿透到数据库，对数据库造成额外的压力。

而先更新数据库再删缓存，如果数据库更新失败，缓存保持原状，系统仍然能继续正常提供服务。

public void updateUser(User user) {
    try {
        // 先更新数据库
        database.updateUser(user);
        
        // 再删除缓存
        cache.delete("user:" + user.getId());
    } catch (DatabaseException e) {
        // 数据库更新失败，缓存保持原状，系统仍可正常提供服务
        log.error("Database update failed", e);
        throw e;
    } catch (CacheException e) {
        // 缓存删除失败，数据库已更新，数据会在TTL后自动一致
        log.warn("Cache deletion failed, will be eventually consistent", e);
        // 可以选择不抛异常，因为有TTL兜底
    }
}

memo：2025 年 5 月 22 日，今天给球友修改简历时，碰到一个西北工业大学本、电子科技大学硕的球友，一下子 985 高校又集齐了两所。如果球友们在星球里有所收获，也请给学弟学妹们一个口碑，让大家都能因此受益，拿到更好的 offer。

那假如对缓存数据库一致性要求很高，该怎么办呢？

当业务对缓存与数据库的一致性要求很高时，比如支付系统、库存管理等场景，我会采用多种策略来保证强一致性。

第一种，引入消息队列来保证缓存最终被删除，比如说在数据库更新的事务中插入一条本地消息记录，事务提交后异步发送给 MQ 进行缓存删除。

即使缓存删除失败，消息队列的重试机制也能保证最终一致性。

@Transactional
public void updateUser(UserInfo user) {
    // 在事务中更新数据库
    database.updateUser(user);
    
    // 在同一事务中记录需要删除的缓存信息
    LocalMessage message = new LocalMessage("CACHE_DELETE", "user:" + user.getId());
    database.insertLocalMessage(message);

    // 显式发布事件，供监听器捕获
    eventPublisher.publishEvent(new UserUpdateEvent(this, "user:" + user.getId()));
}

// 事务提交后发送MQ消息
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
public void sendCacheDeleteMessage(UserUpdateEvent event) {
    messageQueue.send("cache-delete-topic", event.getCacheKey());
}

第二种，使用 Canal 监听 MySQL 的 binlog，在数据更新时，将数据变更记录到消息队列中，消费者消息监听到变更后去删除缓存。

这种方案的优势是完全解耦了业务代码和缓存维护逻辑。

@CanalListener
public class CacheUpdateListener {
    
    @EventHandler
    public void handleUserUpdate(UserUpdateEvent event) {
        // 从binlog事件中提取变更信息
        String userId = event.getUserId();
        
        // 发送缓存删除消息
        CacheDeleteMessage message = new CacheDeleteMessage();
        message.setCacheKey("user:" + userId);
        messageQueue.send("cache-delete-topic", message);
    }
}
// 消费者监听消息队列
@KafkaListener(topics = "cache-delete-topic")
public void handleCacheDeleteMessage(CacheDeleteMessage message) {
    // 删除缓存
    cache.delete(message.getCacheKey());
}

当然了，如果说业务比较简单，不需要上消息队列，可以通过延迟双删策略降低缓存和数据库不一致的时间窗口，在第一次删除缓存之后，过一段时间之后，再次尝试删除缓存。

这种方式主要针对缓存不存在，但写入了脏数据的情况。

public void updateUser(UserInfo user) {
    // 第一次删除缓存，减少不一致时间窗口
    cache.delete("user:" + user.getId());
    
    // 更新数据库
    database.updateUser(user);
    
    // 立即删除缓存
    cache.delete("user:" + user.getId());
    
    // 延时删除，应对可能的并发读取
    CompletableFuture.runAsync(() -> {
        try {
            Thread.sleep(1000); // 延时时间根据主从同步延迟调整
            cache.delete("user:" + user.getId());
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    });
}

最后，无论采用哪种策略，最好为缓存设置一个合理的过期时间作为最后的保障。即使所有的主动删除机制都失败了，TTL 也能确保数据最终达到一致：

// 根据数据的重要程度设置不同的TTL
public void setCache(String key, Object value, DataImportance importance) {
    int ttl;
    switch (importance) {
        case HIGH:      // 关键数据，短TTL
            ttl = 300;  // 5分钟
            break;
        case MEDIUM:    // 一般数据
            ttl = 1800; // 30分钟
            break;
        case LOW:       // 不太重要的数据
            ttl = 3600; // 1小时
            break;
    }
    
    cache.setWithTTL(key, value, ttl);
}

这种方式虽然简单，但能确保即使出现极端情况，数据不一致的影响也是可控的。

1. Java 面试指南（付费）收录的华为面经同学 8 技术二面面试原题：怎样保证数据的最终一致性？
2. Java 面试指南（付费）收录的腾讯面经同学 23 QQ 后台技术一面面试原题：数据一致性问题
3. Java 面试指南（付费）收录的微众银行同学 1 Java 后端一面的原题：MySQL 和缓存一致性问题了解吗？
4. Java 面试指南（付费）收录的美团面经同学 3 Java 后端技术一面面试原题：如何保证 redis 缓存与数据库的一致性，为什么这么设计
5. Java 面试指南（付费）收录的比亚迪面经同学 12 Java 技术面试原题：怎么解决redis和mysql的缓存一致性问题
6. Java 面试指南（付费）收录的字节跳动同学 17 后端技术面试原题：双写一致性怎么解决的
7. Java 面试指南（付费）收录的京东面经同学 9 面试原题：redis的数据和缓存不一致应该处理

memo：2025 年 5 月 23 日修改至此，今天在修改球友简历时，看到一条非常温暖的感谢信，球友说改完后的简历，每一句都比之前的好很多，真的很欣慰，感觉自己的付出得到了回报。😄

32.如何保证本地缓存和分布式缓存的一致？

在技术派实战项目中，为了减轻 Redis 的负载压力，我又追加了一层本地缓存 Caffeine。

为了保证 Caffeine 和 Redis 缓存的一致性，我采用的策略是当数据更新时，通过 Redis 的 pub/sub 机制向所有应用实例发送缓存更新通知，收到通知后的实例立即更新或者删除本地缓存。

@Service
public class CacheService {
    
    private final RedisTemplate redisTemplate;
    private final CaffeineCache localCache;
    
    public void updateData(String key, Object value) {
        // 更新数据库
        database.update(key, value);
        
        // 更新分布式缓存
        redisTemplate.opsForValue().set(key, value, 30, TimeUnit.MINUTES);
        
        // 发送缓存更新通知
        CacheUpdateMessage message = new CacheUpdateMessage(key, "UPDATE", value);
        redisTemplate.convertAndSend("cache-update-channel", message);
    }
    
    @EventListener
    public void handleCacheUpdate(CacheUpdateMessage message) {
        if ("UPDATE".equals(message.getAction())) {
            localCache.put(message.getKey(), message.getValue());
        } else if ("DELETE".equals(message.getAction())) {
            localCache.invalidate(message.getKey());
        }
    }
}

考虑到消息可能丢失，我还会引入版本号机制作为补充。每次从 Redis 获取数据时添加一个最新的版本号。从本地缓存获取数据前，先检查自己的版本号是否是最新的，如果发现版本落后，就主动从 Redis 中获取最新数据。

@Component
public class VersionBasedCacheManager {

    @Autowired
    private StringRedisTemplate redisTemplate;

    // 使用 Caffeine 构建本地缓存：最多 1000 项，写入后 10 分钟过期
    private final Cache<String, VersionedData> localCache = Caffeine.newBuilder()
        .maximumSize(1000)
        .expireAfterWrite(10, TimeUnit.MINUTES)
        .build();

    /**
     * 获取缓存数据，优先使用本地缓存，必要时从 Redis 加载
     */
    public Object get(String key) {
        VersionedData cached = localCache.getIfPresent(key); // 从本地缓存取出

        // 从 Redis 获取版本号
        String versionStr = redisTemplate.opsForValue().get(key + ":version");

        // 如果 Redis 中没找到版本号，说明可能数据已失效，强制刷新
        if (versionStr == null) {
            return loadAndCache(key);
        }

        long remoteVersion = Long.parseLong(versionStr);

        // 如果本地没有缓存，或版本落后于 Redis，强制刷新
        if (cached == null || cached.getVersion() < remoteVersion) {
            return loadAndCache(key);
        }

        // 命中本地缓存且版本最新，直接返回
        return cached.getData();
    }

    /**
     * 从 Redis 加载数据和版本，并写入本地缓存
     */
    private Object loadAndCache(String key) {
        Object data = redisTemplate.opsForValue().get(key);
        String versionStr = redisTemplate.opsForValue().get(key + ":version");

        if (data != null && versionStr != null) {
            long version = Long.parseLong(versionStr);
            localCache.put(key, new VersionedData(data, version));
        }

        return data;
    }
}

如果在项目中多个地方都要使用到二级缓存的逻辑，如何设计这一块？

我的思路是将二级缓存抽象成一个统一的组件。设计一个 CacheManager 作为核心入口，提供 get、put、evict 等基本操作，执行先查本地缓存，再查分布式缓存，最后查数据库的完整流程。

public class CacheManager {

    private final LocalCache localCache;
    private final RedisCache redisCache;
    private final Database database;

    public CacheManager(LocalCache localCache, RedisCache redisCache, Database database) {
        this.localCache = localCache;
        this.redisCache = redisCache;
        this.database = database;
    }

    public Object get(String key) {
        // 先查本地缓存
        Object value = localCache.get(key);
        if (value != null) {
            return value;
        }

        // 再查分布式缓存
        value = redisCache.get(key);
        if (value != null) {
            // 更新本地缓存
            localCache.put(key, value);
            return value;
        }

        // 最后查数据库
        value = database.get(key);
        if (value != null) {
            // 更新分布式缓存和本地缓存
            redisCache.put(key, value);
            localCache.put(key, value);
        }
        
        return value;
    }
}

本地缓存和 Redis 的区别了解吗？

Redis 可以部署在多个节点上，支持数据分片、主从复制和集群。而本地缓存只能在单个服务器上使用。

对于读取频率极高、数据相对稳定、允许短暂不一致的数据，我优先选择本地缓存。比如系统配置信息、用户权限数据、商品分类信息等。

而对于需要实时同步、数据变化频繁、多个服务需要共享的数据，我会选择 Redis。比如用户会话信息、购物车数据、实时统计信息等。

1. Java 面试指南（付费）收录的字节跳动同学 7 Java 后端实习一面的原题：怎么保证二级缓存和 Redis 缓存的数据一致性？
2. Java 面试指南（付费）收录的华为面经同学 11 面试原题：使用的 guava cache 和 redis 是如何组合使用的？如果在项目中多个地方都要使用到二级缓存的逻辑，如何设计这一块？
3. Java 面试指南（付费）收录的去哪儿同学 1 技术二面的原题：redis 和本地缓存的区别，哪个效率高
4. Java 面试指南（付费）收录的拼多多面经同学 8 一面面试原题：缓存一致性如何保证

33.什么是热Key？

所谓的热 Key，就是指在很短时间内被频繁访问的键。比如电商大促期间爆款商品的详情信息，流量明星爆瓜时的个人资料、热门话题等，都可能成为热Key。

由于 Redis 是单线程模型，大量请求集中到同一个键会导致该 Redis 节点的 CPU 使用率飙升，响应时间变长。

在 Redis 集群环境下，热Key 还会导致数据分布不均衡，某个节点承受的压力过大而其他节点相对空闲。

更严重的情况是，当热Key 过期或被误删时，会引发缓存击穿问题。

那怎么监控热Key 呢？

临时的方案可以使用 redis-cli --hotkeys 命令来监控 Redis 中的热 Key。

redis-cli -h <address> -p <port> -a<password> — hotkey

或者在访问缓存时，在本地维护一个计数器，当某个键的访问次数在一分钟内超过设定阈值，就将其标记为热Key。

@Component
public class HotKeyDetector {
    private final ConcurrentHashMap<String, AtomicLong> accessCounter = new ConcurrentHashMap<>();
    private final int HOT_KEY_THRESHOLD = 1000;
    
    public boolean isHotKey(String key) {
        long count = accessCounter.computeIfAbsent(key, k -> new AtomicLong(0))
                                  .incrementAndGet();
        return count > HOT_KEY_THRESHOLD;
    }
}

34.那怎么处理热Key 呢？

最有效的解决方法是增加本地缓存，将热 Key 缓存到本地内存中，这样请求就不需要访问 Redis 了。

对于一些特别热的 Key，可以将其拆分成多个子 Key，然后随机分布到不同的 Redis 节点上。比如将 hot_product:12345 拆分成 hot_product:12345:1、hot_product:12345:2 等多个副本，读取时随机选择其中一个。

public String getHotData(String key) {
    if (isHotKey(key)) {
        // 随机选择一个副本
        int replica = ThreadLocalRandom.current().nextInt(HOT_KEY_REPLICAS);
        return redis.get(key + ":" + replica);
    }
    return redis.get(key);
}

35.怎么处理大 Key 呢？

大Key 是指占用内存空间较大的缓存键，比如超过 10M 的键值对。常见的大Key 类型包括：包含大量元素的 List、Set、Hash 结构，存储大文件的 String 类型，以及包含复杂嵌套对象的 JSON 数据等。

在内存有限的情况下，可能导致 Redis 内存不足。另外，大Key 还会导致主从复制同步延迟，甚至引发网络拥塞。

可以通过 redis-cli --bigkeys 命令来监控 Redis 中的大 Key。

或者编写脚本进行全量扫描：

@Component
public class BigKeyScanner {
    
    private final RedisTemplate redisTemplate;
    private final int BIG_KEY_THRESHOLD = 1024 * 1024; // 1MB
    
    public List<BigKeyInfo> scanBigKeys() {
        List<BigKeyInfo> bigKeys = new ArrayList<>();
        
        // 使用SCAN命令遍历所有键
        ScanOptions options = ScanOptions.scanOptions().count(1000).build();
        Cursor<byte[]> cursor = redisTemplate.executeWithStickyConnection(
            connection -> connection.scan(options)
        );
        
        while (cursor.hasNext()) {
            String key = new String(cursor.next());
            long memory = getKeyMemoryUsage(key);
            
            if (memory > BIG_KEY_THRESHOLD) {
                bigKeys.add(new BigKeyInfo(key, memory, getKeyType(key)));
            }
        }
        
        return bigKeys;
    }
    
    private long getKeyMemoryUsage(String key) {
        // 使用MEMORY USAGE命令获取键的内存占用
        return redisTemplate.execute((RedisCallback<Long>) connection -> 
            connection.memoryUsage(key.getBytes())
        );
    }
}  

对于大 Key 问题，最根本的解决方案是拆分大 Key，将其拆分成多个小 Key 存储。比如将一个包含大量用户信息的 Hash 拆分成多个小 Hash。

public void splitBigKey(String bigKey) {
    Map<String, String> bigData = redisTemplate.opsForHash().entries(bigKey);
    
    // 将大 Key 拆分成多个小 Key
    for (Map.Entry<String, String> entry : bigData.entrySet()) {
        String smallKey = bigKey + ":" + entry.getKey();
        redisTemplate.opsForValue().set(smallKey, entry.getValue());
    }
    
    // 删除原始大 Key
    redisTemplate.delete(bigKey);
}

另外，对于 JSON 数据，可以进行 Gzip 压缩后再存储，虽然会增加一些 CPU 开销，但在内存敏感的场景在是值得的。

public void setCompressedData(String key, Object data) {
    try {
        String json = objectMapper.writeValueAsString(data);
        byte[] compressed = compress(json.getBytes());
        redisTemplate.opsForValue().set(key, compressed);
    } catch (Exception e) {
        log.error("Failed to compress data", e);
    }
}

private byte[] compress(byte[] data) throws IOException {
    ByteArrayOutputStream out = new ByteArrayOutputStream();
    try (GZIPOutputStream gzip = new GZIPOutputStream(out)) {
        gzip.write(data);
    }
    return out.toByteArray();
}

推荐阅读：

• 阿里：发现并处理 Redis 的大 Key 和热 Key
• 董宗磊：Redis 热 Key 发现以及解决办法

1. Java 面试指南（付费）收录的华为 OD 的面试中出现过该题：讲一讲 Redis 的热 Key 和大 Key

memo：2025 年 5 月 24 日，今天球友发私信说，拿到了荣耀通软的实习 offer，恭喜他！🎉

36.缓存预热怎么做呢？

缓存预热是指在系统启动或者特定时间点，提前将热点数据加载到缓存中，避免冷启动时大量请求直接打到数据库。

缓存预热的方法有多种，在技术派实战项目中，我会在项目启动时将热门文章提前加载到 Redis 中，在每天凌晨定时将最新的站点地图更新到 Redis中，以确保用户在第一次访问时就能获取到缓存数据，从而减轻数据库的压力。

/**
 * 采用定时器方案，每天5:15分刷新站点地图，确保数据的一致性
 */
@Scheduled(cron = "0 15 5 * * ?")
public void autoRefreshCache() {
    log.info("开始刷新sitemap.xml的url地址，避免出现数据不一致问题!");
    refreshSitemap();
    log.info("刷新完成！");
}

@Override
public void refreshSitemap() {
    initSiteMap();
}

private synchronized void initSiteMap() {
    long lastId = 0L;
    RedisClient.del(SITE_MAP_CACHE_KEY);
    while (true) {
        List<SimpleArticleDTO> list = articleDao.getBaseMapper().listArticlesOrderById(lastId, SCAN_SIZE);

        // 刷新站点地图信息，放到 Redis 当中
        Map<String, Long> map = list.stream().collect(Collectors.toMap(s -> String.valueOf(s.getId()), s -> s.getCreateTime().getTime(), (a, b) -> a));
        RedisClient.hMSet(SITE_MAP_CACHE_KEY, map);
        if (list.size() < SCAN_SIZE) {
            break;
        }
        lastId = list.get(list.size() - 1).getId();
    }
}

1. Java 面试指南（付费）收录的字节跳动面经同学 1 技术二面面试原题：什么是缓存预热？如何解决？

37.无底洞问题听说过吗？如何解决？

无底洞问题的核心在于，随着缓存节点数量的增加，虽然总的存储容量和理论吞吐量都在增长，但是单个请求的响应时间反而变长了。

这个问题的根本原因是网络通信开销的增加。当节点数量从几十个增长到几千个时，客户端需要与更多的节点进行通信。

其次就是数据分布的碎片化。随着节点增多，数据分散得更加细碎，原本可以在一个节点获取的相关数据，现在可能分散在多个节点上。

针对这个问题，可以采取以下几种解决方案：

第一，可以将同一节点的多个请求合并成一个批量请求，减少网络往返次数。

public Map<String, Object> batchGet(List<String> keys) {
    // 按节点分组keys
    Map<String, List<String>> nodeKeysMap = groupKeysByNode(keys);
    Map<String, Object> results = new ConcurrentHashMap<>();
    
    // 并发访问各个节点
    List<CompletableFuture<Void>> futures = nodeKeysMap.entrySet().stream()
        .map(entry -> CompletableFuture.runAsync(() -> {
            String node = entry.getKey();
            List<String> nodeKeys = entry.getValue();
            
            // 批量获取该节点的数据
            Map<String, Object> nodeResults = getFromNode(node, nodeKeys);
            results.putAll(nodeResults);
        }))
        .collect(Collectors.toList());
    
    // 等待所有请求完成
    CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
    
    return results;
}

第二，可以使用一致性哈希算法来优化数据分布，减少数据迁移和重分布的开销。

public class LocalityAwareSharding {
    
    public String getNodeForKey(String key, String category) {
        // 相同类别的数据尽量分配到相同节点
        String shardKey = category + ":" + (key.hashCode() % SHARDS_PER_CATEGORY);
        return consistentHash.getNode(shardKey);
    }
    
    // 用户相关数据尽量在同一个节点
    public String getUserDataNode(String userId) {
        return "user_cluster_" + (userId.hashCode() % USER_CLUSTERS);
    }
}

Redis 运维

38.Redis 报内存不足怎么处理？

Redis 报内存不足时，通常是因为 Redis 占用的物理内存已经接近或者超过了配置的最大内存限制。这时可以采取以下几种步骤来处理：

第一，使用 INFO memory 命令查看 Redis 的内存使用情况，看看是否真的达到了最大内存限制。

redis-cli INFO memory

第二，如果服务器还有可用内存的话，修改 redis.conf 中的 maxmemory 参数，增加 Redis 的最大内存限制。比如将最大内存设置为 8GB：

maxmemory 8gb

第三，修改 maxmemory-policy 参数来调整内存淘汰策略。比如可以选择 allkeys-lru 策略，让 Redis 自动删除最近最少使用的键。

maxmemory-policy allkeys-lru

memo：2025 年 5 月 25 日修改至此，今天在修改球友简历时，碰到一个西安交通大学本、上海交通大学硕的球友，985 本硕学历真的非常顶了，我会竭尽所能去帮助他，在秋招中斩获一个 SSP offer，冲！

39.Redis key过期策略有哪些？

Redis 主要采用了两种过期删除策略来保证过期的 key 能够被及时删除，包括惰性删除和定期删除。

惰性删除是最基本的策略，当客户端访问一个 key 时，Redis 会检查该 key 是否已过期，如果过期就会立即删除并返回 nil。

// 模拟惰性删除的逻辑
public Object get(String key) {
    RedisKey redisKey = getKeyFromMemory(key);
    
    if (redisKey != null && isExpired(redisKey)) {
        // key已过期，删除并返回null
        deleteKey(key);
        return null;
    }
    
    return redisKey != null ? redisKey.getValue() : null;
}

这种策略的优点是不会有额外的 CPU 开销，只在访问 key 时才检查。但问题是如果一个过期的 key 永远不被访问，它就会一直占用内存。

于是就有了定期删除策略，Redis 会定期随机选择一些设置了过期时间的 key 进行检查，删除其中已过期的 key。这个过程默认每秒执行 10 次，每次随机选择 20 个 key 进行检查。

----这部分面试中可以不背 start----

可以通过 config get hz 命令查看 Redis 内部定时任务的频率。

hz 的值为“10”意味着 Redis 每秒执行 10 次定时任务 。可以通过 CONFIG SET hz 20 进行调整。

----这部分面试中可以不背 end----

1. Java 面试指南（付费）收录的腾讯面经同学 22 暑期实习一面面试原题：Redis key 删除策略
2. Java 面试指南（付费）收录的去哪儿面经同学 1 技术 2 面面试原题：redis 内存淘汰和过期策略
3. Java 面试指南（付费）收录的京东面经同学 5 Java 后端技术一面面试原题：redis key过期策略

40.🌟Redis有哪些内存淘汰策略？

当内存使用接近 maxmemory 限制时，Redis 会依据内存淘汰策略来决定删除哪些 key 以缓解内存压力。

常用的内存淘汰策略有八种，分别是默认的 noeviction，内存不足时不会删除任何 key，直接返回错误信息，生产环境下基本上不会使用。

然后是针对所有 key 的 allkeys-lru、allkeys-lfu 和 allkeys-random。lru 会删除最近最少使用的 key，在纯缓存场景中最常用，能自动保留热点数据；lfu 会删除访问频率最低的 key，更适合长期运行的系统；random 会随机删除一些 key，一般不推荐使用。

其次是针对设置了过期时间的 key，有 volatile-lru、volatile-lfu、volatile-ttl 和 volatile-random。

lru 在混合存储场景中经常使用。

@Service
public class HybridStorageService {
    
    // 重要数据不设置过期时间，临时数据设置过期时间
    public void storeData(String key, Object data, DataImportance importance) {
        if (importance == DataImportance.HIGH) {
            // 重要数据不设置过期时间，在volatile-*策略下不会被淘汰
            redisTemplate.opsForValue().set(key, data);
        } else {
            // 临时数据设置过期时间，可以被volatile-*策略淘汰
            redisTemplate.opsForValue().set(key, data, Duration.ofHours(1));
        }
    }
}

lfu 适合需要保护某些重要数据不被淘汰的场景；ttl 优先删除即将过期的 key，在用户会话管理系统中推荐使用；random 仍然很少用。

1. Java 面试指南（付费）收录的小米春招同学 K 一面面试原题：为什么 redis 快，淘汰策略 持久化
2. Java 面试指南（付费）收录的去哪儿面经同学 1 技术 2 面面试原题：redis 内存淘汰和过期策略
3. Java 面试指南（付费）收录的作业帮面经同学 1 Java 后端一面面试原题：redis内存淘汰策略

41.LRU 和 LFU 的区别是什么？

LRU 是 Least Recently Used 的缩写，基于时间维度，淘汰最近最少访问的键。

LFU 是 Least Frequently Used 的缩写，基于次数维度，淘汰访问频率最低的键。

假设缓存中有三个数据 A、B、C，在 LRU 场景下，如果访问顺序是 A→B→C→A，那么此时的 LRU 顺序是B→C→A，如果需要淘汰，会先删除 B。

但在 LFU 场景下，如果 A 被访问了 5 次，B 被访问了 2 次，C 被访问了 1 次，那么无论最近的访问顺序如何，都会优先淘汰 C，因为它的访问频率最低。

LRU 更适合有明显时间局部性的场景，比如在新闻网站中，用户更关心最新的新闻，而昨天的新闻访问量会急剧下降。这种情况下，LRU 能很好地保留用户当前关心的热点内容。

LFU 则更适合有长期访问模式的场景，更强调“热度”，比如在电商平台中，某些商品可能长期保持热销状态，即使它们的访问时间间隔较长，但由于访问频率高，LFU 会优先保留这些商品的信息。

1. Java 面试指南（付费）收录的阿里系面经同学 19 饿了么面试原题：redis内存淘汰机制 延伸到LRU LFU

memo：2025 年 5 月 27 日，今天球友发私信说，拿到了哈啰和得物的实习 offer，恭喜他！🎉 还特意感谢了一下之前对他简历的修改和学习上的建议。

42.Redis发生阻塞了怎么解决？

Redis 发生阻塞在生产环境中是比较严重的问题，当发现 Redis 变慢时，我会先通过 monitor 命令查看当前正在执行的命令，或者使用 slowlog 命令查看慢查询日志。

# 查看当前正在执行的命令
redis-cli MONITOR

# 查看慢查询日志
redis-cli SLOWLOG GET 10

# 检查客户端连接状况
redis-cli CLIENT LIST

通常情况下，大Key 是导致 Redis 阻塞的主要原因之一。比如说直接 DEL 一个包含几百万个元素的 Set，就会导致 Redis 阻塞几秒钟甚至更久。

这时候可以用 UNLINK 命令替代 DEL 来异步删除，避免阻塞主线程。

# 使用 UNLINK 异步删除大 Key
redis-cli UNLINK big_key

对于非常大的集合，可以使用 SCAN 命令分批删除。

public void safeBatchProcess(String key) {
    ScanOptions options = ScanOptions.scanOptions().count(1000).build();
    Cursor<String> cursor = redisTemplate.opsForSet().scan(key, options);
    
    while (cursor.hasNext()) {
        String member = cursor.next();
        // 分批处理，避免阻塞
        processElement(member);
    }
}

另外，当 Redis 使用的内存超过物理内存时，操作系统会将部分内存交换到磁盘，这时候会导致 Redis 响应变慢。我的处理方式是：

使用 free -h 检查内存的使用情况 ；确认 Redis 的 maxmemory 设置是否合理；如果发生了内存交换，立即调整 maxmemory 并清理一些不重要的数据。

大量的客户端连接也可能会导致阻塞，这时候最好检查一下连接池的配置。

@Configuration
public class RedisConnectionConfig {
    
    @Bean
    public JedisConnectionFactory jedisConnectionFactory() {
        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(200);        // 最大连接数
        poolConfig.setMaxIdle(50);          // 最大空闲连接
        poolConfig.setMinIdle(10);          // 最小空闲连接
        poolConfig.setMaxWaitMillis(3000);  // 获取连接最大等待时间
        poolConfig.setTestOnBorrow(true);   // 获取连接时检测有效性
        
        return new JedisConnectionFactory(poolConfig);
    }
}

Redis 应用

43.Redis如何实现异步消息队列？

Redis 实现异步消息队列是一个很实用的技术方案，最简单的方式是使用 List 配合 LPUSH 和 RPOP 命令。

@Service
public class SimpleRedisQueue {
    
    private final RedisTemplate<String, Object> redisTemplate;
    
    // 生产者：向队列发送消息
    public void sendMessage(String queueName, Object message) {
        redisTemplate.opsForList().leftPush(queueName, message);
    }
    
    // 消费者：从队列获取消息
    public Object receiveMessage(String queueName) {
        return redisTemplate.opsForList().rightPop(queueName);
    }
    
    // 阻塞式消费，避免轮询
    public Object blockingReceive(String queueName, int timeoutSeconds) {
        List<Object> result = redisTemplate.opsForList()
            .rightPop(queueName, timeoutSeconds, TimeUnit.SECONDS);
        return result != null && !result.isEmpty() ? result.get(0) : null;
    }
}

另外就是用 Redis 的 Pub/Sub 来实现简单的消息广播和订阅。

@Service
public class RedisPubSubService {
    
    private final RedisTemplate<String, Object> redisTemplate;
    
    // 发布消息到指定频道
    public void publish(String channel, Object message) {
        redisTemplate.convertAndSend(channel, message);
    }
    
    // 订阅频道
    @PostConstruct
    public void subscribe() {
        redisTemplate.setMessageListener((message, pattern) -> {
            System.out.println("Received message: " + message);
        });
        redisTemplate.getConnectionFactory().getConnection().subscribe(
            new ChannelTopic("myChannel").getTopic().getBytes()
        );
    }
}

发布者将消息发布到指定的频道，订阅该频道的客户端就能收到消息。

但是这两种方式都是不可靠的，因为没有 ACK 机制所以不能保证订阅者一定能收到消息，也不支持消息持久化。

44.Redis如何实现延时消息队列?

延时消息队列在实际业务中很常见，比如订单超时取消、定时提醒等场景。Redis 虽然不是专业的消息队列，但可以很好地实现延时队列功能。

核心思路是利用 ZSet 的有序特性，将消息作为 member，把消息的执行时间作为 score。这样消息就会按照执行时间自动排序，我们只需要定期扫描当前时间之前的消息进行处理就可以了。

@Service
public class DelayedMessageQueue {
    
    private final RedisTemplate<String, Object> redisTemplate;
    
    // 发送延时消息
    public void sendDelayedMessage(String queueName, Object message, long delaySeconds) {
        // 计算消息的执行时间
        long executeTime = System.currentTimeMillis() + (delaySeconds * 1000);
        
        // 将消息加入ZSet，以执行时间作为score
        redisTemplate.opsForZSet().add(queueName, message, executeTime);
        
        log.info("发送延时消息: {}, 延时: {}秒", message, delaySeconds);
    }
    
    // 消费延时消息
    @Scheduled(fixedDelay = 1000) // 每秒扫描一次
    public void consumeDelayedMessages() {
        String queueName = "delayed:queue";
        long currentTime = System.currentTimeMillis();
        
        // 获取已到期的消息（score <= 当前时间）
        Set<Object> messages = redisTemplate.opsForZSet()
            .rangeByScore(queueName, 0, currentTime);
        
        for (Object message : messages) {
            try {
                // 处理消息
                processMessage(message);
                
                // 处理成功后从队列中移除
                redisTemplate.opsForZSet().remove(queueName, message);
                
                log.info("处理延时消息成功: {}", message);
            } catch (Exception e) {
                log.error("处理延时消息失败: {}", message, e);
                // 可以实现重试机制
                handleFailedMessage(queueName, message);
            }
        }
    }
}

具体实现上，我会在生产者发送延时消息时，计算消息应该执行的时间戳，然后用 ZADD 命令将消息添加到 ZSet 中。

ZADD delay_queue 1617024000 task1

消费者通过定时任务，使用 ZRANGEBYSCORE 命令获取当前时间之前的所有消息。

ZREMRANGEBYSCORE delay_queue -inf 1617024000

处理完成后再用 ZREM 删除消息。

ZREM delay_queue task1

在技术派实战项目中，我就用这种方式实现了文章定时发布的功能。作者在发布文章时，可以选择一个未来的时间节点，比如说 30 分钟后，系统就会向延时队列发送一条延时消息，然后定时任务就会在 30 分钟后将这条消息从延时队列中取出并发布文章。

1. Java 面试指南（付费）收录的腾讯面经同学 23 QQ 后台技术一面面试原题：Redis 实现延迟队列
2. Java 面试指南（付费）收录的字节跳动面经同学 8 Java 后端实习一面面试原题：redis 数据结构，用什么结构实现延迟消息队列

memo：2025 年 5 月 28 日修改至此，今天有球友在 VIP群里发消息说拿到了荣耀的暑期实习 offer，虽然时间节点已经不早了，但越是到这个时候，确实容易捡漏。

45.🌟Redis支持事务吗？

是的，Redis 支持简单的事务，可以将 multi、exec、discard 和 watch 命令打包，然后一次性的按顺序执行。

基本流程是用 multi 开启事务，然后执行一系列命令，最后用 exec 提交。这些命令会被放入队列，在 exec 时批量执行。

当客户端处于非事务状态时，所有发送给 Redis 服务的命令都会立即执行；但当客户端进入事务状态之后，这些命令会被放入一个事务队列中，然后立即返回 QUEUED，表示命令已入队。

当 exec 命令执行时，Redis 会将事务队列中的所有命令按先进先出的顺序执行。当事务队列里的命令全部执行完毕后，Redis 会返回一个数组，包含每个命令的执行结果。

discard 命令用于取消一个事务，它会清空事务队列并退出事务状态。

watch 命令用于监视一个或者多个 key，如果这个 key 在事务执行之前 被其他命令改动，那么事务将会被打断。

但 Redis 的事务与 MySQL 的有很大不同，它并不支持回滚，也不支持隔离级别。

说一下 Redis 事务的原理？

Redis 事务的原理并不复杂，核心就是一个"先排队，后执行"的机制。

当执行 MULTI 命令时，Redis 会给这个客户端打一个事务的标记，表示这个客户端后面发送的命令不会被立即执行，而是被放到一个队列里排队等着。

当 Redis 收到 EXEC 命令时，它会把队列里的命令一个个拿出来执行。因为 Redis 是单线程的，所以这个过程不会被其他命令打断，这就保证了Redis 事务的原子性。

当执行 WATCH 命令时，Redis 会将 key 添加到全局监视字典中；只要这些 key 在 EXEC 前被其他客户端修改，Redis 就会给相关客户端打上脏标记，EXEC 时发现事务已被干扰就会直接取消整个事务。

// 全局监视字典
dict *watched_keys;

typedef struct watchedKey {
    robj *key;
    redisDb *db;
} watchedKey;

DISCARD 做的事情很简单直接，首先检查客户端是否真的在事务状态，如果不在就报错；如果在事务状态，就清空事务队列并退出事务状态。

void discardCommand(client *c) {
    if (!(c->flags & CLIENT_MULTI)) {
        addReplyError(c,"DISCARD without MULTI");
        return;
    }
    discardTransaction(c);
    addReply(c,shared.ok);
}

Redis 事务有哪些注意点？

最重要的的一点是，Redis 事务不支持回滚，一旦 EXEC 命令被调用，所有命令都会被执行，即使有些命令可能执行失败。

Redis事务为什么不支持回滚？

Redis 的核心设计理念是简单、高效，而不是完整的 ACID 特性。而实现回滚需要在执行过程中保存大量的状态信息，并在发生错误时逆向执行命令以恢复原始状态。这会增加 Redis 的复杂性和性能开销。

Redis事务满足原子性吗？要怎么改进？

Redis 的事务不能满足标准的原子性，因为它不支持事务回滚，也就是说，假如某个命令执行失败，整个事务并不会自动回滚到初始状态。

// 一个转账事务
redisTemplate.multi();
redisTemplate.opsForValue().decrement("user:1:balance", 100); // 成功
redisTemplate.opsForList().leftPush("user:1:balance", "log");  // 类型错误，失败
redisTemplate.opsForValue().increment("user:2:balance", 100);  // 还是会执行
List<Object> results = redisTemplate.exec();

// 结果：用户1被扣了钱，用户2也收到了钱，但中间的日志操作失败了
// 这符合Redis的原子性定义，但不符合业务期望

可以使用 Lua 脚本来替代事务，脚本运行期间，Redis 不会处理其他命令，并且我们可以在脚本中处理整个业务逻辑，包括条件检查和错误处理，保证要么执行成功，要么保持最初的状态，不会出现一个命令执行失败、其他命令执行成功的情况。

@Service
public class ImprovedTransactionService {
    
    public boolean atomicTransfer(String fromUser, String toUser, int amount) {
        String luaScript = 
            "local from_key = KEYS[1] " +
            "local to_key = KEYS[2] " +
            "local amount = tonumber(ARGV[1]) " +
            
            // 检查转出账户余额
            "local from_balance = redis.call('GET', from_key) " +
            "if not from_balance then return -1 end " +
            
            "from_balance = tonumber(from_balance) " +
            "if from_balance < amount then return -2 end " +
            
            // 检查转入账户是否存在
            "if redis.call('EXISTS', to_key) == 0 then return -3 end " +
            
            // 所有检查通过，执行转账
            "redis.call('DECRBY', from_key, amount) " +
            "redis.call('INCRBY', to_key, amount) " +
            
            // 记录转账日志
            "local log = from_key .. ':' .. to_key .. ':' .. amount " +
            "redis.call('LPUSH', 'transfer:log', log) " +
            
            "return 1";
        
        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
        script.setScriptText(luaScript);
        script.setResultType(Long.class);
        
        Long result = redisTemplate.execute(script, 
            Arrays.asList("user:" + fromUser + ":balance", "user:" + toUser + ":balance"),
            amount);
        
        return result != null && result == 1;
    }
}

Redis 事务的 ACID 特性如何体现？

单个 Redis 命令的执行是原子性的，但 Redis 没有在事务上增加任何维持原子性的机制，所以 Redis 事务在执行过程中如果某个命令失败了，其他命令还是会继续执行，不会回滚。

一致性指的是，如果数据在执行事务之前是一致的，那么在事务执行之后，无论事务是否执行成功，数据也应该是一致的。但 Redis 事务并不保证一致性，因为如果事务中的某个命令失败了，其他命令仍然会执行，就会出现数据不一致的情况。

Redis 是单线程执行事务的，并且不会中断，直到执行完所有事务队列中的命令为止。因此，我认为 Redis 的事务具有隔离性的特征。

Redis 事务的持久性完全依赖于 Redis 本身的持久化机制，如果开启了 AOF，那么事务中的命令会作为一个整体记录到 AOF 文件中，当然也要看 AOF 的 fsync 策略。

如果只开启了 RDB，事务中的命令可能会在下次快照前丢失。如果两个都没有开启，肯定是不满足持久性的。

1. Java 面试指南（付费）收录的华为一面原题：说下 Redis 事务
2. 二哥编程星球球友枕云眠美团 AI 面试原题：什么是 redis 的事务，它的 ACID 属性如何体现
3. Java 面试指南（付费）收录的快手同学 4 一面原题：Redis事务满足原子性吗？要怎么改进？

memo：2025 年 5 月 29 日，今天给球友修改简历时，碰到一个东南大学本硕博 3 985 的球友，这也是我已知信息中学历最高的球友了。

46.有Lua脚本操作Redis的经验吗？

Lua 脚本是处理 Redis 复杂操作的首选方案，比如说原子扣减库存、分布式锁、限流等业务场景，都可以通过 Lua 脚本来实现。

在秒杀场景下，可以用 Lua 脚本把所有检查逻辑都写在一起：先看库存够不够，再看用户有没有买过，所有条件都满足才扣减库存。因为整个脚本是原子执行的，Redis 在执行期间不会处理其他命令，所以可以彻底解决超卖问题。

// 这个秒杀脚本救了我的命
String luaScript = 
    "local stock = redis.call('GET', KEYS[1]) " +
    "if not stock or tonumber(stock) < tonumber(ARGV[2]) then " +
    "    return -1 " +  // 库存不足
    "end " +
    "if redis.call('SISMEMBER', KEYS[2], ARGV[1]) == 1 then " +
    "    return -2 " +  // 重复购买
    "end " +
    "redis.call('DECRBY', KEYS[1], ARGV[2]) " +
    "redis.call('SADD', KEYS[2], ARGV[1]) " +
    "return 1";

在分布式锁场景下，我一开始用的 SETNX 命令来实现，结果发现如果程序异常退出，锁就死掉了。后来加了过期时间，但又发现可能误删其他线程的锁。最后还是用 Lua 脚本彻底解决了这个问题，确保只有锁的持有者才能释放锁。

// 解锁脚本特别重要，必须验证是自己的锁才能删
private final String UNLOCK_SCRIPT = 
    "if redis.call('GET', KEYS[1]) == ARGV[1] then " +
    "    return redis.call('DEL', KEYS[1]) " +
    "else " +
    "    return 0 " +
    "end";

甚至还可以用 Lua脚本实现滑动窗口限流器，一次性完成过期数据清理、计数检查、新记录添加三个操作，而且完全原子化。

// 滑动窗口限流，逻辑清晰，性能还好
String luaScript = 
    "local key = KEYS[1] " +
    "local now = tonumber(ARGV[1]) " +
    "local window = tonumber(ARGV[2]) " +
    "local limit = tonumber(ARGV[3]) " +
    
    // 先清理过期记录
    "redis.call('ZREMRANGEBYSCORE', key, 0, now - window) " +
    
    // 检查当前请求数
    "local current = redis.call('ZCARD', key) " +
    "if current < limit then " +
    "    redis.call('ZADD', key, now, now) " +
    "    return 1 " +
    "else " +
    "    return 0 " +
    "end"; 

memo：2025 年 5 月 30 日，今天有球友在星球里发消息说拿到了金山办公的 offer，问我该选 cpp 还是go，我的建议大家可以看看符合是否合理，不管如何选择，真的恭喜球友！

47.Redis的管道Pipeline了解吗？

了解，Pipeline 允许客户端一次性向 Redis 服务器发送多个命令，而不必等待一个命令响应后才能发送下一个。Redis 服务器会按照命令的顺序依次执行，并将所有结果打包返回给客户端。

正常情况下，每执行一个 Redis 命令都需要一次网络往返：发送命令 -> 等待响应 -> 发送下一个命令。

客户端                    Redis服务器
  |                           |
  |------- SET key1 val1 ---->|
  |<------ OK ---------------|
  |------- SET key2 val2 ---->|
  |<------ OK ---------------|
  |------- GET key1 -------->|
  |<------ val1 -------------|

如果大量请求依次发送，网络延迟会显著增加请求的总执行时间，假如一次 RTT 的时间是 1 毫秒，3 个就是 3 毫秒。有了 Pipeline 后，可以一次性发送 3 个命令，总时间就只需要 1 毫秒。

@Service
public class RedisBatchService {
    
    public void batchInsertUsers(List<User> users) {
        // 不用Pipeline的错误做法 - 很慢
        // for (User user : users) {
        //     redisTemplate.opsForValue().set("user:" + user.getId(), user);
        // }
        
        // 使用Pipeline的正确做法
        redisTemplate.executePipelined(new RedisCallback<Object>() {
            @Override
            public Object doInRedis(RedisConnection connection) throws DataAccessException {
                for (User user : users) {
                    String key = "user:" + user.getId();
                    byte[] keyBytes = key.getBytes();
                    byte[] valueBytes = serialize(user);
                    
                    connection.set(keyBytes, valueBytes);
                }
                return null; // Pipeline不需要返回值
            }
        });
    }
}

当然了，Pipeline 不是越大越好，太大会占用过多内存，通常建议每个 Pipeline 包含 1000 到 5000 个命令。可以根据实际情况调整。

public void smartBatchInsert(List<String> data) {
    int batchSize = 1000; // 经验值，根据数据大小调整
    
    for (int i = 0; i < data.size(); i += batchSize) {
        List<String> batch = data.subList(i, Math.min(i + batchSize, data.size()));
        
        redisTemplate.executePipelined(new RedisCallback<Object>() {
            @Override
            public Object doInRedis(RedisConnection connection) throws DataAccessException {
                for (String item : batch) {
                    connection.set(item.getBytes(), item.getBytes());
                }
                return null;
            }
        });
    }
}

什么场景下适合使用 Pipeline呢？

需要批量插入、更新或删除数据，或者需要执行大量相似的命令时。比如：系统启动时的缓存预热 -> 批量加载热点数据；比如统计数据的批量更新；比如大批量数据的导入导出；比如批量删除过期或无效的缓存。

有了解过 Pipeline 的底层原理吗？

有，其实就是缓冲的思想。在技术派实战项目中，我就在 RedisClient 类中封装了一个 PipelineAction 内部类，用来缓存命令。

add 方法将命令包装成 Runnable 对象，放入 List 中。当执行 execute 方法时，再调用 RedisTemplate 的 executePipelined 方法开启管道模式将多个命令发送到 Redis 服务端。

Redis 服务端从输入缓冲区读到命令后，会按照 RESP 协议进行命令拆解，再依次执行这些命令。执行结果会写入到输出缓冲区，最后再将所有结果一次性返回给客户端。

typedef struct client {
    sds querybuf;           // 输入缓冲区
    list *reply;            // 输出缓冲区链表
    unsigned long reply_bytes; // 输出缓冲区大小
} client;

1. Java 面试指南（付费）收录的京东面经同学 8 面试原题：对pipeline的理解，什么场景适合使用pipeline？有了解过pipeline的底层？

memo：2025 年 6 月 1 日，今天有球友在星球里发消息说拿到了百得思维的offer，他是民办二本，对这个结果很满意，也很感谢面渣逆袭和星球的实战项目，让他摆脱了浑浑噩噩的日子。恭喜他！

48.🌟Redis能实现分布式锁吗？

分布式锁是一种用于控制多个不同进程在分布式系统中访问共享资源的锁机制。它能确保在同一时刻，只有一个节点可以对资源进行访问，从而避免分布式场景下的并发问题。

可以使用 Redis 的 SETNX 命令实现简单的分布式锁。比如 SET key value NX PX 3000 就创建了一个锁名为 key 的分布式锁，锁的持有者为 value。NX 保证只有在 key 不存在时才能创建成功，EX 设置过期时间用以防止死锁。

Redis如何保证 SETNX 不会发生冲突？

当我们使用 SET key value NX EX 30 这个命令进行加锁时，Redis 会把整个操作当作一个原子指令来执行。因为 Redis 的命令处理是单线程的，所以在同一时刻只能有一个命令在执行。

比如说两个客户端 A 和 B 同时请求同一个锁：

客户端A: SET lock_key uuid_a NX EX 30
客户端B: SET lock_key uuid_b NX EX 30

虽然这两个请求可能几乎同时到达 Redis 服务器，但 Redis 会严格按照到达的先后顺序来处理。假设 A 的请求先到，Redis 会先执行 A 的 SET 命令，这时 lock_key 被设置为 uuid_a。

当处理 B 的请求时，因为 lock_key 已经存在了，NX 条件不满足，所以 B 的 SET 命令会失败，返回 NULL。这样就保证了只有 A 能获取到锁。

关键点在于 NX 的语义：NOT EXISTS，只有在 key 不存在的时候才会设置成功。Redis 在执行这个命令时，会先检查 key 是否存在，如果不存在才会设置值，这整个过程是原子的，不会被其他命令打断。

SETNX有什么问题，如何解决？

使用 SETNX 创建分布式锁时，虽然可以通过设置过期时间来避免死锁，但会误删锁。比如线程 A 获取锁后，业务执行时间比较长，锁过期了。这时线程 B 获取到锁，但线程 A 执行完业务逻辑后，会尝试删除锁，这时候删掉的其实是线程 B 的锁。

可以通过锁的自动续期机制来解决锁过期的问题，比如 Redisson 的看门狗机制，在后台启动一个定时任务，每隔一段时间就检查锁是否还被当前线程持有，如果是就自动延长过期时间。这样既避免了死锁，又防止了锁被提前释放。

memo：2025 年 6 月 2 日修改至此，今天在帮一个学院本球友分析 offer 选择后，他又回复说多亏了星球才能一路走到现在，很满足这个结果。看多了拿大厂 offer 球友的感谢，看到学院本也能取得满意的成绩，我也很开心。

Redisson了解多少？

Redisson 是一个基于 Redis 的 Java 客户端，它不只是对 Redis 的操作进行简单地封装，还提供了很多分布式的数据结构和服务，比如最常用的分布式锁。

RLock lock = redisson.getLock("lock");
lock.lock();
try {
    // do something
} finally {
    lock.unlock();
}

Redisson 的分布式锁比 SETNX 完善的得多，它的看门狗机制可以让我们在获取锁的时候省去手动设置过期时间的步骤，它在内部封装了一个定时任务，每隔 10 秒会检查一次，如果当前线程还持有锁就自动续期 30 秒。

private Long tryAcquire(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
    return get(tryAcquireAsync(waitTime, leaseTime, unit, threadId));
}

private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
    RFuture<Long> ttlRemainingFuture;
    if (leaseTime != -1) {
        // 手动设置过期时间
        ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
    } else {
        // 启用看门狗机制，使用默认的30秒过期时间
        ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime,
                TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
    }
    
    // 处理获取锁成功的情况
    ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
        if (e != null) {
            return;
        }
        // 如果获取锁成功且启用看门狗机制
        if (ttlRemaining == null) {
            if (leaseTime != -1) {
                internalLockLeaseTime = unit.toMillis(leaseTime);
            } else {
                scheduleExpirationRenewal(threadId); // 启动看门狗
            }
        }
    });
    return ttlRemainingFuture;
}

另外，Redisson 还提供了分布式限流器 RRateLimiter，基于令牌桶算法实现，用于控制分布式环境下的访问频率。

// API 接口限流
@RestController
public class ApiController {
    
    @Autowired
    private RedissonClient redissonClient;
    
    @GetMapping("/api/data")
    public ResponseEntity<?> getData() {
        RRateLimiter limiter = redissonClient.getRateLimiter("api.data");
        limiter.trySetRate(RateType.OVERALL, 100, 1, RateIntervalUnit.MINUTES);
        
        if (limiter.tryAcquire()) {
            // 处理请求
            return ResponseEntity.ok(processData());
        } else {
            // 限流触发
            return ResponseEntity.status(429).body("Rate limit exceeded");
        }
    }
}

详细说说Redisson的看门狗机制？

Redisson 的看门狗机制是一种自动续期机制，用于解决分布式锁的过期问题。

基本原理是这样的：当调用 lock() 方法加锁时，如果没有显式设置过期时间，Redisson 会默认给锁加一个 30 秒的过期时间，同时启用一个名为“看门狗”的定时任务，每隔 10 秒（默认是过期时间的 1/3），去检查一次锁是否还被当前线程持有，如果是，就自动续期，将过期时间延长到 30 秒。

// 伪代码展示核心逻辑
private void renewExpiration() {
    Timeout task = commandExecutor.getConnectionManager()
        .newTimeout(new TimerTask() {
            @Override
            public void run(Timeout timeout) {
                // 用 Lua 脚本检查并续期
                if (redis.call("get", lockKey) == currentThreadId) {
                    redis.call("expire", lockKey, 30);
                    // 递归调用，继续下一次续期
                    renewExpiration();
                }
            }
        }, 10, TimeUnit.SECONDS);
}

续期的 Lua 脚本会检查锁的 value 是否匹配当前线程，如果匹配就延长过期时间。这样就能保证只有锁的真正持有者才能续期。

当调用 unlock() 方法时，看门狗任务会被取消。或者如果业务逻辑执行完但忘记 unlock 了，看门狗也会帮我们自动检查锁，如果锁已经不属于当前线程了，也会自动停止续期。

这样我们就不用担心业务执行时间过长导致锁被提前释放，也避免了手动估算过期时间的麻烦，同时也解决了分布式环境下的死锁问题。

看门狗机制中的检查锁过程是原子操作吗？

是的，Redisson 使用了 Lua 脚本来保证锁检查的原子性。

Redis 在执行 Lua 脚本时，会把整个脚本当作一个命令来处理，期间不会执行其他命令。所以 hexists 检查和 expire 续期是原子执行的。

Redlock你了解多少？

Redlock 是 Redis 作者 antirez 提出的一种分布式锁算法，用于解决单个 Redis 实例作为分布式锁时存在的单点故障问题。

Redlock 的核心思想是通过在多个完全独立的 Redis 实例上同时获取锁来实现容错。

minLocksAmount 方法返回的 locks.size()/2 + 1，正是 Redlock 算法要求的少数服从多数原则。failedLocksLimit 方法会计算允许失败的锁数量，确保即使部分实例失败，只要成功的实例数量超过一半就认为获取锁成功。

红锁会尝试依次向所有 Redis 实例获取锁，并记录成功获取的锁数量，当数量达到 minLocksAmount 时就认为获取成功，否则释放已获取的锁并返回失败。

虽然 Redlock 存在一些争议，比如说时钟漂移问题、网络分区导致的脑裂问题，但它仍然是一个相对成熟的分布式锁解决方案。

红锁能不能保证百分百上锁？

不能，Redlock 无法保证百分百上锁成功，这是由分布式系统的本质特性决定的。

当有网络分区时，客户端可能无法与足够数量的 Redis 实例通信。比如在 5 个 Redis 实例的部署中，如果网络分区导致客户端只能访问到 2 个实例，那么无论如何都无法满足红锁要求的少数服从多数原则，获取锁的时候必然失败。

public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
    // ...
    for (ListIterator<RLock> iterator = locks.listIterator(); iterator.hasNext();) {
        RLock lock = iterator.next();
        boolean lockAcquired;
        try {
            lockAcquired = lock.tryLock(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS);
        } catch (RedisResponseTimeoutException e) {
            lockAcquired = false; // 网络超时导致失败
        } catch (Exception e) {
            lockAcquired = false; // 其他异常导致失败
        }
        
        // 如果剩余可尝试的实例数量不足以达到多数派，直接退出
        if (locks.size() - acquiredLocks.size() == failedLocksLimit()) {
            break;
        }
    }
    
    // 检查是否达到多数派要求
    if (acquiredLocks.size() >= minLocksAmount(locks)) {
        return true;
    } else {
        unlockInner(acquiredLocks);
        return false; // 未达到多数派，获取失败
    }
}

时钟漂移也会影响成功率。即使所有实例都可达，如果各个 Redis 实例之间存在明显的时钟漂移，或者客户端在获取锁的过程中耗时过长，比如网络延迟、GC 停顿等，都可能会导致锁在获取完成前就过期，从而获取失败。

在实际应用中，可以通过重试机制来提高锁的成功率。

for (int i = 0; i < maxRetries; i++) {
    if (redLock.tryLock(waitTime, leaseTime, TimeUnit.MILLISECONDS)) {
        return true;
    }
    Thread.sleep(retryDelay);
}
return false;

项目中有用到分布式锁吗？

在PmHub项目中，我有使用 Redission 的分布式锁来确保流程状态的更新按顺序执行，且不被其他流程服务干扰。

1. Java 面试指南（付费）收录的腾讯 Java 后端实习一面原题：分布式锁用了 Redis 的什么数据结构
2. Java 面试指南（付费）收录的小公司面经合集同学 1 Java 后端面试原题：Redisson 的底层原理？以及与 SETNX 的区别？
3. Java 面试指南（付费）收录的百度面经同学 1 文心一言 25 实习 Java 后端面试原题：redis 分布式锁的实现原理？setnx？
4. Java 面试指南（付费）收录的小米同学 F 面试原题：自己实现 redis 分布式锁的坑（主动提了 Redission）
5. Java 面试指南（付费）收录的腾讯云智面经同学 20 二面面试原题：redission 的原理是什么？ setnx + lua 脚本？
6. Java 面试指南（付费）收录的收钱吧面经同学 1 Java 后端一面面试原题：系统里面分布式锁是怎么做的？你提到了redlock，那它机制是怎么样的？红锁能不能保证百分百上锁？
7. Java 面试指南（付费）收录的字节跳动面经同学 21 抖音商城一面面试原题：加分布式锁时redis如何保证不会发生冲突？分布式锁过期怎么办？
8. Java 面试指南（付费）收录的拼多多面经同学 8 一面面试原题：Redis分布式锁如何实现的
9. Java 面试指南（付费）收录的百度同学 4 面试原题：Setnx,知道吗? 用这个加锁有什么问题吗?怎么解决?
10. Java 面试指南（付费）收录的阿里系面经同学 19 饿了么面试原题：分布式锁用redis实现思路
11. Java 面试指南（付费）收录的京东面经同学 9 面试原题：redis的分布式锁有了解过吗
12. Java 面试指南（付费）收录的同学 30 腾讯音乐面试原题：redis锁有几种实现方式

memo：2025 年 6 月 3 日修改至此，今天在修改球友的简历时，碰到一个 211 本科、北京大学软微学院的球友，我只能说，星球的球友真是人才济济啊！祝大家都有一个美好的未来。

底层结构

49.🌟Redis都有哪些底层数据结构？

Redis 之所以快，除了基于内存读写之外，还有很重要的一点就是它精心设计的底层数据结构。Redis 总共有 8 种核心的底层数据结构，我按照重要程度来说一下。

首先是 SDS，这是 Redis 自己实现的动态字符串，它保留了 C 语言原生的字符串长度，所以获取长度的时间复杂度是 O(1)，在此基础上还支持动态扩容，以及存储二进制数据。

然后是字典，更底层是用数组+链表实现的哈希表。它的设计很巧妙，用了两个哈希表，平时用第一个，rehash 的时候用第二个，这样可以渐进式地进行扩容，不会阻塞太久。

接下来压缩列表 ziplist，这个设计很有意思。Redis 为了节省内存，设计了这种紧凑型的数据结构，把所有元素连续存储在一块内存里。但是它有个致命问题叫"连锁更新"，就是当我们修改一个元素的时候，可能会导致后面所有的元素都要重新编码，性能会急剧下降。

为了解决压缩列表的问题，Redis 后来设计了 quicklist。这个设计思路很聪明，它把 ziplist 拆分成小块，然后用双向链表把这些小块串起来。这样既保持了 ziplist 节省内存的优势，又避免了连锁更新的问题，因为每个小块的 ziplist 都不会太大。

再后来，Redis 又设计了 listpack，这个可以说是 ziplist 的完美替代品。它最大的特点是每个元素只记录自己的长度，不记录前一个元素的长度，这样就彻底解决了连锁更新的问题。Redis 5.0 已经用 listpack 替换了 ziplist。

跳表skiplist 主要用在 ZSet 中。它的设计很巧妙，通过多层指针来实现快速查找，平均时间复杂度是 O(log N)。相比红黑树，跳表的实现更简单，而且支持范围查询，这对 Redis 的有序集合来说很重要。

还有整数集合intset，当 Set 中都是整数且元素数量较少时使用，内部是一个有序数组，查找用的二分法。

最后是双向链表LinkedList，早期版本的 Redis 会在 List 中用到，但 Redis 3.2 后就被 quicklist 替代了，因为纯链表的问题是内存不连续，影响 CPU 缓存性能。

memo：2025 年 6 月 4 日，今天有球友发喜报说拿到了京东零售的实习 offer，并且部门和业务还是挺不错的，恭喜他！6 月份还有机会，冲。

简单介绍下链表？

Redis 的 linkedlist 是⼀个双向⽆环链表结构，和 Java 中的 LinkedList 类似。

节点由 listNode 表示，每个节点都有指向其前置节点和后置节点的指针，头节点的前置和尾节点的后置均指向 null。

关于整数集合，能再详细说说吗？

整数集合是 Redis 中一个非常精巧的数据结构，当一个 Set 只包含整数元素，并且数量不多时，默认不超过 512 个，Redis 就会用 intset 来存储这些数据。

intset 最有意思的地方是类型升级机制。它有三种编码方式：16位、32位和 64位，会根据存储的整数大小动态调整。比如原来存的都是小整数，用 16 位编码就够了，但突然插入了一个很大的数，超出了 16 位的范围，这时整个数组会升级到 32 位编码。

typedef struct intset {
    uint32_t encoding;   // 编码方式：16位、32位或64位
    uint32_t length;     // 元素数量
    int8_t contents[];   // 保存元素的数组
} intset;

当然了，这种升级是有代价的，因为需要重新分配内存并复制数据，并且是不可逆的，但它的好处是可以节省内存空间，特别是在存储大量小整数时。

另外，所有元素在数组中按照从小到大的顺序排列，这样就可以使用二分查找来定位元素，时间复杂度为 O(log N)。

说一下zset 的底层原理？

ZSet 是 Redis 最复杂的数据类型，它有两种底层实现方式：压缩列表和跳表。

当保存的元素数量少于 128 个，且保存的所有元素大小都小于 64 字节时，Redis 会采用压缩列表的编码方式；否则就用跳表。

当然，这两个条件都可以通过参数进行调整。

选择压缩列表作为底层实现时，每个元素会使用两个紧挨在一起的节点来保存：第一个节点保存元素的成员，第二个节点保存元素的分值。

所有元素按分值从小到大有序排列，小的放在靠近表头的位置，大的放在靠近表尾的位置。

但跳表的缺点是查找只能按顺序进行，时间复杂度为 O(N)，而且在最坏的情况下，插入和删除操作还可能会引起连锁更新。

当元素数量较多或元素较大时，Redis 会使用 skiplist 的编码方式；这个设计非常的巧妙，同时使用了两种数据结构：

typedef struct zset {
    zskiplist *zsl;  // 跳跃表
    dict *dict;      // 字典
} zset;

跳表按分数有序保存所有元素，且支持范围查询（如 ZRANGE、ZRANGEBYSCORE），平均时间复杂度为 O(log N)。而哈希表则用来存储成员和分值的映射关系，查找时间复杂度为 O(1)。

虽然同时使用两种结构，但它们会通过指针来共享相同元素的成员和分值，因此不会浪费额外的内存。

你知道为什么Redis 7.0要用listpack来替代ziplist吗？

答：主要是为了解决压缩列表的一个核心问题——连锁更新。在压缩列表中，每个节点都需要记录前一个节点的长度信息。

当插入或删除一个节点时，如果这个操作导致某个节点的长度发生了变化，那么后续的节点可能都需要更新它们存储的"前一个节点长度"字段。最坏的情况下，一次操作可能触发整个链表的更新，时间复杂度会从 O(1)退化到 O(n²)。

而 listpack 的设计理念完全不同。它让每个节点只记录自己的长度信息，不再依赖前一个节点的长度。这样就从根本上避免了连锁更新的问题。

listpack 中的节点不再保存其前一个节点的长度，而是保存当前节点的编码类型、数据和长度。

连锁更新是怎么发生的？

比如说我们有一个压缩列表，其中有几个节点的长度都是 253 个字节。在 ziplist 的编码中，如果前一个节点的长度小于 254 字节，我们只需要 1 个字节来存储这个长度信息。

但如果在这些节点前面插入一个长度为 254 字节的节点，那么原来只需要 1 个字节存储长度的节点现在需要 5 个字节来存储长度信息。这就会导致后续所有节点的长度信息都需要更新。

1. Java 面试指南（付费）收录的字节跳动商业化一面的原题：说说 Redis 的 zset，什么是跳表，插入一个节点要构建几层索引
2. Java 面试指南（付费）收录的字节跳动面经同学 9 飞书后端技术一面面试原题：Redis 的数据类型，ZSet 的实现
3. Java 面试指南（付费）收录的小米暑期实习同学 E 一面面试原题：你知道 Redis 的 zset 底层实现吗
4. Java 面试指南（付费）收录的腾讯面经同学 23 QQ 后台技术一面面试原题：zset 的底层原理
5. Java 面试指南（付费）收录的快手面经同学 7 Java 后端技术一面面试原题：说一下 ZSet 底层结构
6. Java 面试指南（付费）收录的美团同学 9 一面面试原题：redis的数据结构底层原理？
7. Java 面试指南（付费）收录的腾讯面经同学 27 云后台技术一面面试原题：Zset的底层实现？
8. Java 面试指南（付费）收录的得物面经同学 9 面试题目原题：Zset的底层如何实现？

memo：2025 年 6 月 5 日，今天有球友在VIP群里咨询 offer 的选择，一个拼多多，一个快手，真让人羡慕的要死啊，😄

50.Redis 为什么不用 C 语言的原生字符串？

第一，C 语言的字符串其实就是字符数组，以 \0 结尾，这意味着如果数据本身包含 \0 字节，就会被误认为字符串结束。但 Redis 需要存储各种类型的数据，包括图片、序列化对象等二进制数据，这些数据中很可能包含 \0。

第二，如果需要获取字符串长度，C 语言只能调用 strlen() 函数，时间复杂度是 O(N)，因为要遍历整个字符串直到遇到 \0。

第三，C 语言的字符串不会自动检查边界，如果往一个字符数组里写入超过其容量的数据，就会出现缓冲区溢出。

第四，C 语言的字符串不支持动态扩容，如果需要修改内容，就必须重新分配内存并复制数据，开销很大。

Redis 设计的 SDS 完美解决了这些问题，获取长度可以直接通过 len 字段，时间复杂度为 O(1)；free 字段会记录剩余空间，因此 Redis 可以根据预分配策略动态扩容，不用在追加数据时重新分配内存；并且不依赖于 \0 结尾，可以存储任意二进制数据。

struct sds {
    int len;        // 字符串长度
    int free;       // 剩余空间
    char buf[];     // 字符数组
}

51.你研究过 Redis 的字典源码吗？

是的，有研究过。Redis 的字典分为三层，最外层是一个 dict 结构，包含两个哈希表 ht[0] 和 ht[1]，用于存储键值对。每个哈希表由一个数组和链表组成，数组用于快速定位，链表用于解决哈希冲突。

// 最外层的字典结构
typedef struct dict {
    dictht ht[2];       // 两个哈希表！这是关键
    long rehashidx;     // rehash索引，-1表示没有进行rehash
    // ...
} dict;

// 哈希表结构
typedef struct dictht {
    dictEntry **table;  // 哈希表数组
    unsigned long size; // 哈希表大小
    unsigned long sizemask; // 哈希表大小掩码，用于计算索引值
    unsigned long used; // 该哈希表已有节点的数量
} dictht;

// 哈希表节点
typedef struct dictEntry {
    void *key;              // 键
    v;                 // 值
    struct dictEntry *next; // 指向下个哈希表节点，形成链表
} dictEntry;

字典最核心的特点是渐进式 rehash，这是我觉得最精彩的部分。传统的哈希表扩容都是一次性完成的，但 Redis 不是这样的。

当负载因子触发 rehash 条件时，Redis 会为哈希表1 分配新的空间，通常是哈希表 0 的两倍大小，然后将 rehashidx 设置为 0。

接下来的关键是，Redis 不会一次性把所有数据从哈希表0 迁移到哈希表1，而是每次操作字典时，顺便迁移哈希表0 中 rehashidx 位置上的所有键值对。迁移完一个槽位后，rehashidx 递增，直到整个哈希表0 迁移完毕。

这种设计的巧妙之处在于把 rehash 的开销分摊到了每次操作中。假设有一个几百万键的哈希表，如果一次性 rehash 可能需要几百毫秒，这对单线程的 Redis 来说是灾难性的。但通过渐进式 rehash，每次操作只增加很少的额外开销，用户基本感觉不到延迟。

在 rehash 期间，查找操作会先查 哈希表 0，没找到再查哈希表 1；但是新插入的数据只会放到哈希表 1 中。这样既可以保证数据的完整性，又能避免数据的重复。

遇到哈希冲突怎么办？

Redis 是通过链地址法来解决哈希冲突的，每个哈希表的槽位实际上是一个链表的头指针，当多个键的哈希值映射到同一个槽位时，这些键会以链表的形式串联起来。

具体实现上，Redis 会通过哈希表节点的 next 指针，指向下一个具有相同哈希值的节点。当发生冲突时，新的键值对会插入到链表的头部，时间复杂度是 O(1)。查找时需要遍历整个链表，最坏的情况下时间复杂度为 O(n)，但通常链表都比较短。

另外，Redis 设计的哈希函数在分布上也比较均匀，能够有效减少哈希冲突的发生。

/* MurmurHash2, by Austin Appleby
 * Note - This code makes a few assumptions about how your machine behaves -
 * 1. We can read a 4-byte value from any address without crashing
 * 2. sizeof(int) == 4
 *
 * And it has a few limitations -
 *
 * 1. It will not work incrementally.
 * 2. It will not produce the same results on little-endian and big-endian
 *    machines.
 */
unsigned int dictGenHashFunction(const void *key, int len) {
    /* 'm' and 'r' are mixing constants generated offline.
       They're not really 'magic', they just happen to work well.  */
    uint32_t seed = dict_hash_function_seed;
    const uint32_t m = 0x5bd1e995;
    const int r = 24;

    /* Initialize the hash to a 'random' value */
    uint32_t h = seed ^ len;

    /* Mix 4 bytes at a time into the hash */
    const unsigned char *data = (const unsigned char *)key;

    while(len >= 4) {
        uint32_t k = *(uint32_t*)data;

        k *= m;
        k ^= k >> r;
        k *= m;

        h *= m;
        h ^= k;

        data += 4;
        len -= 4;
    }

    /* Handle the last few bytes of the input array  */
    switch(len) {
    case 3: h ^= data[2] << 16;
    case 2: h ^= data[1] << 8;
    case 1: h ^= data[0]; h *= m;
    };

    /* Do a few final mixes of the hash to ensure the last few
       * bytes are well-incorporated. */
    h ^= h >> 13;
    h *= m;
    h ^= h >> 15;

    return (unsigned int)h;
}

memo：2025 年 6 月 6 日，今天有球友咨询去金山办公暑期实习，要提前学点什么？又一个凭借 Java 这个载体拿到 Go offer 的球友，说明大家在求职 Go 岗的时候，也不用说非要提前刻意去学习 Go，当然有一些基础是最好的，我之前也整理过 Go 的学习路线在 Java 进阶之路上。

52.🌟你了解跳表吗？

跳表是一种非常巧妙的数据结构，它在有序链表的基础上建立了多层索引，最底层包含所有数据，每往上一层，节点数量就减少一半。

它的核心思想是"用空间换时间"，通过多层索引来跳过大量节点，从而提高查找效率。

每个节点有 50% 的概率只在第 1 层出现，25% 的概率在第 2 层出现，依此类推。查找的时候从最高层开始水平移动，当下一个节点值大于目标时，就向下跳一层，直到找到目标节点。

怎么往跳表插入节点呢？

首先是找到插入位置，从最高层的头节点开始，在每一层都找到应该插入位置的前驱节点，用一个 update 数组把这些前驱节点记录下来。这个查找过程和普通查找一样，在每层向右移动直到下个节点的值大于要插入的值，然后下降到下一层。

// 记录每层的插入位置
zskiplistNode *update[ZSKIPLIST_MAXLEVEL];
zskiplistNode *x;
int i, level;

// 从最高层开始查找
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
    // 在当前层水平移动，找到插入位置
    while (x->level[i].forward &&
           (x->level[i].forward->score < score ||
            (x->level[i].forward->score == score &&
             sdscmp(x->level[i].forward->ele, ele) < 0)))
    {
        x = x->level[i].forward;
    }
    update[i] = x;  // 记录每层的前驱节点
}

接下来随机生成新节点的层数。通常用一个循环，每次有 50% 的概率继续往上，直到随机失败或达到最大层数限制。

// Redis 中的随机层数生成
int zslRandomLevel(void) {
    int level = 1;
    while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
        level += 1;
    return (level < ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}

// 生成新节点的层数
level = zslRandomLevel();

创建新节点后，从底层开始到新节点的最高层，在每一层都进行标准的链表插入操作。这一步要利用之前记录的 update 数组，将新节点插入到正确位置，然后更新前后指针的连接关系。

// 更新前进指针
for (i = 0; i < level; i++) {
    x->level[i].forward = update[i]->level[i].forward;
    update[i]->level[i].forward = x;
    
    // 更新跨度信息
    x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
    update[i]->level[i].span = (rank[0] - rank[i]) + 1;
}

// 更新未涉及层的跨度
for (i = level; i < zsl->level; i++) {
    update[i]->level[i].span++;
}

// 更新后退指针
x->backward = (update[0] == zsl->header) ? NULL : update[0];
if (x->level[0].forward)
    x->level[0].forward->backward = x;
else
    zsl->tail = x;

// 更新跳表长度
zsl->length++;

我们来模拟一个跳表的插入过程，假设插入的数据依次是 22、19、7、3、37、11、26。

那假如我们在一个已经分布了 1、14、27、31、44、56、63、70、80、91 的跳表中插入一个 67 的节点，插入过程是这样的：

zset为什么要使用跳表呢？

第一，跳表天然就是有序的数据结构，查找、插入和删除都能保持 O(log n) 的时间复杂度。

第二，跳表支持范围查询，找到起始位置后可以直接沿着底层链表顺序遍历，满足 ZRANGE 按排名获取元素，或者 ZRANGEBYSCORE 按分值范围获取元素。

memo：2025 年 6 月 7 日，今天给一个学院本球友修改简历的时候，他提到实习的同事，都拿到了 20k 以上的 offer，甚至还有 25k 携程 offer 的，自己并不比他们差，问在实习、项目和能力上还能怎么提高？

我想说的是，这就是为什么很多人选择跑来卷互联网开发的原因啊，上线比其他行业高太多了，虽然互联网开发的工作强度也大，但最起码能劳有所获。

跳表是如何定义的呢？

跳表本质上是一个多层链表，底层是一个包含所有元素的有序链表，上一层作为索引层，包含了下一层的部分节点；层数通过随机算法确定，理论上可以无限高。

跳表节点包含分值 score、成员对象 obj、一个后退指针 backward，以及一个层级数组 level。每个层级包含 forward 前进指针和 span 跨度信息。

typedef struct skiplistNode {
    double score;                    // 分值（用于排序）
    robj *obj;                      // 数据对象
    struct skiplistNode *backward;   // 后退指针
    struct skiplistLevel {
        struct skiplistNode *forward; // 前进指针
        unsigned int span;           // 跨度（到下个节点的距离）
    } level[];                      // 层级数组
} skiplistNode;

跳表本身包含头尾节点指针、节点总数 length 和当前最大层数 level。

typedef struct skiplist {
    struct skiplistNode *header, *tail; // 头尾节点
    unsigned long length;               // 节点数量
    int level;                         // 最大层数
} skiplist;

span 跨度有什么用？

span 记录了当前节点到下一节点之间，底层到底跨越了几个节点，它的主要作用是快速找到 ZSet 中某个分值的排名。

比如说我们执行 ZRANK 命令时，如果没有 span，就需要从头节点开始遍历每个节点，直到找到目标分值，这样时间复杂度是 O(n)。

// 没有span的排名查询 - O(n)
int getRankWithoutSpan(skiplist *zsl, double score, robj *obj) {
    skiplistNode *x = zsl->header->level[0].forward;
    int rank = 0;
    
    while (x) {
        if (x->score == score && equalStringObjects(x->obj, obj)) {
            return rank + 1;  // 排名从1开始
        }
        rank++;
        x = x->level[0].forward;
    }
    return 0;
}

但有了 span，我们在从高层往低层搜索的时候，可以直接跳过一些节点，快速定位到目标分值所在的范围。这样就能把时间复杂度降到 O(log n)。

long skiplistGetRank(skiplist *zsl, double score, robj *obj) {
    skiplistNode *x = zsl->header;
    unsigned long rank = 0;
    
    // 从最高层开始查找
    for (int i = zsl->level - 1; i >= 0; i--) {
        while (x->level[i].forward &&
               (x->level[i].forward->score < score ||
                (x->level[i].forward->score == score &&
                 compareStringObjects(x->level[i].forward->obj, obj) < 0))) {
            
            rank += x->level[i].span;  // 累加跨度
            x = x->level[i].forward;
        }
        
        // 找到目标节点
        if (x->level[i].forward &&
            x->level[i].forward->score == score &&
            equalStringObjects(x->level[i].forward->obj, obj)) {
            rank += x->level[i].span;
            return rank;
        }
    }
    
    return 0;
}

为什么跳表的范围查询效率比字典高？

字典是通过哈希函数将键值对分散存储的，元素在内存中是无序分布的，没有任何顺序关系。而跳表本身就是有序的数据结构，所有元素按照分值从小到大排列。

当需要进行范围查询时，字典必须遍历所有元素，逐个检查每个元素是否在指定范围内，时间复杂度是 O(n)。比如要找分值在 60 到 80 之间的所有元素，字典只能把整个哈希表扫描一遍，因为它无法知道符合条件的元素在哪里。

而跳表的范围查询就高效多了。首先用 O(log n) 时间找到范围的起始位置，然后沿着底层的有序链表顺序遍历，直到超出范围为止。总时间复杂度是 O(log n + k)，其中 k 是结果集的大小。这种效率差异在数据量大的时候非常明显。

这也是为什么 Redis 的 zset 要用跳表而不是纯哈希表的重要原因，因为 zset 经常需要 ZRANGE、ZRANGEBYSCORE 这类范围操作。实际上 Redis 的 zset 是跳表和哈希表的组合：跳表保证有序性支持范围查询，哈希表保证 O(1) 的单点查找效率，两者互补。

1. Java 面试指南（付费）收录的小米暑期实习同学 E 一面面试原题：为什么 hash 表范围查询效率比跳表低
2. Java 面试指南（付费）收录的得物面经同学 8 一面面试原题：跳表的结构
3. Java 面试指南（付费）收录的美团面经同学 4 一面面试原题：Redis 跳表
4. Java 面试指南（付费）收录的阿里系面经同学 19 饿了么面试原题：跳表了解吗

memo：2025 年 6 月 8 日，今天有球友发信息称赞 Java 进阶之路的内容写得好，说实话，我是有这个自信的，基本上所写的内容也都是我这些年从读到的所有书籍、视频、教程中提炼到的精华，把一些难懂晦涩的知识都用通俗易懂的语言表达出来，配合手绘图，能让人更容易理解。

53.压缩列表了解吗？

答：压缩列表是 Redis 为了节省内存而设计的一种紧凑型数据结构，它会把所有数据连续存储在一块内存当中。

整个结构包含头部信息，如总的字节数、尾部偏移量、节点数量，以及连续的节点数据。

当 list、hash 和 set 的数据量较小且值都不大时，底层会使用压缩列表来实现。

通常情况在，每个节点包含三个部分：前一个节点的长度、编码类型和实际的数据。

前一个节点的长度是为了支持从后往前遍历；当前一个节点的长度小于 254 字节时，使用 1 字节存储；否则用 5 字节存储，第一个字节设置为 254，后四个字节存储实际长度。

编码类型会根据数据的实际情况选择最紧凑的存储方式。

但压缩列表有个致命问题，就是连锁更新。当插入或删除节点导致某个节点长度发生变化时，可能会影响后续所有节点存储的“前一个节点长度”字段，最坏情况下时间复杂度会退化到 O(n²)。

ziplist 的节点数量会超过 65535 吗？

不会。

Zllen 字段的类型是 uint16_t，最大值为 65535，也就是 2 的 16次方，所以压缩列表的节点数量不会超过 65535。

当节点数量小于 65535 时，该字段会存储实际的数量；否则该字段就固定为 65535，实际存储的数量需要逐个遍历节点来计算。

ziplist 的编码类型了解多少？

ziplist 的编码类型设计得很精巧，主要分为字符串编码和整数编码两大类，目的是用最少的字节存储数据。

比如 0 到 12 这些小整数直接编码在 type 字段中，只需要 1 个字节。

编码	长度	描述
11000000	1字节	int16_t类型整数，2 字节
11010000	1字节	int32_t类型整数，4 字节
11100000	1字节	int64_t类型整数，8 字节
11110000	1字节	24位有符号整数 ，3 字节
1111xxxx	1字节	数据范围在[0-12]，数据包含在编码中

对于字符串编码，根据字符串长度有三种格式。长度小于 63 字节的用 00 开头的单字节编码，剩余 6 位存储长度。长度在 63 到 16383 之间的用 01 开头的双字节编码，剩余 14 位存储长度。超过 16383 字节的用 10 开头，后面跟 4 字节存储长度。

编码	长度	描述
00pppppp	1字节	0-63 字节的字符串
01pppppp qqqqqqqq	2字节	64-16383字节的字符串
10______ qqqqqqqq rrrrrrrr ssssssss tttttttt	5字节	16384-4294967295字节的字符串

1. Java 面试指南（付费）收录的同学 30 腾讯音乐面试原题：什么情况下使用压缩列表

memo：2025 年 6 月 9 日修改至此，今天有球友特意发私信，感谢面渣逆袭对他的帮助。对，这么棒的内容，我依然选择了免费，因为我相信知识是有价值的，只有诚恳的分享出来才能让更多人受益。

54.quicklist 了解吗？

quicklist 是 Redis 在 3.2 版本时引入的，专门用于 List 的底层实现，它实际上是一个混合型数据结构，结合了压缩列表和双向链表的优点。

在早期的版本中，List 会根据元素的数量和大小采用两种不同的底层数据结构，当元素较少或者较小时，会使用压缩列表；否则用双向链表。

但这种设计有个问题，就是当 List 中的元素数量较多时，压缩列表会因为连锁更新导致性能下降，而双向链表又会占用更多内存。

quicklist 通过将 List 拆分为多个小的 ziplist，再通过指针链接成一个双向链表，巧妙的解决了这个问题。

默认情况下，每个 ziplist 可以存储 8KB 的数据，假如每个元素的大小恰好是 1KB，那么一个 quicklist 就可以存储 8 个元素。80 个这样的元素就会被分成 10 个 ziplist。

这样既保留了压缩列表的内存紧凑性，又减少了双向链表指针的数量，进一步降低了内存开销。

除此之外，quicklist 还有一个重要的特性，就是它的可配置性，可以通过填充因子控制每个 ziplist 节点的大小。当填充因子为正数时，它还可以限制每个 ziplist 最多包含的元素数量。

# 填充因子，默认 -2（8KB）
list-max-ziplist-size 10

如果想进一步节省内存，quicklist 还支持对中间节点进行 LZF 压缩，压缩深度为 1 时，表示除了首尾各 1 个节点不压缩外，其他节点都压缩。

# 压缩深度，默认 0（不压缩）
list-compress-depth 1

LZF 压缩算法了解吗？

LZF 是一种快速的无损压缩算法，主要用于减少数据存储空间。它的核心思想是通过查找重复数据来实现压缩，通过一个滑动窗口来查找重复的字节序列，并将这些序列替换为更短的引用。

输入数据: "hello world hello redis"

步骤1: 处理 "hello world "
- 建立字典，记录字节序列位置

步骤2: 遇到重复的 "hello"
- 在字典中找到之前的 "hello" 位置
- 用 (距离, 长度) 对替换: (12, 5)

输出: "hello world " + (12,5) + " redis"

memo：2025 年 6 月 10 日，今天有球友发信息说找我修改了简历后，又按照星球的学习资料好好学了一下之后，拿到了字节跳动的 offer，并特意发了一个大红包来感谢。这种被认可被需要的感觉，真好！

补充

55.假如 Redis 里面有 1 亿个 key，其中有 10w 个 key 是以某个固定的已知的前缀开头的，如何将它们全部找出来？

我会使用 SCAN 命令配合 MATCH 参数来解决。

比如要找以 user: 开头的 key，可以执行 SCAN 0 MATCH user:* COUNT 1000。

SCAN 的优势在于它是基于游标的增量迭代，每次只返回一小批结果，不会阻塞服务器。可以从游标 0 开始，每次处理返回的 key 列表，然后用返回的下一个游标继续扫描，直到游标回到 0 表示扫描完成。

使用 Spring Data Redis 的代码示例：

@Service
public class RedisKeyService {
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    public List<String> scanKeysByPrefix(String prefix, int batchSize) {
        List<String> keys = new ArrayList<>();
        
        ScanOptions options = ScanOptions.scanOptions()
                .match(prefix + "*")
                .count(batchSize)
                .build();
        
        try (Cursor<String> cursor = redisTemplate.scan(options)) {
            while (cursor.hasNext()) {
                keys.add(cursor.next());
            }
        }
        
        return keys;
    }
}

千万不要用 KEYS 命令，因为 KEYS 会阻塞 Redis 服务器直到遍历完所有 key，在生产环境中对 1 亿个 key 执行 KEYS 是非常危险的。

memo：2025 年 6 月 11 日修改至此，今天有读者留言说，找实习的时候背了一个月的面渣逆袭，然后快手和美团都拿到 offer 了。能帮助到大家，也是我做技术博主最开心的一件事情了，也感谢读者给的口碑。

56.Redis在秒杀场景下可以扮演什么角色？

秒杀是一种非常特殊的业务场景，它的特点是在极短时间内会有大量用户涌入系统，对系统的并发处理能力、响应速度和数据一致性都提出了极高的要求。在这种场景下，Redis 作为一种高性能的内存数据库，能够发挥多方面的关键作用。

比如说在秒杀开始前，我们可以将商品信息、库存数据等预先加载到 Redis 中，这样大量的用户读请求就可以直接从 Redis 中获取响应，而不必每次都去访问数据库，这样就能大大减轻数据库的访问压力。

其次，Redis 在库存控制方面具有得天独厚的优势。秒杀最核心的问题之一就是容易发生超卖。Redis 提供的原子操作如 DECR、DECRBY 等命令，可以确保在高并发环境下库存计数的准确性。

更复杂的逻辑，可以通过 Lua 脚本来实现，因为 Lua 脚本在 Redis 中是原子执行的，所以可以包含复杂的判断和操作逻辑，比如先检查库存是否充足，再进行扣减，这整个过程是不会被其他操作打断的。

第三点，Redis 的分布式锁可以确保多个用户同时抢购同一件商品时的操作是互斥的，保证数据一致性的同时，还可以用来防止用户重复下单。

第四点，限流削峰。秒杀开始的瞬间，可能会有成千上万的请求同时到达，如果不加控制，很容易导致系统崩溃。Redis 可以实现多种限流算法，比如简单的计数器限流、令牌桶或漏桶算法等。

通过限流算法我们可以控制单位时间内系统能够处理的请求数量，超出部分可以排队或者直接拒绝，从而保护系统的稳定运行。

memo：2025 年 6 月 12 日修改至此，今天有球友发信息说，大二就拿下了美团的实习 offer，特意发来感谢，说我的付出对他有着巨大的帮助，真的很感动，每一个懂得感恩的球友，你们也是我坚持下去的最强动力。

Redis具体如何实现削峰呢？

削峰的本质是将瞬时的高流量请求缓冲起来，通过排队、限流等机制，使系统以一个可承受的速度来处理请求。

那第一步就是缓存预热。在秒杀活动开始前，先把商品信息这些热点数据提前加载到 Redis 中。这样用户访问商品页面时，可以直接从 Redis 读取，数据库基本上不会有压力。

第二步是引入消息队列，特别是下单这种写操作，不能让用户等太久，但后端处理订单、扣库存这些操作又比较重。所以可以用 Redis 的 List 做了个队列，或者直接用 RocketMQ 这种标准的消息中间件，用户下单后立即返回"订单提交成功"，然后把订单数据丢到队列里，后台服务慢慢消费。这样既保证了用户体验，又避免了系统被瞬时写请求压垮。

第三步，可以在秒杀活动中加入答题环节，只有答对题目的用户才能参与秒杀活动，这样可以最大程度减少无效请求。

一个比较完整的秒杀削峰处理方案：

@Service
public class SeckillServiceImpl implements SeckillService {
    
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    
    @Autowired
    private OrderService orderService;
    
    @Autowired
    private CommodityService commodityService;
    
    /**
     * 秒杀请求入口
     */
    public Result seckill(Long userId, Long commodityId) {
        // 1. 用户请求频率限制
        if (!countRateLimit("user:" + userId, 5, 60)) {
            return Result.error("请求过于频繁");
        }
        
        // 2. 商品是否在秒杀时间内
        if (!isInSeckillTime(commodityId)) {
            return Result.error("秒杀未开始或已结束");
        }
        
        // 3. 是否还有库存(快速失败)
        String stockKey = "seckill:stock:" + commodityId;
        Integer stock = Integer.valueOf(redisTemplate.opsForValue().get(stockKey));
        if (stock != null && stock <= 0) {
            return Result.error("商品已售罄");
        }
        
        // 4. 全局限流
        if (!acquireToken("global", 1000, 100)) {
            // 系统负载过高，将请求放入队列延迟处理
            enqueueDelayedRequest(userId, commodityId);
            return Result.success("秒杀请求已受理，排队处理中");
        }
        
        // 5. 检查用户是否已购买
        if (hasUserBought(userId, commodityId)) {
            return Result.error("您已经购买过该商品");
        }
        
        // 6. 将请求放入队列，返回排队状态
        String requestId = generateRequestId(userId, commodityId);
        enqueueRequest(userId, commodityId, requestId);
        
        return Result.success("秒杀请求已提交，请等待结果", requestId);
    }
    
    /**
     * 异步处理秒杀请求
     */
    @Scheduled(fixedRate = 50) // 每50ms处理一批
    public void processSeckillQueue() {
        String queueKey = "seckill:queue";
        
        // 批量处理，控制处理速度
        for (int i = 0; i < 10; i++) {
            String requestJson = redisTemplate.opsForList().leftPop(queueKey);
            if (requestJson == null) {
                break;
            }
            
            SeckillRequest request = JSON.parseObject(requestJson, SeckillRequest.class);
            try {
                // 执行秒杀核心逻辑
                boolean success = doSeckill(request.getUserId(), request.getCommodityId());
                
                // 更新请求状态，便于用户查询
                String statusKey = "seckill:status:" + request.getRequestId();
                redisTemplate.opsForValue().set(statusKey, success ? "SUCCESS" : "FAILED", 1, TimeUnit.HOURS);
                
            } catch (Exception e) {
                log.error("处理秒杀请求失败", e);
                // 记录失败状态
                String statusKey = "seckill:status:" + request.getRequestId();
                redisTemplate.opsForValue().set(statusKey, "ERROR", 1, TimeUnit.HOURS);
            }
        }
    }
    
    /**
     * 秒杀核心逻辑
     */
    private boolean doSeckill(Long userId, Long commodityId) {
        // 使用Lua脚本保证原子性操作
        String script = 
            "-- 检查库存\n" +
            "local stockKey = KEYS[1]\n" +
            "local stock = tonumber(redis.call('get', stockKey))\n" +
            "if stock == nil or stock <= 0 then\n" +
            "    return 0\n" +
            "end\n" +
            "\n" +
            "-- 检查是否重复购买\n" +
            "local boughtKey = KEYS[2]\n" +
            "local hasBought = redis.call('sismember', boughtKey, ARGV[1])\n" +
            "if hasBought == 1 then\n" +
            "    return -1\n" +
            "end\n" +
            "\n" +
            "-- 扣减库存并记录购买\n" +
            "redis.call('decr', stockKey)\n" +
            "redis.call('sadd', boughtKey, ARGV[1])\n" +
            "\n" +
            "-- 返回成功\n" +
            "return 1";
        
        String stockKey = "seckill:stock:" + commodityId;
        String boughtKey = "seckill:bought:" + commodityId;
        
        Long result = (Long) redisTemplate.execute(
            new DefaultRedisScript<>(script, Long.class),
            Arrays.asList(stockKey, boughtKey),
            userId.toString()
        );
        
        if (result == 1) {
            // 创建订单(可以进一步异步化)
            createOrder(userId, commodityId);
            return true;
        }
        
        return false;
    }
    
    // 其他辅助方法...
}

Redis如何做限流呢？

限流是为了控制系统的请求速率，防止系统被过多的请求压垮。

Redis 实现限流最简单的方法是基于计数器的固定窗口限流。比如限制用户每分钟最多访问 100 次，我们就用 INCR 命令给每个用户设个计数器，key 是 rate_limit:用户ID:分钟时间戳，每次请求就加 1，同时设置 60 秒过期。如果计数超过 100 就拒绝请求。

// 伪代码
String key = "rate_limit:" + userId;
// 尝试获取当前计数
Long count = redis.get(key);
// 如果key不存在，设置为1并设置过期时间
if (count == null) {
    redis.setex(key, 60, "1"); // 60秒窗口期
    return true; // 允许访问
}
// 如果计数未超过限制
if (count < maxRequests) {
    redis.incr(key);
    return true; // 允许访问
}
return false; // 拒绝访问

这种方法简单粗暴，但有个问题就是临界时间会有突刺，比如用户在第 59 秒访问了 100 次，第 61 秒又访问 100 次，相当于 2 秒内访问了 200 次。

第二种就是滑动窗口限流，通过 Redis 的 ZSET 来实现，把每次请求的时间戳作为 score 存进去，然后用 ZREMRANGEBYSCORE 删除窗口外的旧数据，再用 ZCARD 统计当前窗口内的请求数。这样限流就比较均匀了。

// 伪代码
String key = "sliding_window:" + userId;
long now = System.currentTimeMillis();
// 添加当前请求到有序集合，score为当前时间戳
redis.zadd(key, now, String.valueOf(now));
// 移除时间窗口之前的请求数据
redis.zremrangeByScore(key, 0, now - windowSize);
// 设置key过期时间，避免冷用户持续占用内存
redis.expire(key, windowSize / 1000 + 1);
// 获取当前窗口的请求数
Long count = redis.zcard(key);
return count <= maxRequests;

在实际开发中，通常会采用令牌桶算法，它就像在帝都/魔都买车，摇到号才有资格，没摇到就只能等下一次（😁）。

可以在 Redis 里存两个值，一个是令牌数量，一个是上次更新时间。每次请求时用 Lua 脚本计算应该补充多少令牌，然后判断是否有足够的令牌。

-- Redis Lua脚本实现令牌桶算法
local key = KEYS[1]  -- 限流的key
local max_permits = tonumber(ARGV[1])  -- 最大令牌数
local permits_per_second = tonumber(ARGV[2])  -- 每秒产生的令牌数
local required_permits = tonumber(ARGV[3])  -- 请求的令牌数

-- 获取当前时间
local time = redis.call('time')
local now_micros = tonumber(time[1]) * 1000000 + tonumber(time[2])

-- 获取上次更新的时间和当前存储的令牌数
local last_micros = tonumber(redis.call('hget', key, 'last_micros') or 0)
local stored_permits = tonumber(redis.call('hget', key, 'stored_permits') or 0)

-- 计算时间间隔内新产生的令牌数
local interval_micros = now_micros - last_micros
local new_permits = interval_micros * permits_per_second / 1000000
stored_permits = math.min(max_permits, stored_permits + new_permits)

-- 判断令牌是否足够
local result = 0
if stored_permits >= required_permits then
    -- 令牌足够，更新令牌数和时间
    stored_permits = stored_permits - required_permits
    result = 1
end

-- 更新Redis中的数据
redis.call('hset', key, 'last_micros', now_micros)
redis.call('hset', key, 'stored_permits', stored_permits)
redis.call('expire', key, 10)  -- 设置过期时间，避免长期占用内存

return result

1. Java 面试指南（付费）收录的农业银行面经同学 3 Java 后端面试原题：秒杀问题（错峰、削峰、前端、流量控制）
2. Java 面试指南（付费）收录的滴滴面经同学 3 网约车后端开发一面原题：限流算法

memo：2025 年 6 月 13 日修改至此，今天在修改简历的过程中，碰到一位西安交通大学的球友，他整个校园经历是比较丰富的，先去了新加坡国立大学做暑期交换生，然后又去了加州大学伯克利分校做学期交换生，希望大家也都能在学校阶段尽量丰富自己的经历，争取多拿一些奖学金、实习经历，这样才能在毕业时有更强的竞争力。

57.客户端宕机后 Redis 服务端如何感知到？

TCP 的 keepalive 是 Redis 用来检测客户端连接状态的主要机制，默认值为 300 秒。

# 针对低延迟场景，设置为60秒，表示每60秒发送一次keepalive探测
config set tcp-keepalive 60

当客户端与服务器在指定时间内没有任何数据交互时，Redis 服务器会发送 TCP ACK 探测包，如果连续多次没有收到响应，TCP 协议栈会通知 Redis 服务端连接已断开，之后，Redis 服务端会清理相关的连接资源，释放连接。

另外还有一个 timeout 参数，用来控制客户端连接的空闲超时时间。

# 表示600秒内没有任何命令则断开连接
config set timeout 600

默认值为 0，表示永不断开连接；当设置为非零值时，如果客户端在指定时间内没有发送任何命令，服务端会主动断开连接。

Redis 服务器会定期检查空闲连接是否超时，检查频率由 hz 参数控制；这将有助于释放那些客户端异常退出但 TCP 连接未正常关闭的资源。

不同的连接池也会有自己的连接检测机制，比如 Jedis 连接池可以通过设置 testOnBorrow 和 testWhileIdle 来启用连接检测。

# 是否启用连接池
spring.redis.jedis.pool.enabled=true
# 连接池最大连接数（使用负值表示没有限制）
spring.redis.jedis.pool.max-active=200
# 连接池最大空闲连接数
spring.redis.jedis.pool.max-idle=200
# 连接池最小空闲连接数
spring.redis.jedis.pool.min-idle=50
# 连接池最大阻塞等待时间（使用负值表示没有限制）
spring.redis.jedis.pool.max-wait=3000
# 空闲连接检查间隔（毫秒）
spring.redis.jedis.pool.time-between-eviction-runs=60000

1. Java 面试指南（付费）收录的字节跳动面经同学 21 抖音商城一面面试原题：如果客户端宕机服务器如何感知？

memo：2025 年 6 月 14 日修改至此，今天有球友在 VIP 群里发消息说，快手oc 了，恭喜他。球友们都太有实力了。

---

整整一个半月，面渣逆袭 Redis 篇第二版终于整理完了，这一版几乎可以说是重写了，每天耗费了大量的精力在上面，可以说是改头换面，有一种士别俩月，当刮目相看的感觉（从 1.8 万字暴涨到 4.6 万字，加餐的同时区分高频低频版）。

网上的八股其实不少，有些还是付费的，我觉得是一件好事，可以给大家提供更多的选择，但面渣逆袭的含金量懂的都懂。

面渣逆袭第二版是在星球嘉宾三分恶的初版基础上，加入了二哥自己的思考，加入了 1000 多份真实面经之后的结果，并且从 24 届到 25 届，再到 26 届，帮助了很多小伙伴。未来的 27、28 届，也将因此受益，从而拿到心仪的 offer。

能帮助到大家，我很欣慰，并且在重制面渣逆袭的过程中，我也成长了很多，很多薄弱的基础环节都得到了加强，因此第二版的面渣逆袭不只是给大家的礼物，也是我在技术上蜕变的记录。

很多时候，我觉得自己是一个佛系的人，不愿意和别人争个高低，也不愿意去刻意宣传自己的作品。

我喜欢静待花开。

如果你觉得面渣逆袭还不错，可以告诉学弟学妹们有这样一份免费的学习资料，帮我做个口碑。

我还会继续优化，也不确定第三版什么时候会来，但我会尽力。

愿大家都有一个光明的未来。

由于 PDF 没办法自我更新，所以需要最新版的小伙伴，可以微信搜【沉默王二】，或者扫描/长按识别下面的二维码，关注二哥的公众号，回复【222】即可拉取最新版本。

当然了，请允许我的一点点私心，那就是星球的 PDF 版本会比公众号早一个月时间，毕竟星球用户都付费过了，我有必要让他们先享受到一点点福利。相信大家也都能理解，毕竟在线版是免费的，CDN、服务器、域名、OSS 等等都是需要成本的。

更别说我付出的时间和精力了，大家觉得有帮助还请给个口碑，让你身边的同事、同学都能受益到。

我把二哥的 Java 进阶之路、JVM 进阶之路、并发编程进阶之路，以及所有面渣逆袭的版本都放进来了，涵盖 Java基础、Java集合、Java并发、JVM、Spring、MyBatis、计算机网络、操作系统、MySQL、Redis、RocketMQ、分布式、微服务、设计模式、Linux 等 16 个大的主题，共有 40 多万字，2000+张手绘图，可以说是诚意满满。

这次仍然是三个版本，亮白、暗黑和 epub 版本。给大家展示其中一个 epub 版本吧，有些小伙伴很急需这个版本，所以也满足大家了。

图文详解 57 道 Redis 面试高频题，这次吊打面试官，我觉得稳了（手动 dog）。整理：沉默王二，戳转载链接，作者：三分恶，戳原文链接。

没有什么使我停留——除了目的，纵然岸旁有玫瑰、有绿荫、有宁静的港湾，我是不系之舟。

系列内容：

• 面渣逆袭 Java SE 篇 👍
• 面渣逆袭 Java 集合框架篇 👍
• 面渣逆袭 Java 并发编程篇 👍
• 面渣逆袭 JVM 篇 👍
• 面渣逆袭 Spring 篇 👍
• 面渣逆袭 Redis 篇 👍
• 面渣逆袭 MyBatis 篇 👍
• 面渣逆袭 MySQL 篇 👍
• 面渣逆袭操作系统篇 👍
• 面渣逆袭计算机网络篇 👍
• 面渣逆袭 RocketMQ 篇 👍
• 面渣逆袭分布式篇 👍
• 面渣逆袭微服务篇 👍
• 面渣逆袭设计模式篇 👍
• 面渣逆袭 Linux 篇 👍