两种分布式锁实现方案(二)

四。方案2,基于redis的分布式锁

/**
*分布式锁工厂类
*/
public class RedisLockUtil {
	private static final Logger logger =  Logger.getLogger(RedisLockUtil.class);
	private static Object schemeLock = new Object();
	private static Map<String,RedisLockUtil> instances = new ConcurrentHashMap();
	public static RedisLockUtil getInstance(String schema){
		RedisLockUtil u = instances.get(schema);
		if(u==null){
			synchronized(schemeLock){
				u = instances.get(schema);
				if(u == null){
					LockObserver lo = new LockObserver(schema);		
					u = new RedisLockUtil(schema,lo);
					instances.put(schema, u);
				}
			}
		}
		return u;
	}

	private Object mutexLock = new Object();
	private Map<String,Object> mutexLockMap = new ConcurrentHashMap();
	private Map<String,RedisReentrantLock> cache  = new ConcurrentHashMap<String,RedisReentrantLock>();
	private DelayQueue<RedisReentrantLock> dq = new DelayQueue<RedisReentrantLock>();
	private AbstractLockObserver lo;
	public RedisLockUtil(String schema, AbstractLockObserver lo){	
		Thread th = new Thread(lo);
		th.setDaemon(false);
		th.setName("Lock Observer:"+schema);
		th.start();
		clearUselessLocks(schema);
		this.lo = lo;
	}

	public void clearUselessLocks(String schema){
		Thread th = new Thread(new Runnable(){
			@Override
			public void run() {
				while(!SystemExitListener.isOver()){
					try {
						RedisReentrantLock t = dq.take();						
						if(t.clear()){
							String key = t.getKey();
							synchronized(getMutex(key)){
								cache.remove(key);
							}
						}
						t.resetCleartime();
					} catch (InterruptedException e) {
					}
				}
			}

		});
		th.setDaemon(true);
		th.setName("Lock cleaner:"+schema);
		th.start();
	}

	private Object getMutex(String key){
		Object mx = mutexLockMap.get(key);
		if(mx == null){
			synchronized(mutexLock){
				mx = mutexLockMap.get(key);
				if(mx==null){
					mx = new Object();
					mutexLockMap.put(key,mx);
				}
			}
		}
		return mx;
	}

	private RedisReentrantLock getLock(String key,boolean addref){
		RedisReentrantLock lock = cache.get(key);
		if(lock == null){
			synchronized(getMutex(key)){
				lock = cache.get(key);
				if(lock == null){
					lock = new RedisReentrantLock(key,lo);
					cache.put(key, lock);
				}
			}
		}
		if(addref){
			if(!lock.incRef()){
				synchronized(getMutex(key)){
					lock = cache.get(key);
					if(!lock.incRef()){
						lock = new RedisReentrantLock(key,lo);
						cache.put(key, lock);
					}
				}
			}
		}
		return lock;
	}

	public void reset(){
		for(String s : cache.keySet()){
			getLock(s,false).unlock();
		}
	}

	/**
	 * 尝试加锁
	 * 如果当前线程已经拥有该锁的话,直接返回,表示不用再次加锁,此时不应该再调用unlock进行解锁
	 * 
	 * @param key
	 * @return
	 * @throws Exception 
	 * @throws InterruptedException
	 * @throws KeeperException
	 */
	public LockStat lock(String key) {
		return lock(key,-1);
	}

	public LockStat lock(String key,int timeout) {
		RedisReentrantLock ll = getLock(key,true);
		ll.incRef();
		try{
			if(ll.isOwner(false)){
				ll.descrRef();
				return LockStat.NONEED;
			}
			if(ll.lock(timeout)){
				return LockStat.SUCCESS;
			}else{
				ll.descrRef();
				if(ll.setCleartime()){
					dq.put(ll);
				}
				return null;
			}
		}catch(LockNotExistsException e){
			ll.descrRef();
			return lock(key,timeout);
		}catch(RuntimeException e){
			ll.descrRef();
			throw e;
		}
	}

	public void unlock(String key,LockStat stat) {
		unlock(key,stat,false);
	}

	public void unlock(String key,LockStat stat,boolean keepalive){
		if(stat == null) return;
		if(LockStat.SUCCESS.equals(stat)){
			RedisReentrantLock lock =  getLock(key,false);
			boolean candestroy = lock.unlock();
			if(candestroy && !keepalive){
				if(lock.setCleartime()){
					dq.put(lock);
				}
			}
		}
	}

	public static enum LockStat{
		NONEED,
		SUCCESS
	}
}

/**
*分布式锁本地代理类
*/
public class RedisReentrantLock implements Delayed{
	private static final Logger logger =  Logger.getLogger(RedisReentrantLock.class);
    private ReentrantLock reentrantLock = new ReentrantLock();

    private RedisLock redisLock;
    private long timeout = 3*60;
    private CountDownLatch lockcount = new CountDownLatch(1);

    private String key;
    private AbstractLockObserver observer;

    private int ref = 0;
    private Object refLock = new Object();
    private boolean destroyed = false;

    private long cleartime = -1;

    public RedisReentrantLock(String key,AbstractLockObserver observer) {	
    	this.key = key;
    	this.observer = observer;
    	initWriteLock();
    }

	public boolean isDestroyed() {
		return destroyed;
	}

	private synchronized void initWriteLock(){
		redisLock = new RedisLock(key,new LockListener(){
			@Override
			public void lockAcquired() {
				lockcount.countDown();
			}
			@Override
			public long getExpire() {
				return 0;
			}

			@Override
			public void lockError() {
				/*synchronized(mutex){
					mutex.notify();
				}*/
				lockcount.countDown();
			}
    	},observer);
	}

	public boolean incRef(){
		synchronized(refLock){
			if(destroyed) return false;
    		ref ++;
    	}
		return true;
	}

	public void descrRef(){
		synchronized(refLock){
    		ref --;
    	}
	}

	public boolean clear() {
		if(destroyed) return true;
		synchronized(refLock){
    		if(ref > 0){
    			return false;
    		}
    		destroyed = true;
    		redisLock.clear();
    		redisLock = null;
    		return true;
    	}
	}

    public boolean lock(long timeout) throws LockNotExistsException{
    	if(timeout <= 0) timeout = this.timeout;
    	//incRef();
        reentrantLock.lock();//多线程竞争时，先拿到第一层锁
        if(redisLock == null){
        	reentrantLock.unlock();
        	//descrRef();
        	throw new LockNotExistsException();
        }
        try{
        		lockcount = new CountDownLatch(1);
	        	boolean res = redisLock.trylock(timeout);
	        	if(!res){	   
	        		lockcount.await(timeout, TimeUnit.SECONDS);
					//mutex.wait(timeout*1000);
	        		if(!redisLock.doExpire()){
	        			reentrantLock.unlock();
						return false;
					}
	        	}
        	return true;
        }catch(InterruptedException e){
        	reentrantLock.unlock();
        	return false;
        }
    }

    public boolean lock() throws LockNotExistsException {
    	return lock(timeout);
    }

    public boolean unlock(){ 	 
    	if(!isOwner(true)) {
    		try{
    			throw new RuntimeException("big ================================================ error.key:"+key);
    		}catch(Exception e){
    			logger.error("err:"+e,e);
    		}
    		return false;
    	}
        try{
        	redisLock.unlock();
        	reentrantLock.unlock();//多线程竞争时，释放最外层锁
        }catch(RuntimeException e){
        	reentrantLock.unlock();//多线程竞争时，释放最外层锁
        	throw e;
        }finally{
        	descrRef();
        }
        return canDestroy();
    }

    public boolean canDestroy(){
    	synchronized(refLock){
    		return ref <= 0;
    	}
    }

    public String getKey() {
		return key;
	}

	public void setKey(String key) {
		this.key = key;
	}

	public boolean isOwner(boolean check) {
    	synchronized(refLock){
    		if(redisLock == null) {
    			logger.error("reidsLock is null:key="+key);
    			return false;
    		}
    		boolean a = reentrantLock.isHeldByCurrentThread();
    		boolean b = redisLock.isOwner();
    		if(check){
    			if(!a || !b){
    				logger.error(key+";a:"+a+";b:"+b);
    			}
    		}
    		return a && b;
    	}
    }

	public boolean setCleartime() {
		synchronized(this){
			if(cleartime>0) return false;
			this.cleartime = System.currentTimeMillis() + 10*1000; 
			return true;
		}
	}

	public void resetCleartime(){
		synchronized(this){
			this.cleartime = -1;
		}
	}

	@Override
	public int compareTo(Delayed object) {
		if(object instanceof RedisReentrantLock){
			RedisReentrantLock t = (RedisReentrantLock)object;
	        long l = this.cleartime - t.cleartime;

			if(l > 0) return 1 ; //比当前的小则返回1，比当前的大则返回-1，否则为0
			else if(l < 0 ) return -1;
			else return 0;
		}
		return 0;
	}

	@Override
	public long getDelay(TimeUnit unit) {
		 long d = unit.convert(cleartime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
	     return d;
	}

}

/**
*使用Redis实现的分布式锁
*基本工作原理如下：
*1. 使用setnx(key,时间戮+超时)，如果设置成功，则直接拿到锁
*2. 如果设置不成功，获取key的值v1（它的到期时间戮）,跟当前时间对比，看是否已经超时
*3. 如果超时（说明拿到锁的结点已经挂掉)，v2=getset(key,时间戮+超时+1)，判断v2是否等于v1，如果相等，加锁成功，否则加锁失败，等过段时间再重试(200MS)
*/
public class RedisLock implements LockListener{
	private String key;
	private boolean owner = false;
	private AbstractLockObserver observer = null;
	private LockListener lockListener = null;
	private boolean waiting = false;
	private long expire;//锁超时时间，以秒为单位
	private boolean expired = false;

	public RedisLock(String key, LockListener lockListener, AbstractLockObserver observer) {
		this.key = key;
		this.lockListener = lockListener;
		this.observer = observer;
	}

	public boolean trylock(long expire) {
		synchronized(this){
			if(owner){
				return true;
			}
			this.expire = expire;
			this.expired = false;
			if(!waiting){
				owner = observer.tryLock(key,expire);
				if(!owner){
					waiting = true;
					observer.addLockListener(key, this);
				}
			}
			return owner;
		}
	}

	public boolean isOwner() {
		return owner;
	}

	public void unlock() {
		synchronized(this){
			observer.unLock(key);
			owner = false;
		}
	}

	public void clear() {
		synchronized(this){
			if(waiting) {
				observer.removeLockListener(key);
				waiting = false;
			}
		}
	}

	public boolean doExpire(){
		synchronized(this){
			if(owner) return true;
			if(expired) return false;
			expired =  true;
			clear();
		}
		return false;
	}

	@Override
	public void lockAcquired() {
		synchronized(this){
			if(expired){
				unlock();
				return;
			}
			owner = true;
			waiting = false;
		}
		lockListener.lockAcquired();
	}

	@Override
	public long getExpire() {
		return this.expire;
	}

	@Override
	public void lockError() {
		synchronized(this){
			owner = false;
			waiting = false;
			lockListener.lockError();
		}
	}

}

public class LockObserver extends AbstractLockObserver implements Runnable{
	private CacheRedisClient client;
	private Object mutex = new Object();
	private Map<String,LockListener> lockMap = new ConcurrentHashMap();
	private boolean stoped = false;
	private long interval = 500;
	private boolean terminated = false;
	private CountDownLatch doneSignal = new CountDownLatch(1);
	
	public LockObserver(String schema){
		client = new CacheRedisClient(schema);
		
		SystemExitListener.addTerminateListener(new ExitHandler(){
			public void run() {
				stoped = true;
				try {
					doneSignal.await();
				} catch (InterruptedException e) {
				}
			}
		});
	}
	
	
	public void addLockListener(String key,LockListener listener){
		if(terminated){
			listener.lockError();
			return;
		}
		synchronized(mutex){
			lockMap.put(key, listener);
		}
	}
	
	public void removeLockListener(String key){
		synchronized(mutex){
			lockMap.remove(key);
		}
	}
	
	@Override
	public void run() {
		while(!terminated){
			long p1 = System.currentTimeMillis();
			Map<String,LockListener> clone = new HashMap();
			synchronized(mutex){
				clone.putAll(lockMap);
			}			
			Set<String> keyset = clone.keySet();
			if(keyset.size() > 0){
				ConnectionFactory.setSingleConnectionPerThread(keyset.size());
				for(String key : keyset){
					LockListener ll = clone.get(key);
					try{
					    if(tryLock(key,ll.getExpire())) {
					    	ll.lockAcquired();
					    	removeLockListener(key);
					    }
					}catch(Exception e){
						ll.lockError();
						removeLockListener(key);
					}
				}
				ConnectionFactory.releaseThreadConnection();
			}else{
				if(stoped){
					terminated = true;
					doneSignal.countDown();
					return;
				}
			}
			try {
				long p2 = System.currentTimeMillis();
				long cost = p2 - p1;
				if(cost <= interval){
					Thread.sleep(interval - cost);
				}else{
					Thread.sleep(interval*2);
				}
			} catch (InterruptedException e) {
			}
		}
		
	}
	
	
	/**
	 * 超时时间单位为s!!!
	 * @param key
	 * @param expire
	 * @return
	 */
	public boolean tryLock(final String key,final long expireInSecond){
		if(terminated) return false;
		final long tt = System.currentTimeMillis();
		final long expire = expireInSecond * 1000;
		final Long ne = tt + expire;
		List<Object> mm = client.multi(key, new MultiBlock(){
			@Override
			public void execute() {
				transaction.setnxObject(key, ne);
				transaction.get(SafeEncoder.encode(key));
			}
		});
		Long res = (Long)mm.get(0);
	    if(new Long(1).equals(res)) {
	    	return true;
	    }else{
	    	byte[] bb = (byte[])mm.get(1);
			Long ex = client.deserialize(bb);
	    	if(ex == null || tt > ex){
	    		Long old = client.getSet(key, new Long(ne+1));
	    		if(old == null || (ex == null&&old==null) || (ex!=null&&ex.equals(old))){
	    			return true;
	    		}
	    	}
	    }
	    return false;
	}

	public void unLock(String key){
		client.del(key);
	}
}

使用本方案实现的分布式锁，可以完美地解决锁重入问题；通过引入超时也避免了死锁问题；性能方面，笔者自测试结果如下：

500线程 tps = 35000
[root@DB1 benchtest-util]# target/benchtest/bin/TestFastRedis /data/config/util/config_0_11.properties lock 500 500000
线程总时间：6553466；平均:13.106932
实际总时间：13609; 平均：0.027218

TPS达到35000，比方案1强了整整一个数量级；

五。总结
本文介绍了两种分布式锁的实现方案；方案1最大的优势在于避免结点挂掉后导致的死锁；方案2最大的优势在于性能超强；在实际生产过程中，结合自身情况来决定最适合的分布式锁方案是架构师的必修课。

两种分布式锁实现方案(二)

原文：http://my.oschina.net/gaowm/blog/522483