BlockingCache是对Ehcache进行的扩展,BlockingCache内置了读写锁，不需要用户显示调用。
要彻底分析BlockingCache的原理，需要首先来看一下它内部用到的一些类。
//锁的管理器接口
public interface CacheLockProvider {
/**
* 根据key获取锁
* 这个实现需要保证给定相同的key必须返回同一把锁
*/
Sync getSyncForKey(Object key);
}
public interface StripedReadWriteLock extends CacheLockProvider {
/**
* 根据key获得Java的ReadWriteLock
*/
ReadWriteLock getLockForKey(Object key);
/**
* 获取所有锁
*/
List<ReadWriteLockSync> getAllSyncs();
}
上面的俩个接口是用来管理锁的，待会我们再看具体实现，再来看一下锁的接口。
public interface Sync {
/**
* Acquire lock of LockType.READ or WRITE
* @param type the lock type to acquire
*/
void lock(LockType type);
/**
* Tries to acquire a LockType.READ or WRITE for a certain period
* @param type the lock type to acquire
* @param msec timeout
* @return true if the lock got acquired, false otherwise
* @throws InterruptedException Should the thread be interrupted
*/
boolean tryLock(LockType type, long msec) throws InterruptedException;
/**
* Releases the lock held by the current Thread.
* In case of a LockType.WRITE, should the lock not be held by the current Thread, nothing happens
* @param type the lock type to acquire
*/
void unlock(LockType type);
/**
* Returns true is this is lock is held at given level by the current thread.
*
* @param type the lock type to test
* @return true if the lock is held
*/
boolean isHeldByCurrentThread(LockType type);
}
/**
* 这个类是Sync接口的实现类，底层用到了Java原生的ReentrantReadWriteLock
* ReentrantReadWriteLock是读写锁，不了解的可以先去学习一下这个类的使用方法
* 不然会很难理解,其实ReadWriteLockSync的类的方法都是调用ReentrantReadWriteLock。
*/
public class ReadWriteLockSync implements Sync { //Java的读写锁,其实这个类底层都是调用它.
private final ReentrantReadWriteLock rrwl; public ReadWriteLockSync() {
this(new ReentrantReadWriteLock());
} public ReadWriteLockSync(ReentrantReadWriteLock lock) {
this.rrwl = lock;
}
/**
* 根据类型获取锁.
*/
public void lock(final LockType type) {
getLock(type).lock();
} /**
* 在给定时间内尝试获取锁.
*/
public boolean tryLock(final LockType type, final long msec) throws InterruptedException {
return getLock(type).tryLock(msec, TimeUnit.MILLISECONDS);
} /**
* 释放锁.
*/
public void unlock(final LockType type) {
getLock(type).unlock();
} //根据枚举类型获取写入锁OR读取锁.
private Lock getLock(final LockType type) {
switch (type) {
case READ:
return rrwl.readLock();
case WRITE:
return rrwl.writeLock();
default:
throw new IllegalArgumentException("We don't support any other lock type than READ or WRITE!");
}
} public ReadWriteLock getReadWriteLock() {
return rrwl;
} /**
* 判断当前线程是否获取了写入锁.
*/
public boolean isHeldByCurrentThread(LockType type) {
switch (type) {
case READ:
throw new UnsupportedOperationException("Querying of read lock is not supported.");
case WRITE:
return rrwl.isWriteLockedByCurrentThread();
default:
throw new IllegalArgumentException("We don't support any other lock type than READ or WRITE!");
}
}
}
StripedReadWriteLockSync类是负责管理锁的，会根据传入的key的hash值计算出数组的下标，原理与hashmap一样。
/**
* 这个类是锁管理器的实现类，用来管理锁.
* 这个类的实现原理就是内部维护一个锁的数组(默认2048个)
* 然后根据传入的key通过hash算法获取到一个int类型的值，这个值就是锁数组的下标.
* 这样就可以针对不同的key分别锁定提高并发效率
*/
public class StripedReadWriteLockSync implements StripedReadWriteLock { /**
* 默认锁的数量必须是2的幂
*/
public static final int DEFAULT_NUMBER_OF_MUTEXES = 2048; //锁的数组
private final ReadWriteLockSync[] mutexes;
//锁的list
private final List<ReadWriteLockSync> mutexesAsList; public StripedReadWriteLockSync() {
this(DEFAULT_NUMBER_OF_MUTEXES);
} /**
* 构造函数，初始化锁.
*/
public StripedReadWriteLockSync(int numberOfStripes) {
if ((numberOfStripes & (numberOfStripes - 1)) != 0) {
throw new CacheException("Cannot create a CacheLockProvider with a non power-of-two number of stripes");
}
if (numberOfStripes == 0) {
throw new CacheException("A zero size CacheLockProvider does not have useful semantics.");
} //初始化数组.
mutexes = new ReadWriteLockSync[numberOfStripes]; //初始化ReadWriteLockSync锁放入数组.
for (int i = 0; i < mutexes.length; i++) {
mutexes[i] = new ReadWriteLockSync();
}
mutexesAsList = Collections.unmodifiableList(Arrays.asList(mutexes));
} /**
* 根据key获取锁，相同的key获取到同一把锁。
*/
public ReadWriteLockSync getSyncForKey(final Object key) {
//根据key的hash算法在与数组的长度计算出数组下标,这个算法会保证得到的int值在0-数组长度之内不会越界.
int lockNumber = ConcurrencyUtil.selectLock(key, mutexes.length);
return mutexes[lockNumber];
} /**
* 根据key获取锁
*/
public ReadWriteLock getLockForKey(final Object key) {
int lockNumber = ConcurrencyUtil.selectLock(key, mutexes.length);
return mutexes[lockNumber].getReadWriteLock();
} /**
* Returns all internal syncs
* @return all internal syncs
*/
public List<ReadWriteLockSync> getAllSyncs() {
return mutexesAsList;
}
}
BlockingCache核心代码分析，在使用它的时候有一点需要注意，如果一个线程get方法获取element，当获取不到时会返回null，这时线程并没有释放写入锁，这点一定要注意。
所以必须要调用blockingCache.put(new Element(key, null)) 来释放锁。
/**
* A blocking decorator for an Ehcache, backed by a {@link Ehcache}.
*/
public class BlockingCache extends EhcacheDecoratorAdapter { /**
* The amount of time to block a thread before a LockTimeoutException is thrown
*/
protected volatile int timeoutMillis; private final int stripes; //线程安全的引用AtomicReference
private final AtomicReference<CacheLockProvider> cacheLockProviderReference; private final OperationObserver<GetOutcome> getObserver = operation(GetOutcome.class).named("get").of(this).tag("blocking-cache").build(); /**
* Creates a BlockingCache which decorates the supplied cache.
*/
public BlockingCache(final Ehcache cache, int numberOfStripes) throws CacheException {
super(cache);
this.stripes = numberOfStripes;
this.cacheLockProviderReference = new AtomicReference<CacheLockProvider>();
} /**
* Creates a BlockingCache which decorates the supplied cache.
*/
public BlockingCache(final Ehcache cache) throws CacheException {
this(cache, StripedReadWriteLockSync.DEFAULT_NUMBER_OF_MUTEXES);
} private CacheLockProvider getCacheLockProvider() {
CacheLockProvider provider = cacheLockProviderReference.get();
while (provider == null) {
cacheLockProviderReference.compareAndSet(null, createCacheLockProvider());
provider = cacheLockProviderReference.get();
}
return provider;
} //初始化StripedReadWriteLockSync
private CacheLockProvider createCacheLockProvider() {
Object context = underlyingCache.getInternalContext();
if (underlyingCache.getCacheConfiguration().isTerracottaClustered() && context != null) {
return (CacheLockProvider) context;
} else {
return new StripedReadWriteLockSync(stripes);
}
} /**
* Retrieve the EHCache backing cache
*/
protected Ehcache getCache() {
return underlyingCache;
} /**
* 获取元素
*/
@Override
public Element get(final Object key) throws RuntimeException, LockTimeoutException {
getObserver.begin(); //通过key获取锁,不同的key在大多数情况获取的锁都是不同的，所以性能会很好.
Sync lock = getLockForKey(key); //获取读取锁,读取锁是可以并发读的，所以效率会很好.
acquiredLockForKey(key, lock, LockType.READ); Element element;
try {
//在真正的cache里获取元素
element = underlyingCache.get(key);
} finally {
//释放读取锁
lock.unlock(LockType.READ);
} //如果元素为空，则获取写入锁，写入锁不可以并发进入的。
if (element == null) {
acquiredLockForKey(key, lock, LockType.WRITE); //再次获取元素,如果为空则直接返回，注意当前线程并没有释放锁.这里一定要注意。
element = underlyingCache.get(key);
if (element != null) {
//如果元素不为空,说明已经有线程在其它时刻put进去元素了,那么直接释放锁就OK了。
lock.unlock(LockType.WRITE);
getObserver.end(GetOutcome.HIT);
} else {
getObserver.end(GetOutcome.MISS_AND_LOCKED);
}
return element;
} else {
getObserver.end(GetOutcome.HIT);
return element;
}
} private void acquiredLockForKey(final Object key, final Sync lock, final LockType lockType) {
if (timeoutMillis > 0) {
try {
//如果设置了超时时间,那么在给定时间内获取不到锁就抛异常.
boolean acquired = lock.tryLock(lockType, timeoutMillis);
if (!acquired) {
StringBuilder message = new StringBuilder("Lock timeout. Waited more than ")
.append(timeoutMillis)
.append("ms to acquire lock for key ")
.append(key).append(" on blocking cache ").append(underlyingCache.getName());
throw new LockTimeoutException(message.toString());
}
} catch (InterruptedException e) {
throw new LockTimeoutException("Got interrupted while trying to acquire lock for key " + key, e);
}
} else {
//获取锁,获取不到就会一直等待知道获取到锁。
lock.lock(lockType);
}
} protected Sync getLockForKey(final Object key) {
return getCacheLockProvider().getSyncForKey(key);
} /**
* put元素自动释放锁,主要看doAndReleaseWriteLock.
*/
@Override
public void put(final Element element) { doAndReleaseWriteLock(new PutAction<Void>(element) {
@Override
public Void put() {
if (element.getObjectValue() != null) {
underlyingCache.put(element);
} else {
underlyingCache.remove(element.getObjectKey());
}
return null;
}
});
} private <V> V doAndReleaseWriteLock(PutAction<V> putAction) { if (putAction.element == null) {
return null;
} Object key = putAction.element.getObjectKey(); //根据key获取锁.
Sync lock = getLockForKey(key); //判断一下当前线程是否已经获取了写入锁,如果已经获取到锁，那么说明当前线程是执行get方法是元素为null时获取到锁的.
//否则说明是另一个新的线程直接调用put的方法，所以需要重新获取锁.
if (!lock.isHeldByCurrentThread(LockType.WRITE)) {
lock.lock(LockType.WRITE);
}
try {
return putAction.put();
} finally {
//在这里释放锁.
lock.unlock(LockType.WRITE);
}
} private abstract static class PutAction<V> { private final Element element; private PutAction(Element element) {
this.element = element;
}
abstract V put();
}
}