高并发技巧之Redis和本地缓存使用技巧分享

缓存击穿

缓存穿透

缓存雪崩

LoadingCache<String, String> cache = CacheBuilder.newBuilder()
.maximumSize(1000L)
.expireAfterAccess(Duration.ofHours(1L)) // 多久不访问就过期
.expireAfterWrite(Duration.ofHours(1L)) // 多久这个key没修改就过期
.build(new CacheLoader<String, String>() {
@Override
public String load(String key) throws Exception {
// 数据装载方式，一般就是loadDB
return key + ” world”;
}
});
String value = cache.get(“hello”); // 返回hello world

<dependency>
<groupId>com.github.phantomthief</groupId>
<artifactId>zknotify-cache</artifactId>
<version>0.1.22</version>
</dependency>

public interface ReloadableCache<T> extends Supplier<T> {

/**
* 获取缓存数据
*/
@Override
T get();

/**
* 通知全局缓存更新
* 注意：如果本地缓存没有初始化，本方法并不会初始化本地缓存并重新加载
*
* 如果需要初始化本地缓存，请先调用 {@link ReloadableCache#get()}
*/
void reload();

/**
* 更新本地缓存的本地副本
* 注意：如果本地缓存没有初始化，本方法并不会初始化并刷新本地的缓存
*
* 如果需要初始化本地缓存，请先调用 {@link ReloadableCache#get()}
*/
void reloadLocal();
}

import com.google.common.collect.Lists;
import org.apache.commons.collections4.CollectionUtils;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;

/**
* @Description 分布式加载缓存的rpc服务，如果部署了多台机器那么调用端最好使用id做一致性hash保证相同id的请求打到同一台机器。
**/
public abstract class AbstractCacheSetterService implements CacheSetterService {

private final ConcurrentMap<String, CountDownLatch> loadCache = new ConcurrentHashMap<>();

private final Object lock = new Object();

@Override
public void load(Collection<String> needLoadIds) {
if (CollectionUtils.isEmpty(needLoadIds)) {
return;
}
CountDownLatch latch;
Collection<CountDownLatch> loadingLatchList;
synchronized (lock) {
loadingLatchList = excludeLoadingIds(needLoadIds);

needLoadIds = Collections.unmodifiableCollection(needLoadIds);

latch = saveLatch(needLoadIds);
}
System.out.println(“needLoadIds:” + needLoadIds);
try {
if (CollectionUtils.isNotEmpty(needLoadIds)) {
loadCache(needLoadIds);
}
} finally {
release(needLoadIds, latch);
block(loadingLatchList);
}

}

/**
* 加锁
* @param loadingLatchList 需要加锁的id对应的CountDownLatch
*/
protected void block(Collection<CountDownLatch> loadingLatchList) {
if (CollectionUtils.isEmpty(loadingLatchList)) {
return;
}
System.out.println(“block:” + loadingLatchList);
loadingLatchList.forEach(l -> {
try {
l.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}

/**
* 释放锁
* @param needLoadIds 需要释放锁的id集合
* @param latch 通过该CountDownLatch来释放锁
*/
private void release(Collection<String> needLoadIds, CountDownLatch latch) {
if (CollectionUtils.isEmpty(needLoadIds)) {
return;
}
synchronized (lock) {
needLoadIds.forEach(id -> loadCache.remove(id));
}
if (latch != null) {
latch.countDown();
}
}

/**
* 加载缓存，比如根据id从db查询数据，然后设置到redis中
* @param needLoadIds 加载缓存的id集合
*/
protected abstract void loadCache(Collection<String> needLoadIds);

/**
* 对需要加载缓存的id绑定CountDownLatch，后续相同的id请求来了从map中找到CountDownLatch，并且await，直到该线程加载完了缓存
* @param needLoadIds 能够正在去加载缓存的id集合
* @return 公用的CountDownLatch
*/
protected CountDownLatch saveLatch(Collection<String> needLoadIds) {
if (CollectionUtils.isEmpty(needLoadIds)) {
return null;
}
CountDownLatch latch = new CountDownLatch(1);
needLoadIds.forEach(loadId -> loadCache.put(loadId, latch));
System.out.println(“loadCache:” + loadCache);
return latch;
}

/**
* 哪些id正在加载数据，此时持有相同id的线程需要等待
* @param ids 需要加载缓存的id集合
* @return 正在加载的id所对应的CountDownLatch集合
*/
private Collection<CountDownLatch> excludeLoadingIds(Collection<String> ids) {
List<CountDownLatch> loadingLatchList = Lists.newArrayList();
Iterator<String> iterator = ids.iterator();
while (iterator.hasNext()) {
String id = iterator.next();
CountDownLatch latch = loadCache.get(id);
if (latch != null) {
loadingLatchList.add(latch);
iterator.remove();
}
}
System.out.println(“loadingLatchList:” + loadingLatchList);
return loadingLatchList;
}
}

import java.util.Collection;
public class BizCacheSetterRpcService extends AbstractCacheSetterService {
@Override
protected void loadCache(Collection<String> needLoadIds) {
// 读取db进行处理
// 设置缓存
}
}