细粒度高并行锁 KeyLock
本帖最后由 icebamboo_moyun 于 2013-07-20 18:26:34 编辑 一个细粒度的锁，在某些场景能比synchronized，ReentrantLock等获得更高的并行度更好的性能，感兴趣的可以看看
具体讲解在：http://blog.csdn.net/icebamboo_moyun/article/details/9391915
KeyLock代码：
public class KeyLock<K> {
// 保存所有锁定的KEY及其信号量
private final ConcurrentMap<K, Semaphore> map = new ConcurrentHashMap<K, Semaphore>();
// 保存每个线程锁定的KEY及其锁定计数
private final ThreadLocal<Map<K, LockInfo>> local = new ThreadLocal<Map<K, LockInfo>>() {
@Override
protected Map<K, LockInfo> initialValue() {
return new HashMap<K, LockInfo>();
}
};
/**
* 锁定key，其他等待此key的线程将进入等待，直到调用{@link #unlock(K)}
* 使用hashcode和equals来判断key是否相同，因此key必须实现{@link #hashCode()}和
* {@link #equals(Object)}方法
*
* @param key
*/
public void lock(K key) {
if (key == null)
return;
LockInfo info = local.get().get(key);
if (info == null) {
Semaphore current = new Semaphore(1);
current.acquireUninterruptibly();
Semaphore previous = map.put(key, current);
if (previous != null)
previous.acquireUninterruptibly();
local.get().put(key, new LockInfo(current));
} else {
info.lockCount++;
}
}
/**
* 释放key，唤醒其他等待此key的线程
* @param key
*/
public void unlock(K key) {
if (key == null)
return;
LockInfo info = local.get().get(key);
if (info != null && --info.lockCount == 0) {
info.current.release();
map.remove(key, info.current);
local.get().remove(key);
}
}
/**
* 锁定多个key
* 建议在调用此方法前先对keys进行排序，使用相同的锁定顺序，防止死锁发生
* @param keys
*/
public void lock(K[] keys) {
if (keys == null)
return;
for (K key : keys) {
lock(key);
}
}
/**
* 释放多个key
* @param keys
*/
public void unlock(K[] keys) {
if (keys == null)
return;
for (K key : keys) {
unlock(key);
}
}
private static class LockInfo {
private final Semaphore current;
private int lockCount;
private LockInfo(Semaphore current) {
this.current = current;
this.lockCount = 1;
}
}
}
java 锁 高并发 KeyLock [解决办法]
看了你博客上做的测试之后，我有一种更快的方法。
方法是不用transfer函数，代码如下：
public class Test {
private final AtomicInteger[] iAccounts;
public Test(int count, int money) {
iAccounts = new AtomicInteger[count];
for (int i = 0; i < count; i++)
iAccounts[i] = new AtomicInteger(money);
}
void transferAtomic(int from, int to, int money) {
if (iAccounts[from].get() < money)
return;
iAccounts[from].getAndAdd(-money);
try {
Thread.sleep(2);
}
catch (Exception e) {
}
iAccounts[to].getAndAdd(money);
}
//...以下略
新建一个TransferRunner，用transferAtomic而不用transfer函数。
此方法比用KeyLock快了10倍。测试的时候去掉那句sleep，就得到了代码的纯效率
不同点在于：
1. KeyLock每次针对两个账户进行了lock，而transferAtomic从本质上来说每次只lock一个账户，也就是它将转出和转入当做两个独立的操作进行：这并不影响账面的正确性
2. KeyLock有很多内存分配（装箱操作、Semaphore、包括lock之前的Integer{key1, key2}）、Hash查询，而transferAtomic除了初始化之外没有任何内存分配及Hash查询，因而效率更高
事实上在需要使用KeyLock的时候总会有更优的替代方法（至少我想不到非得用KeyLock的场景），因为它的本质就是分别锁住两个对象（并没有同时，除非在lock()函数中再次加入同步机制）。
[解决办法]
以前从来没用过信号量，仿照你的代码写了一个用普通 Object 加锁的类，用了 ConcurrentHashMap 和 double-checked lock 来保证获取锁时重复的读操作 lock free:
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
public class KeyLocks<L> {
private final Map<L, Object> locks = new ConcurrentHashMap<L, Object>();
public Object getLock(L key) {
Object result = locks.get(key);
if( result == null ) synchronized (locks) {
result = locks.get(key);
if( result == null ) {
result = new Object();
locks.put(key, result);
}
}
return result;
}
}
class Test {
private final int[] accounts;
private final KeyLocks<Integer> keyLocks;
Test(int count, int money) {
if( count <= 0 )
throw new IllegalArgumentException("Invalid number of accounts: " + count);
if( money < 0 )
throw new IllegalArgumentException("Invalid initial balance: " + money);
accounts = new int[count];
for(int i=0; i<accounts.length; i++)
accounts[i] = money;
keyLocks = new KeyLocks<Integer>();
}
boolean transfer(int from, int to, int money) {
if( from < 0
[解决办法]
from >= accounts.length )
throw new IndexOutOfBoundsException("Invalid from account: " + from);
if( to < 0
[解决办法]
to >= accounts.length )
throw new IndexOutOfBoundsException("Invalid to account: " + to);
if( from == to )
throw new IllegalArgumentException("Cannot transfer between same account.");
if( money < 0 )
throw new IllegalArgumentException("Invalid transfer amount: " + money);
synchronized (keyLocks.getLock(from)) {
if( accounts[from] < money )
return false;
accounts[from] -= money;
}
synchronized (keyLocks.getLock(to)) {
accounts[to] += money;
}
return true;
}
}
我不知道每次访问 ThreadLocal 和 建立 Semaphore 实例的开销有多大，也许是可以忽略不计的，那么这两种实现也许效率差不多。