分布式锁实现：数据库、redis、zookeeper、memcache

分布式锁实现：数据库、redis、zookeeper、memcache

二、基于 Redis 实现分布式锁

2.1、通过setNx和getSet来实现

这是现在网上大部分版本的实现方式，笔者之前项目里面用到分布式锁也是通过这样的方式实现 public boolean lock(Jedis jedis, String lockName, Integer expire) { // 返回是否设置成功 // setNx加锁 px long now = System.currentTimeMillis(); boolean result = jedis.setnx(lockName, String.valueOf(now + expire * 1000)) == 1; if (!result) { // 防止死锁的容错 String timestamp = jedis.get(lockName); if (timestamp != null && Long.parseLong(timestamp) < now) { // 不通过del方法来删除锁。而是通过同步的getSet String oldValue = jedis.getSet(lockName, String.valueOf(now + expire)); if (oldValue != null && oldValue.equals(timestamp)) { result = true; jedis.expire(lockName, expire); } } } if (result) { jedis.expire(lockName, expire); } return result; } 代码分析：通过setNx命令保证操作的原子性，获取到锁，并且把过期时间设置到value里面。通过expire方法设置过期时间，如果设置过期时间失败的话，再通过value的时间戳来和当前时间戳比较，防止出现死锁。通过getSet命令在发现锁过期未被释放的情况下，避免删除了在这个过程中有可能被其余的线程获取到了锁存在问题防止死锁的解决方案是通过系统当前时间决定的，不过线上服务器系统时间一般来说都是一致的，这个不算是严重的问题锁过期的时候可能会有多个线程执行getSet命令，在竞争的情况下，会修改value的时间戳，理论上来说会有误差锁无法具备客户端标识，在解锁的时候可能被其余的客户端删除同一个key虽然有小问题，不过大体上来说这种分布式锁的实现方案基本上是符合要求的，能够做到锁的互斥和避免死锁2.2、 通过Redis高版本的原子命令jedis的set命令可以自带复杂参数，通过这些参数可以实现原子的分布式锁命令jedis.set(lockName, "", "NX", "PX", expireTime); 合并普通的setNx()和expire()操作，使其具有原子性。说明：redis的set命令可以携带复杂参数，第一个是锁的key，第二个是value，可以存放获取锁的客户端ID，通过这个校验是否当前客户端获取到了锁，第三个参数取值NX/XX,第四个参数 EX|PX，第五个就是时间NX:如果不存在就设置这个key，XX:如果存在就设置这个keyEX:如果为EX时，第5个参数的单位为秒，如果为PX时:第5个参数的单位为毫秒这个命令实质上就是把我们之前的setNx和expire命令合并成一个原子操作命令，不需要我们考虑set失败或者expire失败的情况再看看其源码：redis.clients.jedis.jedis.java源码片段如下：

/** * Set the string value as value of the key. The string can't be longer than 1073741824 bytes (1 * GB). * @param key * @param value * @param nxxx NX|XX, NX -- Only set the key if it does not already exist. XX -- Only set the key * if it already exist. * @param expx EX|PX, expire time units: EX = seconds; PX = milliseconds * @param time expire time in the units of expx * @return Status code reply */ public String set(final String key, final String value, final String nxxx, final String expx, final long time) { checkIsInMultiOrPipeline(); client.set(key, value, nxxx, expx, time); return client.getStatusCodeReply(); }

加锁

// NX 表示只有key不存的时候，才会成功，存在就失败，PX 30000 表示30秒后锁自动释放

SET mylock 随机值 NX PX 30000

1、NX:表示只有key不存在的时候，才能写成功

2、PX：30秒表示过期失效

3、随机值：用于删除时，自己只能删除自己创建的lock，根据随机数判断，别人不知道你写入的随机数。为啥要用随机值呢？因为如果某个客户端获取到了锁，但是阻塞了很长时间才执行完，此时可能已经自动释放锁了，此时可能别的客户端已经获取到了这个锁，要是你这个时候直接删除key的话就会删除本来不属于你自己的锁，所以得用随机值加上面的lua脚本来释放锁。

4、为什么删除时要用lua脚本，因为value的比较和删除动作是2个操作，为了性能及事务完整性

通过set(key,value,NX,EX,timeout)方法，我们就可以轻松实现分布式锁。值得注意的是这里的value作为客户端锁的唯一标识，不能重复。

public boolean lock1(KeyPrefix prefix, String key, String value, Long lockExpireTimeOut, Long lockWaitTimeOut) { Jedis jedis = null; try { jedis = jedisPool.getResource(); String realKey = prefix.getPrefix() + key; Long deadTimeLine = System.currentTimeMillis() + lockWaitTimeOut; for (;;) { String result = jedis.set(realKey, value, "NX", "PX", lockExpireTimeOut); if ("OK".equals(result)) { return true; } lockWaitTimeOut = deadTimeLine - System.currentTimeMillis(); if (lockWaitTimeOut <= 0L) { return false; } } } catch (Exception ex) { log.info("lock error"); } finally { returnToPool(jedis); } return false; }

解锁

我们可以使用lua脚本合并get()和del()操作，使其具有原子性。一切大功告成。

public boolean unlock1(KeyPrefix prefix, String key, String value) { Jedis jedis = null; try { jedis = jedisPool.getResource(); String realKey = prefix.getPrefix() + key; String luaScript = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end"; Object result = jedis.eval(luaScript, Collections.singletonList(realKey), Collections.singletonList(value)); if ("1".equals(result)) { return true; } } catch (Exception ex) { log.info("unlock error"); } finally { returnToPool(jedis); } return false; }

具体lua代码是这样的：

// 1获取锁// NX是指如果key不存在就成功，key存在返回false，PX可以指定过期时间SET anyLock unique_value NX PX 30000// 2释放锁：通过执行一段lua脚本// 释放锁涉及到两条指令[call("get")、call("del")]，这两条指令不是原子性的// 需要用到redis的lua脚本支持特性，redis执行lua脚本是原子性的if redis.call("get",KEYS[1]) == ARGV[1] then return redis.call("del",KEYS[1])else return 0

除了要考虑客户端要怎么实现分布式锁之外，还需要考虑redis的部署问题。

存在的问题

A: redis单点问题，如果redis挂了那么就会无法获取锁。

B: redis就算是主从架构也会出现问题，比如key还没来的级复制给从节点，主节点就挂了。

2.3、redis 有3种部署方式：

单机模式master-slave + sentinel选举模式redis cluster模式

获取当前时间戳，单位是毫秒轮流尝试在每个master节点上创建锁，过期时间设置较短，一般就几十毫秒尝试在大多数节点上建立一个锁，比如5个节点就要求是3个节点（n / 2 +1）客户端计算建立好锁的时间，如果建立锁的时间小于超时时间，就算建立成功了要是锁建立失败了，那么就依次删除这个锁只要别人建立了一把分布式锁，你就得不断轮询去尝试获取锁

另一种方式：Redisson

此外，实现Redis的分布式锁，除了自己基于redis client原生api来实现之外，还可以使用开源框架：RedissionRedisson 是一个企业级的开源 Redis Client，也提供了分布式锁的支持。我也非常推荐大家使用，为什么呢？

SET anyLock unique_value NX PX 30000

这里设置的超时时间是30s，假如我超过30s都还没有完成业务逻辑的情况下，key会过期，其他线程有可能会获取到锁。这样一来的话，第一个线程还没执行完业务逻辑，第二个线程进来了也会出现线程安全问题。所以我们还需要额外的去维护这个过期时间，太麻烦了~我们来看看redisson是怎么实现的？先感受一下使用redission的爽：

1、引用redisson包

org.redisson redisson 3.5.0

lock和unlock示例：

Config config = new Config();config.useClusterServers() .setScanInterval(2000) // 集群状态扫描间隔时间，单位是毫秒 //可以用"rediss://"来启用SSL连接 .addNodeAddress("redis://127.0.0.1:7000", "redis://127.0.0.1:7001") .addNodeAddress("redis://127.0.0.1:7002");RedissonClient redisson = Redisson.create(config);RLocklock = redisson.getLock("anyLock");lock.lock();lock.unlock();

封装的java实现类：

import org.redisson.api.RLock;import org.redisson.api.RedissonClient;import java.util.concurrent.TimeUnit;public class RedissonDistributedLocker implements DistributedLocker { private RedissonClient redissonClient; @Override public RLock lock(String lockKey) { RLock lock = redissonClient.getLock(lockKey); lock.lock(); return lock; } @Override public RLock lock(String lockKey, int leaseTime) { RLock lock = redissonClient.getLock(lockKey); lock.lock(leaseTime, TimeUnit.SECONDS); return lock; } @Override public RLock lock(String lockKey, TimeUnit unit ,int timeout) { RLock lock = redissonClient.getLock(lockKey); lock.lock(timeout, unit); return lock; } @Override public boolean tryLock(String lockKey, TimeUnit unit, int waitTime, int leaseTime) { RLock lock = redissonClient.getLock(lockKey); try { return lock.tryLock(waitTime, leaseTime, unit); } catch (InterruptedException e) { return false; } } @Override public void unlock(String lockKey) { RLock lock = redissonClient.getLock(lockKey); lock.unlock(); } @Override public void unlock(RLock lock) { lock.unlock(); } public void setRedissonClient(RedissonClient redissonClient) { this.redissonClient = redissonClient; }}

就是这么简单，我们只需要通过它的api中的lock和unlock即可完成分布式锁，他帮我们考虑了很多细节：

redisson所有指令都通过lua脚本执行，redis支持lua脚本原子性执行redisson设置一个key的默认过期时间为30s,如果某个客户端持有一个锁超过了30s怎么办？redisson中有一个watchdog的概念，翻译过来就是看门狗，它会在你获取锁之后，每隔10秒帮你把key的超时时间设为30s,这样的话，就算一直持有锁也不会出现key过期了，其他线程获取到锁的问题了。redisson的“看门狗”逻辑保证了没有死锁发生。(如果机器宕机了，看门狗也就没了。此时就不会延长key的过期时间，到了30s之后就会自动过期了，其他线程可以获取到锁)

lua脚本

if (redis.call('exists', KEYS[1]) == 0) then redis.call('hset', KEYS[1], ARGV[2], 1); redis.call('pexpire', KEYS[1], ARGV[1]); return nil; end; if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then redis.call('hincrby', KEYS[1], ARGV[2], 1); redis.call('pexpire', KEYS[1], ARGV[1]); return nil; end; return redis.call('pttl', KEYS[1]);

三、基于Zookeeper的临时有序节点

常见的分布式锁实现方案里面，除了使用redis来实现之外，使用zookeeper也可以实现分布式锁。在介绍zookeeper(下文用zk代替)实现分布式锁的机制之前，先粗略介绍一下zk是什么东西：Zookeeper是一种提供配置管理、分布式协同以及命名的中心化服务。zk的模型是这样的：zk包含一系列的节点，叫做znode，就好像文件系统一样每个znode表示一个目录，然后znode有一些特性：

有序节点：假如当前有一个父节点为/lock，我们可以在这个父节点下面创建子节点；zookeeper提供了一个可选的有序特性，例如我们可以创建子节点“/lock/node-”并且指明有序，那么zookeeper在生成子节点时会根据当前的子节点数量自动添加整数序号也就是说，如果是第一个创建的子节点，那么生成的子节点为/lock/node-0000000000，下一个节点则为/lock/node-0000000001，依次类推。临时节点：客户端可以建立一个临时节点，在会话结束或者会话超时后，zookeeper会自动删除该节点。

节点操作：

节点创建节点删除节点数据修改子节点变更

基于以上的一些zk的特性，我们很容易得出使用zk实现分布式锁的落地方案：

①基于临时节点实现

zk分布式锁，其实可以做的比较简单，就是某个节点尝试创建临时znode，此时创建成功了就获取了这个锁；这个时候别的客户端来创建锁会失败，只能注册个***监听这个锁。释放锁就是删除这个znode，一旦释放掉就会通知客户端，然后有一个等待着的客户端就可以再次重新加锁。

存在的问题

A: 需要动态的去创建、销毁节点，性能没有基于redis的高。

B: 使用Zookeeper也有可能带来并发问题，只是并不常见而已。考虑这样的情况，由于网络抖动，客户端可ZK集群的session连接断了，那么zk以为客户端挂了，就会删除临时节点，这时候其他客户端就可以获取到分布式锁了。就可能产生并发问题。这个问题不常见是因为zk有重试机制，一旦zk集群检测不到客户端的心跳，就会重试。多次重试之后还不行的话才会删除临时节点。

C: 尝试获取锁，锁销毁之后需要获取锁的线程要去竞争锁，复杂。

zookeeper集群的每个节点的数据都是一致的, 那么我们可以通过这些节点来作为锁的标志.

首先给锁设置一下API, 至少要包含：lock(锁住), unlock(解锁), isLocked(是否锁住)三个方法，然后我们可以创建一个工厂(LockFactory), 用来专门生产锁.锁的创建过程如下描述:

前提:每个锁都需要一个路径来指定(如:/lock/)

使用zk的临时节点和有序节点，每个线程获取锁就是在zk创建一个临时有序的节点，比如在/lock/目录下。创建节点成功后，获取/lock目录下的所有临时节点，再判断当前线程创建的节点是否是所有的节点的序号最小的节点如果当前线程创建的节点是所有节点序号最小的节点，则认为获取锁成功。如果当前线程创建的节点不是所有节点序号最小的节点，则对节点序号的前一个节点添加一个事件监听。比如当前线程获取到的节点序号为/lock/003,然后所有的节点列表为[/lock/001,/lock/002,/lock/003],则对/lock/002这个节点添加一个事件***。如果锁释放了，会唤醒下一个序号的节点，然后重新执行第3步，判断是否自己的节点序号是最小。

比如/lock/001释放了，/lock/002监听到时间，此时节点集合为[/lock/002,/lock/003],则/lock/002为最小序号节点，获取到锁。整个过程如下：

具体的实现思路就是这样，简易代码：

public class ZooKeeperDistributedLock implements Watcher{ private ZooKeeper zk; private String locksRoot= "/locks"; private String productId; private String waitNode; private String lockNode; private CountDownLatch latch; private CountDownLatch connectedLatch = new CountDownLatch(1);private int sessionTimeout = 30000; public ZooKeeperDistributedLock(String productId){ this.productId = productId; try { String address = "192.168.31.187:2181,192.168.31.19:2181,192.168.31.227:2181"; zk = new ZooKeeper(address, sessionTimeout, this); connectedLatch.await(); } catch (IOException e) { throw new LockException(e); } catch (KeeperException e) { throw new LockException(e); } catch (InterruptedException e) { throw new LockException(e); } } public void process(WatchedEvent event) { if(event.getState()==KeeperState.SyncConnected){ connectedLatch.countDown(); return; } if(this.latch != null) { this.latch.countDown(); } } public void acquireDistributedLock() { try { if(this.tryLock()){ return; } else{ waitForLock(waitNode, sessionTimeout); } } catch (KeeperException e) { throw new LockException(e); } catch (InterruptedException e) { throw new LockException(e); } } public boolean tryLock() { try { // 传入进去的locksRoot + “/” + productId // 假设productId代表了一个商品id，比如说1 // locksRoot = locks // /locks/10000000000，/locks/10000000001，/locks/10000000002 lockNode = zk.create(locksRoot + "/" + productId, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); // 看看刚创建的节点是不是最小的节点 // locks：10000000000，10000000001，10000000002 List locks = zk.getChildren(locksRoot, false); Collections.sort(locks); if(lockNode.equals(locksRoot+"/"+ locks.get(0))){ //如果是最小的节点,则表示取得锁 return true; } //如果不是最小的节点，找到比自己小1的节点 int previousLockIndex = -1; for(int i = 0; i < locks.size(); i++) { if(lockNode.equals(locksRoot + “/” + locks.get(i))) { previousLockIndex = i - 1; break; } } this.waitNode = locks.get(previousLockIndex); } catch (KeeperException e) { throw new LockException(e); } catch (InterruptedException e) { throw new LockException(e); } return false; } private boolean waitForLock(String waitNode, long waitTime) throws InterruptedException, KeeperException { Stat stat = zk.exists(locksRoot + "/" + waitNode, true); if(stat != null){ this.latch = new CountDownLatch(1); this.latch.await(waitTime, TimeUnit.MILLISECONDS); this.latch = null; } return true; } public void unlock() { try { // 删除/locks/10000000000节点 // 删除/locks/10000000001节点 System.out.println("unlock " + lockNode); zk.delete(lockNode,-1); lockNode = null; zk.close(); } catch (InterruptedException e) { e.printStackTrace(); } catch (KeeperException e) { e.printStackTrace(); } } public class LockException extends RuntimeException { private static final long serialVersionUID = 1L; public LockException(String e){ super(e); } public LockException(Exception e){ super(e); } }}

小结：学完了两种分布式锁的实现方案之后，本节需要讨论的是redis和zk的实现方案中各自的优缺点。对于redis的分布式锁而言，它有以下缺点：

redis分布式锁获取锁的方式简单粗暴，获取不到锁直接不断尝试获取锁，比较消耗性能。另外来说的话，redis的设计定位决定了它的数据并不是强一致性的，在某些极端情况下，可能会出现问题。锁的模型不够健壮即便使用redlock算法来实现，在某些复杂场景下，也无法保证其实现100%没有问题，关于redlock的讨论可以看How to do distributed lockingredis分布式锁，其实需要自己不断去尝试获取锁，比较消耗性能。但是另一方面使用redis实现分布式锁在很多企业中非常常见，而且大部分情况下都不会遇到所谓的“极端复杂场景”

所以使用redis作为分布式锁也不失为一种好的方案，最重要的一点是redis的性能很高，可以支撑高并发的获取、释放锁操作。对于 zk 分布式锁而言:

zookeeper天生设计定位就是分布式协调，强一致性。锁的模型健壮、简单易用、适合做分布式锁。如果获取不到锁，只需要添加一个***就可以了，不用一直轮询，性能消耗较小。但是zk也有其缺点：如果有较多的客户端频繁的申请加锁、释放锁，对于zk集群的压力会比较大。

小结：综上所述，redis和zookeeper都有其优缺点。我们在做技术选型的时候可以根据这些问题作为参考因素。

四、memcache实现的分布式互斥锁

实现原理：memcached带有add函数，利用add函数的特性即可实现分布式锁。add和set的区别在于：如果多线程并发set，则每个set都会成功，但最后存储的值以最后的set的线程为准。而add的话则相反，add会添加第一个到达的值，并返回true，后续的添加则都会返回false。利用该点即可很轻松地实现分布式锁。

if (memcache.get(key) == null) { // 3 min timeout to avoid mutex holder crash if (memcache.add(key_mutex, 3 * 60 * 1000) == true) { value = db.get(key); memcache.set(key, value); memcache.delete(key_mutex); } else { sleep(50); retry(); } }

memcache 缺点：

（1）memcached采用列入LRU置换策略，所以如果内存不够，可能导致缓存中的锁信息丢失。

（2）memcached无法持久化，一旦重启，将导致信息丢失。

五、比较

三种方案的比较

上面几种方式，哪种方式都无法做到完美。就像CAP一样，在复杂性、可靠性、性能等方面无法同时满足，所以，根据不同的应用场景选择最适合自己的才是王道。

从理解的难易程度角度（从低到高）

数据库 > 缓存 > Zookeeper

从实现的复杂性角度（从低到高）

Zookeeper >= 缓存 > 数据库

从性能角度（从高到低）

缓存 > Zookeeper >= 数据库

从可靠性角度（从高到低）

Zookeeper > 缓存 > 数据库

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。