从一个简单的例子聊分布式锁
从一个简单的列子聊分布式锁
最近公司做活动的小哥在部门内分享他们应用分布式锁保证数据一致性。笔者虽然在实际开发中没有用到，但是对分布式锁一直比较感兴趣。所以今天就写个小例子研究一下分布式锁。
什么是分布式锁
分布式锁就是在分布式系统中，为解决共享资源排他性式访问而设定的锁。用于解决分布式系统中操作共享资源数据一致性问题。
应用场景
分布式锁的应用场景还是很多的。比如“秒杀”活动，大家到了某个时间点去抢“小米”手机（然后失败了）。或者某个活动，让大家去领优惠券，每个人只能拿几张不能多拿，或者一天只发放一定数量的券等等。这些活动底层基本都是用分布式锁保证，这些手机不会被“超卖”，优惠券不会被“多拿”。
分布式锁的设计原则
分布式锁需要注意以下几点：
1、互斥
确保某一个时刻只有一个线程拿到锁。这是设计分布式锁的基本要求。
2、死锁
分布式系统的产生死锁的情况比较复杂，比如当一个线程挂了，或则网络问题的解锁操作没有得到执行，将导致其他线程永远拿不到锁。因此设计时需要考虑无论线程出现什么问题，都必须释放锁。
3、性能
像“秒杀”这种活动，在瞬间会产生高并发的情况，同时访问共享资源，如果线程持有锁的时间太长，将导致大量其他线程阻塞。如何提高分布式锁的性能也是设计的关键。
4、重入
这把锁要是一把可重入锁（避免死锁）。可重入就意味着：线程可以进入任何一个它已经拥有的锁所同步着的代码块。 可重入性可以参考这边文章
分布式锁的实现方法
目前，主流的分布式锁实现主要有三种：数据库、redis、zookeeper。而这三者中应用比较广泛的是前面两个。
数据库实现分布式锁
数据库锁可以分为乐观锁和悲观锁，这里主要介绍用MySQL数据库实现。
-
乐观锁的使用
image.png
比如上面这张表，如果用state 这个字段表示用户是否已经取货。那么，但用户取货完成后我们执行下面的SQL语句更新state字段。
update user set state=2 where state=1 and id=1;
update 本身就是原子操作。但是这里存在一个问题就是"ABA"问题。
什么是“ABA问题”
如果另一个线程修改V值假设原来是A，先修改成B，再修改回成A。当前线程的CAS操作无法分辨当前V值是否发生过变化。
上面的例子一个线程把state改成2再改成1，当前线程是不知道的。上面的例子大家可能觉得没有关系，但是在其他场景中比如银行转账等就比较危险了。
解决上面“ABA问题”的方法也很简单，就是用乐观锁。而乐观锁常用的方法就是加个版本号。就是上面表中的version字段。
比如当前version=1，我们可以执行下面的SQL，更新state。
select state,version from user
update user set state=2,version=2 where state=1 and id=1 and version=1;
- 悲观锁的使用
借助数据库中自带的锁来实现分布式的锁。
下面看一个并发测试中常用的例子。这个例子用Spring boot 框架，ORM框架Mybatis，文章末尾会给出下载链接， 整篇文章会围绕这个例子展开。
@Autowired
UserService userService;
@RequestMapping(path = {"/user"})
@ResponseBody
public String index(){
for(int i=0;i<20000;i++){
User user=userDAO.selectById(1);
int age=user.getAge();
age=age+1;
user.setAge(age);
userDAO.updateAge(user);
}
打开两个浏览器分别输入http://localhost:8080/user 回车，会发现数据库中user表age字段没有加到40000。这是因为：
例如：
1、此时age=1
2、浏览器A取出age=1，紧接着浏览器B的请求也到了，也将age=1取出
3、A取出后立即加1，并将age=2存回去
4、此时B也紧跟着，也将age=2存进去了
整体上age字段是少加了。那么如何加到40000呢？看下面代码：
@Transactional(propagation = Propagation.REQUIRED,isolation = Isolation.DEFAULT,timeout=36000,rollbackFor=Exception.class)
public void addAge(){
User user= userDAO.selectByIdForUpdate(1);
int age=user.getAge();
age=age+1;
user.setAge(age);
userDAO.updateAge(user);
}
其中 selectByIdForUpdate的代码如下：
@Select({"select ", SELECT_FIELDS, " from ", TABLE_NAME, " where id=#{id} for update"})
User selectByIdForUpdate(int id);
和之前的查询不一样的是，我们在查询语句后面增加for update，数据库会在查询过程中给数据库表增加悲观锁。同时采用@Transactional开启事务。此时在user表中，id为1的 那条数据就被我们锁定了，其它的事务必须等本次事务提交之后才能执行。这样我们可以保证当前的数据不会被其它事务修改。
PS：InnoDB引擎在加锁的时候，只有通过索引进行检索的时候才会使用行级锁，否则会使用表级锁。 但是MySQL可以对查询进行优化，MySQL会根据执行计划的代价判断是否使用索引检索数据。因此如果MySQL认为全表扫描效率更高，InnoDB会选择表锁，而不是行锁。
回顾下之前讲的分布式锁的四点设计原则，对于互斥 和死锁可以满足要求，for update如果执行失败就处于阻塞状态，直到成功，如果成功就立刻返回。对于连接无法释放导致死锁问题，使用悲观锁在服务器宕机之后数据库自己把锁释放掉。
使用数据库来实现分布式锁，比较容易理解，但是性能问题是他的最大缺点，因为操作数据库是要一定开销的，而且当我们的表不是很大的时候，我们不能保证数据库一定是行级锁。
数据库乐观锁和悲观锁的总结和实践可以参照下面的链接：
http://chenzhou123520.iteye.com/blog/1863407
http://chenzhou123520.iteye.com/blog/1860954
Redis实现分布式锁
下面是着重要介绍的使用Redis实现分布式锁。
@RequestMapping("/")
public String index(){
Jedis jedis = new Jedis("redis://localhost:6379/9");
for(int i=0;i<200000;i++){
int count=Integer.parseInt(jedis.get("count"));
count=count+1;
// jedis.incr("count"); //最简单的原子操作
jedis.set("count",Integer.toString(count));
System.out.println(count);
}
jedis.close();
return "Hello Spring boot";
}
上面的代码是取redis中的count 然后对他+1，再放回到redis中，由于注释1 ,count=count+1不是原子操作，因此如果我们打开两个浏览器同时访问 localhost:8080 发现count在redis中的值并没有达到400000
image.png
其实对于这种+1 操作redis 提供了简单的incr 原子操作(见注释2)。但是是在实际的业务场景中没有这么简单。那么如果用jedis 设计分布式锁呢？
采用Jedis实现
@RequestMapping("/distribute")
public String distribute(){
RedisLock redisLock=new RedisLock("distribute");
for (int i=0;i<200000;i++){
String identifier=redisLock.tryLock(2);
if(!identifier.equals("false")){
int count=Integer.parseInt(redisLock.jedis.get("count"));
count=count+1;
redisLock.jedis.set("count",Integer.toString(count));
redisLock.unlock("distribute",identifier);
}
}
redisLock.jedis.close();
return "Hello Spring boot";
}
看一下核心代码tryLock 加锁和 unlock 解锁操作。trylock 方法利用了Redis的setnx 、ttl、expire 方法构建分布式锁。 简单介绍一下这几个命令，已经熟悉的同学可以跳过这小块。
setnx
语法 ： SETNX key value
将 key 的值设为 value ，当且仅当 key 不存在。
若给定的 key 已经存在，则 SETNX 不做任何动作。
ttl
语法：TTL KEY_NAME
当 key 不存在时，返回 -2 。 当 key 存在但没有设置剩余生存时间时，返回 -1 。 否则，以秒为单位，返回 key 的剩余生存时间。
expire
语法：EXPIRE key seconds
为给定 key 设置生存时间，当 key 过期时(生存时间为 0 )，它会被自动删除。
网上有很多资料关于如何用redis实现分布式锁。很多地方都大同小异。
这是我写的第一个版本代码：
用两个浏览器访问，可以看到正确的结果。
后来发现上面代码可以进一步优化。
setnex 与 expire 因为这两条语句不是原子操作。所以还要加上ttl操作的判断。可能存在这样的情况：客户端A上一步没能设置时间就进程奔溃了，客户端B就可检测出来，并设置时间。把上面两句合并成一句，代买如下：
public String tryLock(int lockSeconds) {
long nowTime = System.currentTimeMillis();
lockValue= UUID.randomUUID().toString();
long end=nowTime+lockSeconds*1000;
while(System.currentTimeMillis()<end){
if (jedis.set(lockKey,lockValue,"NX","EX",lockSeconds)!=null) {
return lockValue;
}
long millis=1;
try {
Thread.sleep(millis);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
return "false";
}
setnx 方法 提供了设置过期时间的参数，并且保证操作原子性。并且官网也是推荐使用set ，未来在之后的版本中很有可能把 SETNX, SETEX, PSETEX都废弃掉。
再看一下解锁方法
public boolean unlock(String lockKey,String identifier) {
if(jedis.get(lockKey).equals(identifier)){ #判断是锁有没有被其他客户端修改
Transaction tx=jedis.multi();
tx.del(lockKey);
tx.exec();
return true;
}
return false;
}
因为解锁操作不能只是简单的DEL KEY，防止最后的解锁操作会误解掉其他客户端的操作。所以在加锁操作的时候就把一个唯一的UUID作为key的 value，解锁之前先判断一下是否是自己的value，然后然后再删除。用上面的代码测试出来的数据也是没有问题的。
如果大家去网上查看使用Redis实现分布式锁的方法的话，会看到代码与上面的不太一样，会用到getset方法，
getset
语法：GETSET key value
将给定 key 的值设为 value ，并返回 key 的旧值(old value)。
比如这边文章把key过期时间戳，直接设置到value了。在加锁时使用了getset，解锁方法也有点不同。如果采用把过期时间直接赋值到value里，并且不调用expire,在这用情况下不用getset会出现问题。C0超时了，还持有锁,C1/C2同时请求进入了方法里面，会出现下面的情况：
C1 DEL key
C1 SETNX key <expireTime>
C2 DEL key
C2 SETNX key <expireTime>
当Redis服务器收到这样的指令序列时，C1和C2的SETNX都同时返回了1，此时C1和C2都认为自己拿到了锁，这种情况明显是不符合预期的。所以先用GET 拿到旧的时间v1，然后用GETSET 方法设置自己的时间戳，拿到返回值v2，比较v1 和 v2是否相同，相同才说明真正拿到锁。具体分析可以看上面给的文章链接。
我们再次回顾一下设计分布式锁的四条原则：1、2 都是可以满足的，任何时候只有一个线程获得锁，同时因为设置了过期时间，可以解决锁不能释放导致死锁问题 ，但是3、4两条没有满足。性能问题，会在文末进行比较。
采用Redisson实现
Redisson是redis分布式方向落地的产品，不仅开源免费，而且内置分布式锁，分布式服务等诸多功能，是基于redis实现分布式的最佳选择。
Redisson的官方地址,可以看到Redisson,在国内已经被阿里巴巴和百度互联网公司使用。
image.png
对比Jedis，Redisson具有以下几个特征：
1、Redisson 提供了分布式Java常用数据结构，官方给出如下分布式数据结构，并且这些数据接口都是线程安全的。
image.png
2、Jedis的中的方法基本与Redis中API一一对应。Redisson 中的方法进行了比较高的抽象。
3、Jedis使用的是阻塞I/O，不支持异步。Redisson底层使用Netty作为网络通信，方法调用是异步的。
4、Redissson 与第三方框架整合较好。
总结 ：redisson实现了分布式和可扩展的java数据结构，支持的数据结构有：List, Set, Map, Queue, SortedSet, ConcureentMap, Lock, AtomicLong, CountDownLatch。并且是线程安全的，底层使用Netty 4实现网络通信。和jedis相比，功能比较简单，不支持排序，事务，管道，分区等redis特性，可以认为是jedis的补充，不能替换jedis。
回到上面的例子采用Redisson进行加锁的操作非常简单。代码如下：
@RequestMapping("/redisson")
public String redisson(){
Config config = new Config();
config.useSingleServer().setAddress("http://127.0.0.1:6379");
RedissonClient client = Redisson.create(config);
RLock lock = client.getLock("lock");
lock.lock();
for (int i=0;i<200000;i++){
count=count+1;
}
lock.unlock();
client.shutdown();
System.out.println("count value:"+count);
return "Hello Spring boot "+count;
}
性能分析
笔者用wrk,分别循环7次和2000次对采用不同方法实现的分布式锁进行比较。
首先看循环7次的情况。
image.png
从上到小分别是采用自己写的分布式锁、Redisson分布式锁、Redis的
incr方法的压测结果，发现采用原生的incr方法QPS最高。但是不知道为什么Redisson的QPS非常低。当采用循环2000次的时候结果如下：
image.png
采用
incr的方法，QPS急剧下降、Redisson加锁的QPS还是稳定在3左右，可以看到两种分布式锁的QPS是相当的。而自己写的分布式锁QPS只有可怜的0.27。
简书上有文章分析了Redisson底层实现。通过底层代码的分析，Redisson满足上面的设计原则的前三点要求，又通过wrk,压测发现性能也是比较好的。看过这篇文章我总结一下Redisson就是下面三者的结合。
Redisson=java.util.concurrent+Netty+Redis
总结
事实上还有第三种方法: 使用zookeeper实现分布式锁。笔者暂时还没有动手实践过，有兴趣的同学可以参考这篇文章，实现一下。
这篇文章分析和比较了这三种方案。
从理解的难易程度角度（从低到高）
数据库 > 缓存 > Zookeeper
从实现的复杂性角度（从低到高）
Zookeeper >= 缓存 > 数据库
从性能角度（从高到低）
缓存 > Zookeeper >= 数据库
从可靠性角度（从高到低）
Zookeeper > 缓存 > 数据库
个人感觉从易用性和性能上说还是比较推荐使用Redisson作为分布式锁的实现方法。
