分布式锁实现方式二 基于Redis的分布式锁

分布式锁实现方式二 基于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 == 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;
}
}

<a href="http://www.nbso.ca/">
nbso online
casino reviews
</a>

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最大的优势在于性能超强；在实际生产过程中，结合自身情况来决定最适合的分布式锁方案是架构师的必修课。

-------------------------------------------

三 另一篇关于redis分布式锁实现的文章

背景

在很多互联网产品应用中，有些场景需要加锁处理，比如：秒杀，全局递增ID，楼层生成等等。大部分的解决方案是基于DB实现的，Redis为单进程单线程模式，采用队列模式将并发访问变成串行访问，且多客户端对Redis的连接并不存在竞争关系。其次Redis提供一些命令SETNX，GETSET，可以方便实现分布式锁机制。

Redis命令介绍

使用Redis实现分布式锁，有两个重要函数需要介绍

SETNX命令（SET if Not eXists）

语法：

SETNX key value

功能：

当且仅当 key 不存在，将 key 的值设为 value ，并返回1；若给定的 key 已经存在，则 SETNX 不做任何动作，并返回0。

GETSET命令

语法：

GETSET key value

功能：

将给定 key 的值设为 value ，并返回 key 的旧值 (old value)，当 key 存在但不是字符串类型时，返回一个错误，当key不存在时，返回nil。

GET命令

语法：

GET key

功能：

返回 key 所关联的字符串值，如果 key 不存在那么返回特殊值 nil 。

DEL命令

语法：

DEL key [KEY …]

功能：

删除给定的一个或多个 key ,不存在的 key 会被忽略。

兵贵精，不在多。分布式锁，我们就依靠这四个命令。但在具体实现，还有很多细节，需要仔细斟酌，因为在分布式并发多进程中，任何一点出现差错，都会导致死锁，hold住所有进程。

加锁实现

SETNX 可以直接加锁操作，比如说对某个关键词foo加锁，客户端可以尝试

SETNX foo.lock <current unix time>

如果返回1，表示客户端已经获取锁，可以往下操作，操作完成后，通过

DEL foo.lock

命令来释放锁。

如果返回0，说明foo已经被其他客户端上锁，如果锁是非堵塞的，可以选择返回调用。如果是堵塞调用调用，就需要进入以下个重试循环，直至成功获得锁或者重试超时。理想是美好的，现实是残酷的。仅仅使用SETNX加锁带有竞争条件的，在某些特定的情况会造成死锁错误。

处理死锁

在上面的处理方式中，如果获取锁的客户端端执行时间过长，进程被kill掉，或者因为其他异常崩溃，导致无法释放锁，就会造成死锁。所以，需要对加锁要做时效性检测。因此，我们在加锁时，把当前时间戳作为value存入此锁中，通过当前时间戳和Redis中的时间戳进行对比，如果超过一定差值，认为锁已经时效，防止锁无限期的锁下去，但是，在大并发情况，如果同时检测锁失效，并简单粗暴的删除死锁，再通过SETNX上锁，可能会导致竞争条件的产生，即多个客户端同时获取锁。

C1获取锁，并崩溃。C2和C3调用SETNX上锁返回0后，获得foo.lock的时间戳，通过比对时间戳，发现锁超时。

C2 向foo.lock发送DEL命令。

C2 向foo.lock发送SETNX获取锁。

C3 向foo.lock发送DEL命令，此时C3发送DEL时，其实DEL掉的是C2的锁。

C3 向foo.lock发送SETNX获取锁。

此时C2和C3都获取了锁，产生竞争条件，如果在更高并发的情况，可能会有更多客户端获取锁。所以，DEL锁的操作，不能直接使用在锁超时的情况下，幸好我们有GETSET方法，假设我们现在有另外一个客户端C4，看看如何使用GETSET方式，避免这种情况产生。

C1获取锁，并崩溃。C2和C3调用SETNX上锁返回0后，调用GET命令获得foo.lock的时间戳T1，通过比对时间戳，发现锁超时。

C4 向foo.lock发送GESET命令，

GETSET foo.lock <current unix time>

并得到foo.lock中老的时间戳T2

如果T1=T2，说明C4获得时间戳。

如果T1!=T2，说明C4之前有另外一个客户端C5通过调用GETSET方式获取了时间戳，C4未获得锁。只能sleep下，进入下次循环中。

现在唯一的问题是，C4设置foo.lock的新时间戳，是否会对锁产生影响。其实我们可以看到C4和C5执行的时间差值极小，并且写入foo.lock中的都是有效时间错，所以对锁并没有影响。

为了让这个锁更加强壮，获取锁的客户端，应该在调用关键业务时，再次调用GET方法获取T1，和写入的T0时间戳进行对比，以免锁因其他情况被执行DEL意外解开而不知。以上步骤和情况，很容易从其他参考资料中看到。客户端处理和失败的情况非常复杂，不仅仅是崩溃这么简单，还可能是客户端因为某些操作被阻塞了相当长时间，紧接着 DEL 命令被尝试执行(但这时锁却在另外的客户端手上)。也可能因为处理不当，导致死锁。还有可能因为sleep设置不合理，导致Redis在大并发下被压垮。最为常见的问题还有

GET返回nil时应该走那种逻辑？

第一种走超时逻辑

C1客户端获取锁，并且处理完后，DEL掉锁，在DEL锁之前。C2通过SETNX向foo.lock设置时间戳T0 发现有客户端获取锁，进入GET操作。

C2 向foo.lock发送GET命令，获取返回值T1(nil)。

C2 通过T0>T1+expire对比，进入GETSET流程。

C2 调用GETSET向foo.lock发送T0时间戳，返回foo.lock的原值T2

C2 如果T2=T1相等，获得锁，如果T2!=T1，未获得锁。

第二种情况走循环走setnx逻辑

C1客户端获取锁，并且处理完后，DEL掉锁，在DEL锁之前。C2通过SETNX向foo.lock设置时间戳T0 发现有客户端获取锁，进入GET操作。

C2 向foo.lock发送GET命令，获取返回值T1(nil)。

C2 循环，进入下一次SETNX逻辑

两种逻辑貌似都是OK，但是从逻辑处理上来说，第一种情况存在问题。当GET返回nil表示，锁是被删除的，而不是超时，应该走SETNX逻辑加锁。走第一种情况的问题是，正常的加锁逻辑应该走SETNX，而现在当锁被解除后，走的是GETST，如果判断条件不当，就会引起死锁，很悲催，我在做的时候就碰到了，具体怎么碰到的看下面的问题

GETSET返回nil时应该怎么处理？

C1和C2客户端调用GET接口，C1返回T1，此时C3网络情况更好，快速进入获取锁，并执行DEL删除锁，C2返回T2(nil)，C1和C2都进入超时处理逻辑。

C1 向foo.lock发送GETSET命令，获取返回值T11(nil)。

C1 比对C1和C11发现两者不同，处理逻辑认为未获取锁。

C2 向foo.lock发送GETSET命令，获取返回值T22(C1写入的时间戳)。

C2 比对C2和C22发现两者不同，处理逻辑认为未获取锁。

此时C1和C2都认为未获取锁，其实C1是已经获取锁了，但是他的处理逻辑没有考虑GETSET返回nil的情况，只是单纯的用GET和GETSET值就行对比，至于为什么会出现这种情况？一种是多客户端时，每个客户端连接Redis的后，发出的命令并不是连续的，导致从单客户端看到的好像连续的命令，到Redis server后，这两条命令之间可能已经插入大量的其他客户端发出的命令，比如DEL,SETNX等。第二种情况，多客户端之间时间不同步，或者不是严格意义的同步。

时间戳的问题

我们看到foo.lock的value值为时间戳，所以要在多客户端情况下，保证锁有效，一定要同步各服务器的时间，如果各服务器间，时间有差异。时间不一致的客户端，在判断锁超时，就会出现偏差，从而产生竞争条件。

锁的超时与否，严格依赖时间戳，时间戳本身也是有精度限制，假如我们的时间精度为秒，从加锁到执行操作再到解锁，一般操作肯定都能在一秒内完成。这样的话，我们上面的CASE，就很容易出现。所以，最好把时间精度提升到毫秒级。这样的话，可以保证毫秒级别的锁是安全的。

分布式锁的问题

1：必要的超时机制：获取锁的客户端一旦崩溃，一定要有过期机制，否则其他客户端都降无法获取锁，造成死锁问题。

2：分布式锁，多客户端的时间戳不能保证严格意义的一致性，所以在某些特定因素下，有可能存在锁串的情况。要适度的机制，可以承受小概率的事件产生。

3：只对关键处理节点加锁，良好的习惯是，把相关的资源准备好，比如连接数据库后，调用加锁机制获取锁，直接进行操作，然后释放，尽量减少持有锁的时间。

4：在持有锁期间要不要CHECK锁，如果需要严格依赖锁的状态，最好在关键步骤中做锁的CHECK检查机制，但是根据我们的测试发现，在大并发时，每一次CHECK锁操作，都要消耗掉几个毫秒，而我们的整个持锁处理逻辑才不到10毫秒，玩客没有选择做锁的检查。

5：sleep学问，为了减少对Redis的压力，获取锁尝试时，循环之间一定要做sleep操作。但是sleep时间是多少是门学问。需要根据自己的Redis的QPS，加上持锁处理时间等进行合理计算。

6：至于为什么不使用Redis的muti，expire，watch等机制，可以查一参考资料，找下原因。

锁测试数据

未使用sleep

第一种，锁重试时未做sleep。单次请求，加锁，执行，解锁时间 



可以看到加锁和解锁时间都很快，当我们使用

ab -n1000 -c100 'http://sandbox6.wanke.etao.com/test/test_sequence.PHP?tbpm=t'

AB 并发100累计1000次请求，对这个方法进行压测时。 



我们会发现，获取锁的时间变成，同时持有锁后，执行时间也变成，而delete锁的时间，将近10ms时间，为什么会这样？

1：持有锁后，我们的执行逻辑中包含了再次调用Redis操作，在大并发情况下，Redis执行明显变慢。

2：锁的删除时间变长，从之前的0.2ms，变成9.8ms，性能下降近50倍。

在这种情况下，我们压测的QPS为49，最终发现QPS和压测总量有关，当我们并发100总共100次请求时，QPS得到110多。当我们使用sleep时

使用Sleep时

单次执行请求时



我们看到，和不使用sleep机制时，性能相当。当时用相同的压测条件进行压缩时 



获取锁的时间明显变长，而锁的释放时间明显变短，仅是不采用sleep机制的一半。当然执行时间变成就是因为，我们在执行过程中，重新创建数据库连接，导致时间变长的。同时我们可以对比下Redis的命令执行压力情况 



上图中细高部分是为未采用sleep机制的时的压测图，矮胖部分为采用sleep机制的压测图，通上图看到压力减少50%左右，当然，sleep这种方式还有个缺点QPS下降明显，在我们的压测条件下，仅为35，并且有部分请求出现超时情况。不过综合各种情况后，我们还是决定采用sleep机制，主要是为了防止在大并发情况下把Redis压垮，很不行，我们之前碰到过，所以肯定会采用sleep机制。

参考资料
http://www.worlduc.com/FileSystem/18/2518/590664/9f63555e6079482f831c8ab1dcb8c19c.pdf http://redis.io/commands/setnx http://www.blogjava.net/caojianhua/archive/2013/01/28/394847.html
--- end ---