高並行技巧之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進行處理
   	// 設定快取
    }
}