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进行处理
   	// 设置缓存
    }
}