java如何实现高并发场景下三级缓存的数据一致性_F11

下面代码是一个使用Java和Redisson实现的三级缓存服务，主要功能包括：

1.缓存结构：

本地缓存：使用Caffeine实现，最大容量10,000，写入后10分钟过期
分布式缓存：使用Redisson的RMap结构操作Redis
数据库：作为最终数据源

2.数据读取流程：

先查本地缓存，命中则返回
未命中则查Redis，命中则更新本地缓存并返回
仍未命中则获取锁，再次检查两级缓存（双重检查）
最后从数据库读取，更新两级缓存后返回

3.数据更新流程：

使用分布式锁保证写操作原子性
先更新数据库
删除本地缓存和Redis缓存
通过Redis Pub/Sub发布缓存清除消息给集群内其他节点
执行延迟双删（100毫秒后再次删除Redis缓存）

4.并发控制：

读取时使用本地锁（ReentrantLock）防止缓存击穿
更新时使用Redisson分布式锁（RLock）保证跨节点原子性
锁使用完成后从ConcurrentHashMap中移除

5.集群同步：

使用Redis的RTopic实现消息发布订阅
接收到清除消息时自动删除本地缓存
确保集群内各节点缓存一致性

该实现综合运用了延迟双删、发布订阅、锁机制和TTL等多种策略，保障了高并发场景下三级缓存的数据一致性，尤其适合分布式微服务架构。

实战代码：

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

import java.util.concurrent.*;

import java.util.concurrent.locks.ReentrantLock;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Service;

import org.redisson.api.*;

import com.github.benmanes.caffeine.cache.Cache;

import com.github.benmanes.caffeine.cache.CacheBuilder;

@Service

public class CacheService {

// 本地一级缓存（Caffeine）

private final Cache<String, Object> localCache;

// Redisson客户端，用于分布式操作

private final RedissonClient redissonClient;

// 锁缓存，用于控制并发

private final ConcurrentHashMap<String, ReentrantLock> lockMap = new ConcurrentHashMap<>();

// 延迟任务执行器

private final ScheduledExecutorService scheduledExecutorService;

// 主题订阅，用于接收集群消息

private final RTopic cacheClearTopic;

@Autowired

public CacheService(RedissonClient redissonClient) {

this.redissonClient = redissonClient;

this.localCache = CacheBuilder.newBuilder()

.maximumSize(10_000)

.expireAfterWrite(10, TimeUnit.MINUTES)

.build();

this.scheduledExecutorService = Executors.newScheduledThreadPool(5);

this.cacheClearTopic = redissonClient.getTopic("cache:clear");

// 注册消息监听器

cacheClearTopic.addListener(String.class, (channel, key) -> {

localCache.invalidate(key);

});

}

// 读取缓存

public Object get(String key) {

// 1. 先查本地缓存

Object value = localCache.getIfPresent(key);

if (value != null) {

return value;

}

// 2. 本地缓存未命中，查Redis

RMap<String, Object> redisMap = redissonClient.getMap("cache");

value = redisMap.get(key);

if (value != null) {

localCache.put(key, value);

return value;

}

// 3. Redis未命中，查数据库

ReentrantLock lock = lockMap.computeIfAbsent(key, k -> new ReentrantLock());

lock.lock();

try {

// 双重检查

value = localCache.getIfPresent(key);

if (value != null) {

return value;

}

value = redisMap.get(key);

if (value != null) {

localCache.put(key, value);

return value;

}

// 从数据库读取

value = readFromDatabase(key);

if (value != null) {

// 放入Redis并设置TTL

redisMap.put(key, value, 300, TimeUnit.SECONDS);

// 放入本地缓存

localCache.put(key, value);

}

return value;

} finally {

lock.unlock();

lockMap.remove(key);

}

// 更新数据

public void update(String key, Object value) {

// 使用分布式锁保证写操作的原子性

RLock lock = redissonClient.getLock("writeLock:" + key);

lock.lock();

try {

// 1. 更新数据库

boolean success = updateDatabase(key, value);

if (success) {

// 2. 先删除本地缓存

localCache.invalidate(key);

// 3. 删除Redis缓存

RMap<String, Object> redisMap = redissonClient.getMap("cache");

redisMap.remove(key);

// 4. 发布清除缓存的消息到集群

cacheClearTopic.publish(key);

// 5. 延迟双删

scheduledExecutorService.schedule(() -> {

redisMap.remove(key);

}, 100, TimeUnit.MILLISECONDS);

}

} finally {

lock.unlock();

}

// 从数据库读取数据（示例方法）

private Object readFromDatabase(String key) {

// 实际实现中会查询数据库

return "data_from_db_" + key;

}

// 更新数据库（示例方法）

private boolean updateDatabase(String key, Object value) {

// 实际实现中会更新数据库

return true;

}

redisson配置

import org.redisson.Redisson;

import org.redisson.api.RedissonClient;

import org.redisson.config.Config;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

@Configuration

public class RedissonConfig {

@Bean

public RedissonClient redissonClient() {

Config config = new Config();

// 单机模式配置

config.useSingleServer()

.setAddress("redis://localhost:6379")

.setConnectionMinimumIdleSize(5)

.setConnectionPoolSize(50);

// 集群模式配置示例

config.useClusterServers()

.addNodeAddress("redis://node1:6379", "redis://node2:6379")

.setScanInterval(2000)

.setMasterConnectionMinimumIdleSize(10)

.setMasterConnectionPoolSize(64)

.setSlaveConnectionMinimumIdleSize(10)

.setSlaveConnectionPoolSize(64);

return Redisson.create(config);

}