guava cache 源码分析

cache.get(key){
	value = loadFromCache(key);
	if(null != value){
		return value;
	}
	value = loader.load(key);
}
loader.load(key){
	// 此处是竞争点，得有线程安全保护，线程发现有其它线程在查询key时，放弃执行，直接等待结果
	value = loadFromDB(key);	
	storeToCache(value);
	
}
// 增强版本
loader.load(key){
	// map<key,future> futureMap 
	Future future = futureMap.get(key);
	if(null != future){
		value = loadFromDB(key);	
		storeToCache(value);
	}{
		future.get();
	}
}

static class Segment<K, V> extends ReentrantLock {
    volatile @MonotonicNonNull AtomicReferenceArray<ReferenceEntry<K, V>> table;
    V get(K key, int hash, CacheLoader<? super K, V> loader) throws ExecutionException {
        ReferenceEntry<K, V> e = getEntry(key, hash);
        // fast path 
        if (e != null) {
            V value = getLiveValue(e, now);
            if (value != null) {
                return xx                       // 找到值则直接返回
            }
            ValueReference<K, V> valueReference = e.getValueReference();
            if (valueReference.isLoading()) {   // 如果正在加载则进入等待
                return waitForLoadingValue(e, key, valueReference);
            }
        }
        // slow path 进入load 逻辑
        return lockedGetOrLoad(key, hash, loader);  //  ==> loadingValueReference.loadFuture
    }  
    V waitForLoadingValue(ReferenceEntry<K, V> e, K key, ValueReference<K, V> valueReference)
        throws ExecutionException {
        ...
        V value = valueReference.waitForValue(); // LoadingValueReference.waitForValue ==> future.get
        ...
    }
}

V lockedGetOrLoad(K key, int hash, CacheLoader<? super K, V> loader) throws ExecutionException {
    ReferenceEntry<K, V> e;
    ValueReference<K, V> valueReference = null;
    LoadingValueReference<K, V> loadingValueReference = null;
    boolean createNewEntry = true;
    lock(); // 加锁

    int newCount = this.count - 1;
    AtomicReferenceArray<ReferenceEntry<K, V>> table = this.table;
    // 计算key在数组中的落点
    int index = hash & (table.length() - 1);
    ReferenceEntry<K, V> first = table.get(index);
    // 沿着某个index 链表依次遍历
    for (e = first; e != null; e = e.getNext()) {
        K entryKey = e.getKey();
        if (e.getHash() == hash
            && entryKey != null
            && map.keyEquivalence.equivalent(key, entryKey)) {
            valueReference = e.getValueReference();
            V value = valueReference.get();
            if (value == null || map.isExpired(e, now){
                enqueueNotification(...);
            } else {
                return value;
            }
            this.count = newCount; // write-volatile
            break;
        }
    }
    loadingValueReference = new LoadingValueReference<>();
    if (e == null) {
        e = newEntry(key, hash, first);
        e.setValueReference(loadingValueReference);
        table.set(index, e);
    } else {
        e.setValueReference(loadingValueReference);
    }

    unlock(); // 解锁
    synchronized (e) {
        return loadSync(key, hash,loadingValueReference, loader);
    }
}

V loadSync(K key,int hash,
    LoadingValueReference<K, V>,loadingValueReference,CacheLoader<? super K, V> loader)throws ExecutionException {
    ListenableFuture<V> loadingFuture = loadingValueReference.loadFuture(key, loader);
    return getAndRecordStats(key, hash,loadingValueReference, loadingFuture);
}
public ListenableFuture<V> loadFuture(K key, CacheLoader<? super K, V> loader) {
    ...
    V newValue = loader.load(key);
    return set(newValue) ? futureValue : Futures.immediateFuture(newValue);
    ...
}

V waitForLoadingValue(ReferenceEntry<K, V> e, K key, ValueReference<K, V> valueReference)
    throws ExecutionException {
    checkState(!Thread.holdsLock(e), "Recursive load of: %s", key);
    V value = valueReference.waitForValue();
    if (value == null) {
        throw new InvalidCacheLoadException("CacheLoader returned null for key " + key + ".");
    }
    ...
    return value;
}

static class LoadingValueReference<K, V> implements ValueReference<K, V> {
    volatile ValueReference<K, V> oldValue;
    final SettableFuture<V> futureValue =SettableFuture.create();
    final Stopwatch stopwatch =Stopwatch.createUnstarted();

    public boolean set(@Nullable V newValue) {
        return futureValue.set(newValue);
    }
    public V get() {
        return oldValue.get();
    }
    public V waitForValue() throws ExecutionException {
        // 对future.get的封装
        return getUninterruptibly(futureValue);
    }
    public boolean setException(Throwable t) {
        return futureValue.setException(t);
    }
    public void notifyNewValue(@Nullable V newValue) {
        if (newValue != null) {
            // future.get ==> waitForValue即可立即返回
            set(newValue);
        } else {
            oldValue = unset();
        }
    }
}

LoadingCache<String, Object> caches = CacheBuilder.newBuilder() 
    .maximumSize(100) 
    .refreshAfterWrite(10, TimeUnit.MINUTES) 
    .build(new CacheLoader<String, Object>() { 
        @Override 
        public Object load(String key) throws Exception { 
            return generateValueByKey(key); 
        } 
    }); 

ListeningExecutorService backgroundRefreshPools = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(20)); LoadingCache<String, Object> caches = CacheBuilder.newBuilder() 
    .maximumSize(100) 
    .refreshAfterWrite(10, TimeUnit.MINUTES) 
    .build(new CacheLoader<String, Object>() { 
        @Override 
        public Object load(String key) throws Exception { 
            return generateValueByKey(key); 
        } 
        @Override 
        public ListenableFuture<Object> reload(String key, Object oldValue) throws Exception { 
            return backgroundRefreshPools.submit(new Callable<Object>() { 
                @Override 
                public Object call() throws Exception { 
                    return generateValueByKey(key); 
                } 
            }); 
        } 
    }); 

Product productA = new Product(...);
WeakReference<Product> weakProductA = new WeakReference<>(productA);