温馨提示×

温馨提示×

您好,登录后才能下订单哦!

密码登录×
登录注册×
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》

​基于redis的分布式锁怎么实现

发布时间:2021-12-23 12:00:43 阅读:158 作者:iii 栏目:云计算
开发者测试专用服务器限时活动,0元免费领,库存有限,领完即止! 点击查看>>

本篇内容介绍了“基于redis的分布式锁怎么实现”的有关知识,在实际案例的操作过程中,不少人都会遇到这样的困境,接下来就让小编带领大家学习一下如何处理这些情况吧!希望大家仔细阅读,能够学有所成!

基于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 == nullreturn;
		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强了整整一个数量级

“基于redis的分布式锁怎么实现”的内容就介绍到这里了,感谢大家的阅读。如果想了解更多行业相关的知识可以关注亿速云网站,小编将为大家输出更多高质量的实用文章!

亿速云「云服务器」,即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘,价格低至29元/月。点击查看>>

向AI问一下细节

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

原文链接:https://my.oschina.net/boltwu/blog/425652

AI

开发者交流群×