自己动手写带有事务支持的分布式Key-Value存储系统——锁管理器

对于锁事务的实现，里面需要使用大量的读写锁，大量线程同步地对数据加锁，难免会产生死锁，所以锁管理器不仅需要管理所有分配的锁，还要能够自动检测出死锁，并且主动解除死锁状态。

死锁检测的原理：

死锁检测方法中通常使用等待图 WFG（Wait-For Graph）作为表达事务间等待关系的数学模型。在 WFG 中，结点 ti

表示事务，边(ti,tj)表示事务 ti等待 tj，该边在 ti申请对 tj现持有的资源 sr 加锁时建立，在 tj释放其在资源 sr 上的锁后删除。

当 WFG 中出现环时，表明系统中存在死锁。对应于每个事务（因为本系统是基于线程的，每一个事务使用一个线程，所以这里的

事务可以和线程等价），如果事务想要加的读写锁和另一个事务中的所占用的锁互斥，则存在一条有向辺，遍历整个事务，就可以建立一个全局的WFG，然后通过检测

有向图中的环，就可以获得死锁的事务集合。

有向图的环的检测

递归删除入度为0的顶点，直到不存在入度为0的顶点，剩下的就是顶点就组成有向环。具体证明查见《图论》

具体实现：代码(点击打开链接)

有向图（不完整，只是用来实现环的检测，算法的复杂度n^2)：Github中查看.寻找环的函数为findRings()

/**
* Alipay.com Inc.
* Copyright (c) 2004-2014 All Rights Reserved.
*/
package org.footoo.common.datastruct.graphic;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

/**
* 有向图
* <ul>
*  <li>有向图的表示使用邻接表的方式</li>
*  <li>邻接表里存放有向边的起始节点列表</li>
*  <li>有向图节点存放的数据必须不能相等，因为数据相等的节点被当成一个节点</li>
* </ul>
*
* @author jeff
* @version $Id: DirectGraphic.java, v 0.1 2014年3月25日 下午3:30:11 jeff Exp $
*/
public class DirectGraphic<T> {
/** 邻接表方式表示的图 */
private Map<GraphicNode<T>, Set<GraphicNode<T>>> graphic = new HashMap<GraphicNode<T>, Set<GraphicNode<T>>>();

/**
* 默认的构造器
*/
public DirectGraphic() {

}

/**
* 拷贝构造函数
*
* @param other
*/
public DirectGraphic(DirectGraphic<T> other) {
//进行深拷贝
for (Map.Entry<GraphicNode<T>, Set<GraphicNode<T>>> entry : other.graphic.entrySet()) {
Set<GraphicNode<T>> set = new HashSet<>(entry.getValue());
this.graphic.put(entry.getKey(), set);
}
}

/**
* 获取图里面构成环的所有的节点的集合<br />
* 使用的去边法
*
*
* @return
*/
public Set<GraphicNode<T>> findRings() {
//拷贝一份
DirectGraphic<T> copy = new DirectGraphic<>(this);

boolean needAgain = true;
while (needAgain) {
needAgain = false;
Iterator<GraphicNode<T>> iterator = copy.graphic.keySet().iterator();
while (iterator.hasNext()) {
GraphicNode<T> node = iterator.next();
Set<GraphicNode<T>> set = copy.graphic.get(node);
//入度为0
if (set.isEmpty()) {
iterator.remove();
removeSetNode(node, copy.graphic);
needAgain = true;
}
}

}

return copy.graphic.keySet();
}

/**
* 移除邻接表中的node节点
*
* @param node
* @param map
*/
private void removeSetNode(GraphicNode<T> node, Map<GraphicNode<T>, Set<GraphicNode<T>>> map) {
//移除大的节点
//map.remove(node);
//移除邻接表集合中的节点
for (Map.Entry<GraphicNode<T>, Set<GraphicNode<T>>> entry : map.entrySet()) {
entry.getValue().remove(node);
}
}

/**
* 插入有向边
*
* @param from
* @param to
*/
public void insertEdge(T from, T to) {
insertEdge(createOrGetNode(from), createOrGetNode(to));
}

/**
* 插入有向边
*
* @param from
* @param to
*/
public void insertEdge(GraphicNode<T> from, GraphicNode<T> to) {
//获取起始节点的集合
Set<GraphicNode<T>> nodes = graphic.get(to);
if (nodes == null) {
nodes = new HashSet<>();
graphic.put(to, nodes);
}
nodes.add(from);
}

/**
* 创建（如果不存在）或者获取（存在）对应的节点
*
* @param value
* @return
*/
public GraphicNode<T> createOrGetNode(T value) {
GraphicNode<T> ret = new GraphicNode<T>(value);
if (!graphic.containsKey(ret)) {
graphic.put(ret, new HashSet<GraphicNode<T>>());
} else {
ret = getNode(value);
}

assert (ret != null);
return ret;
}

/**
* 获取node
*
* @param value
* @return
*/
private GraphicNode<T> getNode(T value) {
for (GraphicNode<T> node : graphic.keySet()) {
if (node.getValue().equals(value)) {
return node;
}
}
return null;
}
}

锁管理器，完成实际的锁的管理，其中的死锁检测由内部类DeadLockDetecter的定时服务完成，github中查看

/**
* Alipay.com Inc.
* Copyright (c) 2004-2014 All Rights Reserved.
*/
package org.footoo.storage.transaction.lock;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import org.footoo.common.datastruct.graphic.DirectGraphic;
import org.footoo.common.datastruct.graphic.GraphicNode;
import org.footoo.common.log.Logger;
import org.footoo.common.log.LoggerFactory;
import org.footoo.common.service.TaskService;

/**
* 锁管理器
*
* <ul>
*  <li>统计线程说占用的锁</li>
*  <li>死锁检测</li>
* </ul>
*
* @author jeff
* @version $Id: LockManager.java, v 0.1 2014年3月17日 下午11:13:34 jeff Exp $
*/
public class LockManager {
/** 所有的读写锁 */
private final Set<RWLock>                         rwLocks         = new HashSet<>();

/** 每个线程的锁使用情况 */
private ConcurrentHashMap<Thread, ThreadLockInfo> threadLockInfos = new ConcurrentHashMap<>();
/** 日志 */
private static final Logger                       logger          = LoggerFactory
.getLogger(LockManager.class);

public LockManager() {
//启动死锁检测服务
new DeadLockDetecter().start();
}

/**
* 创建一个读写锁
*
* @return
*/
public RWLock createRwLock() {
RWLock rwLock = new RandomSyncRWLock();
rwLocks.add(rwLock);

return rwLock;
}

/**
* 释放锁
*
* @param rwlock
*/
public void release(RWLock rwlock) {
rwlock.release();
rwLocks.remove(rwlock);
}

/**
* 释放当前线程的rwlock的锁信息
*
* @param rwLock
*/
public void releaseSelf(RWLock rwLock) {
rwLock.releaseSelf();
}

/**
* 释放当前线程所有读写锁
*/
public void releaseSelf() {
ThreadLockInfo threadLockInfo = threadLockInfos.remove(Thread.currentThread());
if (threadLockInfo != null) {
//
for (RWLock rwLock : threadLockInfo.getLockSet()) {
rwLock.releaseSelf();
}
}
}

/**
* 释放所有的锁，这个一般在系统终结的时候使用
*/
public void releaseAll() {
for (RWLock rwLock : rwLocks) {
rwLock.release();
}
}

/**
* 加读锁
*
* @param rwLock
*/
public void addReadLock(RWLock rwLock) throws DeadLockInterruptException {
//期待读锁，如果一直阻塞在加锁过程，死锁检测服务可以检测到
setWantedLock(rwLock, WantedLock.LOCK_TYPE_READ);
try {
rwLock.readLock().lock();
addReadCount(rwLock, 1);
} catch (DeadLockInterruptException e) {
//重新抛出异常
throw e;
} finally {
//释放想要加的锁
delWantedLock();
}

}

/**
* 读锁解锁
*
* @param rwLock
*/
public void readUnlock(RWLock rwLock) {
rwLock.readLock().unlock();
addReadCount(rwLock, -1);
}

/**
* 写锁加锁
*
* @param rwLock 读写锁
*/
public void addWriteLock(RWLock rwLock) throws DeadLockInterruptException {
setWantedLock(rwLock, WantedLock.LOCK_TYPE_WRITE);
try {
rwLock.writeLock().lock();
addWriteCount(rwLock, 1);
} catch (DeadLockInterruptException e) {
throw e;
} finally {
delWantedLock();
}
}

/**
* 写锁解锁
*
* @param rwLock
*/
public void writeUnlock(RWLock rwLock) {
rwLock.writeLock().unlock();
addWriteCount(rwLock, -1);
}

/**
* 死锁检测服务
* <ul>
*  <li>死锁检测使用的是有向图的方法，而且整个过程不加锁，检测出来的状态可能不会是死锁，
*      但死锁肯定会被检测出来</li>
* <li>被检测的死锁线程列表中会随机选取一个线程进行终结</li>
* </ul>
*
* @author jeff
* @version $Id: LockManager.java, v 0.1 2014年3月25日 下午9:16:18 jeff Exp $
*/
private class DeadLockDetecter extends TaskService {
public DeadLockDetecter() {
this.setDelay(1000);
//每2分钟启动一次
this.setPeriod(2 * 1000);
}

/**
* 创建死锁检测的有向图,
* 对于每一个线程所期望加的锁，如果和其他线程之间有互斥的依赖，就增加一条有向边
*
* @return
*/
private DirectGraphic<Thread> createDirectGraphic() {
DirectGraphic<Thread> result = new DirectGraphic<>();
Iterator<Thread> iterator = threadLockInfos.keySet().iterator();
//开始建立死锁检测的有向图
while (iterator.hasNext()) {
Thread thread = iterator.next();
ThreadLockInfo threadLockInfo = threadLockInfos.get(thread);
WantedLock wantedLock = threadLockInfo.getWantedLock();
if (wantedLock == null) {
continue;
}
//
for (Map.Entry<Thread, ThreadLockInfo> entry : threadLockInfos.entrySet()) {
//跳过当前线程
if (entry.getKey().equals(thread)) {
continue;
}
ThreadLockInfo threadLockInfo2 = entry.getValue();
//期望的是读锁
if (wantedLock.getLockType() == WantedLock.LOCK_TYPE_READ) {
if (threadLockInfo2.isInWriteLocks(wantedLock.getRwLock())) {
result.insertEdge(thread, entry.getKey());
}
} else {//期望的是写锁
if (threadLockInfo2.isInReadLocks(wantedLock.getRwLock())
|| threadLockInfo2.isInWriteLocks(wantedLock.getRwLock())) {
result.insertEdge(thread, entry.getKey());
}
}
}
}
return result;
}

/**
* @see java.util.TimerTask#run()
*/
@Override
public void run() {
//logger.debug("开始死锁检测");
DirectGraphic<Thread> directGraphic = createDirectGraphic();
//获取有向图的环
Set<GraphicNode<Thread>> threadSet = directGraphic.findRings();
//发生了死锁
if (!threadSet.isEmpty()) {
logger.warn("出现了死锁");
for (GraphicNode<Thread> node : threadSet) {
logger.warn("死锁线程" + node.getValue());
}

//唤醒第一个
Thread t1 = threadSet.iterator().next().getValue();
t1.interrupt();
}
}

}

/**
* 设置期望获取的锁，这个用于死锁检测
*
* @param rwLock
* @param type
*/
private void setWantedLock(RWLock rwLock, int type) {
ThreadLockInfo threadLockInfo = getAndSetThreadLockInfo();
threadLockInfo.setWantedLock(new WantedLock(rwLock, type));
}

/**
* 去掉想添加的锁
*/
private void delWantedLock() {
ThreadLockInfo threadLockInfo = getAndSetThreadLockInfo();
threadLockInfo.removeWantedLock();
}

/**
* 增加读锁的使用计数
* 这个函数是和本线程相关的，不会存在竞争
*
* @param rwLock
* @param count
*/
private void addReadCount(RWLock rwLock, int count) {
ThreadLockInfo threadLockInfo = getAndSetThreadLockInfo();

threadLockInfo.addReadUseCount(rwLock, count);
}

/**
* 增加写锁的使用计数
*
* @param rwLock
* @param count
*/
private void addWriteCount(RWLock rwLock, int count) {
ThreadLockInfo threadLockInfo = getAndSetThreadLockInfo();
threadLockInfo.addWriteUseCount(rwLock, count);
}

/**
* 获取并且设置ThreadLockInfo(如果不存在）
*
* @return
*/
private ThreadLockInfo getAndSetThreadLockInfo() {
ThreadLockInfo threadLockInfo = threadLockInfos.get(Thread.currentThread());
if (threadLockInfo == null) {
threadLockInfo = new ThreadLockInfo();
threadLockInfos.put(Thread.currentThread(), threadLockInfo);
}
return threadLockInfo;
}

/**
* 线程所持有锁的情况
*
* @author jeff
* @version $Id: LockManager.java, v 0.1 2014年3月25日 上午9:04:43 jeff Exp $
*/
public static class ThreadLockInfo {
/** 线程所占有的读锁 */
private final ConcurrentHashMap<RWLock, LockInfo> rLocks = new ConcurrentHashMap<>();
/** 线程所占用的写锁 */
private final ConcurrentHashMap<RWLock, LockInfo> wLocks = new ConcurrentHashMap<>();
/** 所期待的锁 */
private WantedLock                                wantedLock;

/**
* 移除期待的锁
*/
public final void removeWantedLock() {
wantedLock = null;
}

/**
* Getter method for property <tt>wantedLock</tt>.
*
* @return property value of wantedLock
*/
public final WantedLock getWantedLock() {
return wantedLock;
}

/**
* Setter method for property <tt>wantedLock</tt>.
*
* @param wantedLock value to be assigned to property wantedLock
*/
public final void setWantedLock(WantedLock wantedLock) {
this.wantedLock = wantedLock;
}

/**
* Getter method for property <tt>rLocks</tt>.
*
* @return property value of rLocks
*/
public final ConcurrentHashMap<RWLock, LockInfo> getrLocks() {
return rLocks;
}

/**
* Getter method for property <tt>wLocks</tt>.
*
* @return property value of wLocks
*/
public final ConcurrentHashMap<RWLock, LockInfo> getwLocks() {
return wLocks;
}

/**
* 测试rwLock是否在读锁集合
*
* @param rwLock
* @return
*/
public boolean isInReadLocks(RWLock rwLock) {
return rLocks.containsKey(rwLock);
}

/**
* 测试rwLock是否在写锁集合
*
* @param rwLock
* @return
*/
public boolean isInWriteLocks(RWLock rwLock) {
return wLocks.containsKey(rwLock);
}

/**
* 获取读锁集合
*
* @return
*/
public Set<RWLock> getReadSet() {
return rLocks.keySet();
}

/**
* 获取写锁集合
*
* @return
*/
public Set<RWLock> getWriteSet() {
return wLocks.keySet();
}

/**
* 获取线程锁占有的锁的集合
*
* @return
*/
public Set<RWLock> getLockSet() {
Set<RWLock> result = new HashSet<RWLock>(rLocks.keySet());
result.addAll(wLocks.keySet());
return result;
}

/**
* 给读锁增加使用计数
*
* @param rwLock
* @param count
*/
public void addReadUseCount(RWLock rwLock, int count) {
addUseCount(rwLock, count, rLocks);
}

/**
* 给写锁增加使用计数
*
* @param rwLock
* @param count
*/
public void addWriteUseCount(RWLock rwLock, int count) {
addUseCount(rwLock, count, wLocks);
}

/**
* 给rwLock增加number个使用计数
*
* @param rwLock
* @param number
* @param locks
*/
private void addUseCount(RWLock rwLock, int number,
ConcurrentHashMap<RWLock, LockInfo> locks) {
LockInfo lockInfo = locks.get(rwLock);
if (lockInfo == null) {
lockInfo = new LockInfo(rwLock, 0);
locks.put(rwLock, lockInfo);
}
lockInfo.setUseCount(lockInfo.getUseCount() + number);
//
if (lockInfo.getUseCount() <= 0) {
locks.remove(rwLock);
}
}

}

/**
* 线程所期望的锁信息
*
* @author jeff
* @version $Id: LockManager.java, v 0.1 2014年3月25日 上午9:09:31 jeff Exp $
*/
public static class WantedLock {
/** 读锁 */
public static final int LOCK_TYPE_READ  = 0;
/** 写锁 */
public static final int LOCK_TYPE_WRITE = 1;

/** 所期望的锁 */
private RWLock          rwLock;
/** 所期望的锁类型 */
private int             lockType;

/**
* 默认的构造器
*/
public WantedLock() {

}

/**
* 构造器
*
* @param rwLock
* @param lockType
*/
public WantedLock(RWLock rwLock, int lockType) {
this.rwLock = rwLock;
this.lockType = lockType;
}

/**
* Getter method for property <tt>rwLock</tt>.
*
* @return property value of rwLock
*/
public final RWLock getRwLock() {
return rwLock;
}

/**
* Setter method for property <tt>rwLock</tt>.
*
* @param rwLock value to be assigned to property rwLock
*/
public final void setRwLock(RWLock rwLock) {
this.rwLock = rwLock;
}

/**
* Getter method for property <tt>lockType</tt>.
*
* @return property value of lockType
*/
public final int getLockType() {
return lockType;
}

/**
* Setter method for property <tt>lockType</tt>.
*
* @param lockType value to be assigned to property lockType
*/
public final void setLockType(int lockType) {
this.lockType = lockType;
}

}

/**
* 锁使用信息
*
* @author jeff
* @version $Id: LockManager.java, v 0.1 2014年3月20日 下午5:48:35 jeff Exp $
*/
public static class LockInfo {
/** 锁 */
private RWLock rwLock;
/** 使用计数 */
private int    useCount;

public LockInfo() {

}

public LockInfo(RWLock rwLock, int useCount) {
this.rwLock = rwLock;
this.useCount = useCount;
}

public int hashCode() {
return rwLock.hashCode();
}

public boolean equals(Object other) {
if (this == other) {
return true;
}
if (!(other instanceof LockInfo)) {
return false;
}
return this.rwLock.equals(((LockInfo) other).rwLock);
}

/**
* Getter method for property <tt>rwLock</tt>.
*
* @return property value of rwLock
*/
public RWLock getRwLock() {
return rwLock;
}

/**
* Setter method for property <tt>rwLock</tt>.
*
* @param rwLock value to be assigned to property rwLock
*/
public void setRwLock(RWLock rwLock) {
this.rwLock = rwLock;
}

/**
* Getter method for property <tt>useCount</tt>.
*
* @return property value of useCount
*/
public int getUseCount() {
return useCount;
}

/**
* Setter method for property <tt>useCount</tt>.
*
* @param useCount value to be assigned to property useCount
*/
public void setUseCount(int useCount) {
this.useCount = useCount;
}

}

}