高并发分布式场景下的应用---分布式锁

一. 单体架构和垂直应用架构

很久以前，还在我还年轻的时候大部分包括刚开始的淘宝，还不是亚马逊中国的卓越等网站，那时候还没有多少知道上网，大部分年轻人还停留在cs局域网对战的时候，大多国人由于不太清楚上网这个概念时候那时候国内兴起了一系列的互联网公司（活下来的大部分成了行业巨头）那时候流量很小，只需要一个基本应用，将所有的功能都部署在一起（前端和后端都放在一起），用来减少部署节点和成本。那时候，主要关注点就是对业务的增删改查工作
特点：

所有的业务代码都放在一起。
通过部署应用集群和数据库集群来提高系统的性能。

优点：

项目架构简单，开发成本低，周期短一人一马搞定。

缺点：

全部代码放在一个项目中耦合严重，对于往后发展不适合。
修改代码需要人力，还要去学习前任的业务逻辑实现，成本高。
容错率低。

垂直架构
当访问量逐渐增大，功能逐渐复杂起来，单一应用架构就显得有些捉襟见肘，由于所有的功能都写在同一个工程中，整个工程会越来越庞大越来越臃肿，所以将应用拆成互不相干的几个应用，以提升效率。

特点：

将项目以单一应用架构的方式，将一个大应用拆分成为几个互不干扰的应用。
应用于应用之间互不相干，功能和数据都会存在冗余。

优点：

通过垂直拆分，防止单体项目无限扩大。
系统间相互独立。

缺点：

项目拆分之后，项目与项目之间存在数据冗余，耦合性较大。
提高系统性能只能通过扩展集群，成本高，并且存在瓶颈。

上述其实就是多应用相结合，但是彼此之间应用无关。

二. 微服务的出现

当垂直应用越来越多，应用与应用之间的交互不可避免，这时需要将核心业务抽取出来，作为独立的服务，逐渐形成稳定的服务，使前端应用能更快速的响应多变的市场需求。
特点：

系统服务层完全独立，并且抽取成一个一个的独立服务。
微服务遵循单一原则。
使用服务中心进行服务注册与发现。
每个服务有自己的独立数据源。
微服务之间采用 RESTful 等轻量协议传输。

优点：

通过分解巨大单体式应用为多个服务方法解决了复杂性问题。
可以更加精准的制定每个服务的优化方案，提高系统可维护性。
微服务架构模式是每个微服务独立的部署，可以使每个服务独立扩展。
微服务架构采用去中心化思想，服务之间采用 RESTful 等轻量协议通信。

缺点：

微服务过多，服务治理成本高，对系统运维团队挑战大。
分布式系统开发的技术成本高（容错、分布式事务等），对团队挑战大。

三. 满足人们日益增长的物质文化需要

说完微服务，并不是一蹴而就为用而用，如果只是一个非常小的系统，或者只是一个简单的应用程序，后续不会再进行过多的提升或者是版本是直接替换性质的（比如曾经的某驾游，1.0-2.0直接就是替换了所有内容），这个一般都是适用于刚开始创业的小微公司，不会开始就直接上大型分布式架构系统，一方面考虑快速上架，后续迭代开发；另外一方面也是最重要的就是没那么多钱！—体会

那成熟的软件公司尤其是大中型企业他们本身已经经历过洗礼，比如淘宝从开始的PHP—现在的自研等。所以已经有了稳固的和强大的（资本）流量和目标。为此这类型公司早就在行业内抓住了前瞻技术，尤其是业务复杂度和内容多样性的出现，使得他们对待系统稳定/安全/数据一致性等方面都有着很强悍的经验。同样随着系统的不断扩展和内容，流量的不断飙升（甚至可以达到亿级），比如双11，双12，年中大促，各类所谓的节等等，如果只是单体或者简单的微服务架构是肯定不行的，还需要各类所谓的机制保证和相关服务的支撑–人力，硬件，运维等等。

总结一下，微服务虽然可以通过ddd（领域设计）来实现一套完整的系统，但是还需要其他配合，比如人力/相关服务等配合完成。

本次内容不谈那么大而深的概念和实现，而是我们讨论一下在这个大型分布式环境下的一个肯定会用到的内容—分布式锁的应用和实现。

四. 分布式锁

什么是分布式锁？

如果在一个分布式系统中，我们从数据库中读取一个数据，然后修改保存，这种情况很容易遇到并发问题。因为读取和更新保存不是一个原子操作，在并发时就会导致数据的不正确。说白一点就是一次操作有读又有写，且读和写必须保证是当前这一次操作的，不能被其他情况给挟持。

这种场景其实并不少见，比如电商秒杀活动，库存数量的更新就会遇到。如果是单机应用，线程锁就可以避免。如果是分布式应用，不同服务器不同容器对应的应用都是不同的jvm，所以无法通过线程来实现，这时就需要引入分布式锁来解决。—提个话题，大部分情况下想要保证分布式系统的cap，基本都依赖于中间件。

由此可见分布式锁的目的其实很简单，就是为了保证多台服务器在执行某一段代码时保证只有一台服务器执行。

为了保证分布式锁的可用性，至少要确保锁的实现要同时满足以下几点：

互斥性。在任何时刻，保证只有一个客户端持有锁。
不能出现死锁。如果在一个客户端持有锁的期间，这个客户端崩溃了，也要保证后续的其他客户端可以上锁。
保证上锁和解锁都是同一个客户端。

一般来说，实现分布式锁的方式有以下几种：

使用MySQL，基于唯一索引。
使用Redis，基于setnx命令。
使用ZooKeeper，基于临时有序节点。

五. 实现方案

第一个数据库操作

可以针对一列做唯一索引，或者做一张表，去做对应更新记录；或者做乐观锁加version.

但是这种方案是否合适？！！！！！

问题很多，首先最明显的问题就是io问题，其次就是连接耗时问题。

可是为了解决响应时间和分布式环境下的数据一致性问题。

redis实现

Redis实现分布式锁主要利用Redis的setnx命令和Redis的单线程特性（由来以久的到底多还是单的问题，其实这个问题只是这对于处理能力进行细化，比如在6.x之前所有的操作都是由一个单工作线程来实现，对于一个数据的操作包括读/计算/写入等都由该单工作线程实现；而6.x之后分出一个io的子线程，把运算放入了工作线程操作，二读和写入两个操作放入了独立的io子线程操作；那如果多个数据操作，实际上就会出现一个单工作线程，多个io子线程，构成io复用—性能更高）。setnx是SET if not exists(如果不存在，则 SET)的简写。

上述是加了过期时间的，如果不加过期时间直接用setnx key:value。

为什么要使用nx这种模式，实际为了解决一个问题，就是我们如果定义了一个值，在其没有被手动删除或者时间戳自动失效前都不能被其他线程给操作和修改。明白了吗？是不是这样就可以保证在针对于这一个操作的时候可以保证数据的原子性？！

为了保证实现的最基本要求，本案例启动了2个服务，分别是8081，8082，通过用Nginx做了网关，负载使用了轮询。模拟访问地址是Nginx网关地址。库存模拟放入redis中 stock:1000

并发测试工具采用postman自带测试（效果也就玩玩）

以及Jmeters5.3.

数据模拟采用了redis或者数据库。

方案1:
```
public String updateStock01(String pId) throws Exception {String clientId = UUID.randomUUID().toString();try{Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(pId,clientId,30, TimeUnit.SECONDS);if(result) {int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));if (stock > 0) {stock = stock - 1;stringRedisTemplate.opsForValue().set("stock", stock + "");log.info("扣除成功，目前库存还有" + stock + "件");} else {log.info("库存不足");}}}catch (Exception e){log.info(e.getMessage());}finally {if(clientId.equals(stringRedisTemplate.opsForValue().get(pId))) {stringRedisTemplate.delete(pId);}}return "success";
}
```
上述代码可以看到我们通过生成一个30秒ttl的Key -> pId value->clientId的列子。在这个30秒的时间里面，任何一个同样的Key都不可以进来修改。然后通过操纵对于库存结果的扣件后，在最终finally里面保证删除该key留给其他线程。看起来没问题，但是真没有问题吗？

如果流量不大，其实出现问题的情况基本为0.但是如果说在压测时候使用j meters通过大量线程同时进入，并保持不停压测。会发现—凉凉！

我这边操作2个相同服务去实现，如下图：

发现结果有问题了！这个问题究其原因是怎么造成的呢？

首先我们这个是2个服务，都是依赖于独立的jvm，所以有不同的Jmm,就算你做了线程加了重量级锁结果还是一样，毕竟是非原子性的操作。
其次虽然采用了set key value [EX seconds] [PX milliseconds] [NX|XX]的方式，但是如果一个线程a执行的时间较长没有来得及释放，锁就过期了，此时l另外一个线程B是可以获取到锁的。当线程A执行完成之后，释放锁，实际上就把线程B的锁释放掉了。这个时候，再来一个线程C又是可以获取到锁的，而此时如果线程B执行完释放锁实际上就是释放的线程C设置的锁。
这边只是本地机器，如果是云端/线上环境等，服务和服务之间的调度延迟，网络访问延迟等等都会造成问题。

方案1不可行！！！

那上述问题，我们如何规避呢？

可以这样想，既然代码层面直接实现不可以，但是可以通过交给redis这个所谓的单线程去处理不就行了！？

方案2:

set、del是一一映射的，不会出现把其他现成的锁del的情况。从实际情况的角度来看，即使能做到set、del一一映射，也无法保障业务的绝对安全。因为锁的过期时间始终是有界的，除非不设置过期时间或者把过期时间设置的很长，但这样做也会带来其他问题。故没有意义。要想实现相对安全的分布式锁，必须依赖key的value值。在释放锁的时候，通过value值的唯一性来保证不会勿删。

通过lua脚本来实现；

通过redission实现：

public String updateStock02(String pId) throws Exception {//增加分布式锁RLock rLock =redisson.getLock(pId);try{rLock.lock(5000,TimeUnit.MILLISECONDS);int stock  = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));if(stock>0){stock  = stock - 1;stringRedisTemplate.opsForValue().set("stock",stock+"");log.info("扣除成功，目前库存还有"+stock+"件");}else{log.info("库存不足");}}catch (Exception e){log.info(e.getMessage());}finally {rLock.unlock();}return "success";
}

按照原来还有982个，现在做1000个线程，最快执行完毕，连续循环2次，结果到0的时候，显示库存不足。

看代码：

 RLock rLock =redisson.getLock(pId);rLock.lock(5000,TimeUnit.MILLISECONDS);rLock.unlock();

眼熟吗？是否和Lock很像？！

其实这边和Lock相似的地方采用类似自旋锁的方式，做了do while循环判断是否获得锁，然后根据hash算法分配到不同的redis 哨兵主从环境（关于16384的槽以及丢失某个节点后数据问题后续可以讨论），当然单机也没有问题就是。并执行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]);"

其中hash保证通过lua脚本保证了原子性。

具体内容用了一段别人介绍的说法(不去深究，可以结合线程锁的源码来阅读)：

原引：csdn 段子猿
给大家解释一下，第一段if判断语句，就是用“exists myLock”命令判断一下，如果要加锁的那个锁key不存在的话，就进行加锁。 如何加锁呢？很简单，用下面的命令：hset myLock  3d7b5418-a86d-48c5-ae15-7fe13ef0034c:1接着会执行“pexpire myLock 30000”命令，设置myLock这个锁key的生存时间是30秒。这样加锁完成了。几个参数所代表的意思：KEYS[1]代表的是你加锁的那个key，比如说 RLock lock = redisson.getLock("myLock")，这里你自己设置了加锁的那个锁key就是“myLock”。ARGV[1]代表的就是锁key的默认生存时间，默认30秒。ARGV[2]代表的是加锁的客户端的ID，类似于这样： 3d7b5418-a86d-48c5-ae15-7fe13ef0034c:1 2.锁互斥机制在这个时候，如果客户端2来尝试加锁，执行了同样的一段lua脚本，会咋样呢？按照脚本流程，第一个if判断会执行“exists myLock”，发现myLock这个锁key已经存在了。接着第二个if判断，判断一下，myLock锁key的hash数据结构中，是否包含客户端2的ID，但是明显客户端id不一致。所以，客户端2会获取到pttl myLock返回的一个数字，这个数字代表了myLock这个锁key的剩余生存时间。比如还剩15000毫秒的生存时间， 此时客户端2会进入一个while循环，不停的尝试加锁。3.watch dog自动延期机制客户端加锁的key默认生存时间是30秒，如果超过了30秒，客户端还想一直持有这把锁，怎么办呢？只要客户端一旦加锁成功，就会启动一个watch dog看门狗，他是一个后台线程，每到过期时间1/3（默认生存时间是30s的话，就是10s）就去重新刷一次，如果客户端还持有锁key，那么就会不断的延长锁key的生存时间，如果key不存在则停止刷新。4.可重入加锁机制那如果客户端已经持有了这把锁，结果可重入的加锁会怎么样呢？比如下面这种代码：RLock lock = redisson.getLock("myLock");
lock.lock();
//do something
lock.lock();
//do something
lock.unlock();
lock.unlock();
这时我们来分析一下上面那段lua脚本。第一个if判断肯定不成立，“exists myLock”会显示锁key已经存在了。第二个if判断会成立，因为myLock的hash数据结构中包含的那个ID，就是客户端那个ID，也就是“3d7b5418-a86d-48c5-ae15-7fe13ef0034c:1” ，此时就会执行可重入加锁的逻辑，他会用： incrby myLock 3d7b5418-a86d-48c5-ae15-7fe13ef0034c:1 1通过这个命令，对客户端1的加锁次数，累加1。此时myLock数据结构变为下面这样： 大家看到了吧，那个myLock的hash数据结构中的那个客户端ID，就对应着加锁的次数5.释放锁机制如果执行lock.unlock()，就可以释放分布式锁，此时的业务逻辑其实很简单。就是每次都对myLock数据结构中的那个加锁次数减1。如果发现加锁次数是0了，说明这个客户端已经不再持有锁了，此时就会用： “del myLock”命令，从redis里删除这个key。然后呢，另外的客户端就可以尝试完成加锁了。这就是所谓的分布式锁的开源Redisson框架的实现机制。

方案3: Spring Integration

Spring Integration不需要你去关注它到底是基于什么存储技术实现的，它是面向接口编程，低耦合让你不需要关注底层实现。你要做的仅仅是做简单的选择，然后用相同的一套api即可完成分布式锁的操作。

@Override
public String updateStock03(String pId) throws Exception {//增加分布式锁Lock lock = redisLockRegistry.obtain(pId);boolean flag = false;try{flag = lock.tryLock(10,TimeUnit.SECONDS);int stock  = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));if(stock>0){stock  = stock - 1;stringRedisTemplate.opsForValue().set("stock",stock+"");log.info("购买成功，库存还有"+stock+"件");}else{log.info("库存不足");}}catch (Exception e){log.info(e.getMessage());}finally {lock.unlock();}return "success";
}

通过测试
共计100个。

测试成功。

zookeeper实现

通过zk来实现，其中zk通过定义EPHEMERAL_SEQUENTIAL临时顺序节点.比如创建一个/lock/临时有序；
1. 创建节点成功后，获取/lock目录下的所有临时节点，再判断当前线程创建的节点是否是所有的节点的序号最小的节点
2. 如果当前线程创建的节点是所有节点序号最小的节点，则认为获取锁成功。
3. 如果当前线程创建的节点不是所有节点序号最小的节点，则对节点序号的前一个节点添加一个事件监听。
  比如当前线程获取到的节点序号为/lock/003,然后所有的节点列表为[/lock/001,/lock/002,/lock/003],则对/lock/002这个节点添加一个事件监听器。
4. 如果锁被释放，会唤醒下一个序号的节点，然后重新执行第3步，判断是否自己的节点序号是最小。比如/lock/001释放了，/lock/002监听到时间，此时节点集合为[/lock/002,/lock/003],则/lock/002为最小序号节点，获取到锁。
  
  偷懒的写法—上面写法实话太麻烦了：
  
  Curator是一个zookeeper的开源客户端，也提供了分布式锁的实现：
```
 public String updateStock04(String pId) throws Exception {InterProcessMutex interProcessMutex = new InterProcessMutex(curatorFramework, "/product_" + pId);try{//加锁interProcessMutex.acquire(10,TimeUnit.SECONDS);int stock  = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));if(stock>0){stock  = stock - 1;stringRedisTemplate.opsForValue().set("stock",stock+"");log.info("购买成功，库存还有"+stock+"件");}else{log.info("------没有库存------");}}catch (Exception e){e.printStackTrace();}finally {interProcessMutex.release();}return "success";}
```

六. 总结

上述内容就是目前实现分布式锁的几种流行方式，那还有没有其他的？有！比如consul也可以实现，这边也不延展。这边要说的是如果业务如果并没有想象的那么夸张的时候，没有必要不要考虑使用第三方实现。

另外Redission实际上就是将并行的请求，转化为串行请求。这样就降低了并发的响应速度.

可以通过锁分段来实现，比如ps5昨天发布国行价格，商品只有2000台货品，暂时不考虑地区库存计算；一下黄牛们纷涌而至，这个时候2000台货如果按照串行走的话，比如每个50ms,暂时不考虑分布式事务锁或者是最终一致性的死信队列情况，2000个大概需要2000x50ms=100000ms,大概也要100秒左右时间，那如果并行呢？比如我把2000个ps5分成20份，那就是每份里面存100个，按照库存取余数（或者Hash或者就100，100的放等等）分别放置。这样每次可以定义多个锁进行加锁即可。这边注意一个问题就是因为涉及到20个分段，所以注意如果某个分段已经没有库存的时候，需要解锁进入下一个分段继续。

也可以直接在redis预设好（个人比较推崇），我可以给ps5的2000台分成不同的stock:product_ps5_xxx:01: 100;stock:product_ps5_xxx:02: 100…

最后无论你身处一个什么样的公司，最开始的工作可能都需要从最简单的做起。所以能根据自己公司业务场景，选择适合自己项目的方案。