为什么使用分布式锁，redis分布式锁，zookeeper分布式锁

爱被打了一巴掌 2022-12-06 15:07 339阅读 0赞

### 这里写目录标题 ###

*  业务场景
 *   *  无锁情况下下单
     *  单进程jvm加锁
     *  分布式无锁
 *  分布式锁
 *   *  redis分布式锁
     *  基于zookeeper实现分布式锁

# 业务场景 #

在实际开发过程中，会遇到下单操作，这里涉及到库存问题。

## 无锁情况下下单 ##

假如现在有一个场景，系统中有一个用户下单接口，但在用户下订单前一定要去检查库存，确保库存足够了才给用户下单，假如目前是单台服务器。由于系统有一定的并发，所以会预先将商品的库存保存在 Redis 中，用户下单的时候会更新 Redis 的库存。

1.  用户下单
2.  检查库存
3.  更新数据库数量
4.  锁定库存
5.  继续执行后续代码  
    ![无锁状态下下单][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x2b2VsaWZl_size_16_color_FFFFFF_t_70_pic_center]

## 单进程jvm加锁 ##

现在有两个线程【用户】访问，线程1【用户1】和线程2【用户2】，假如在不加锁的情况下，当某个时刻，Redis和mysql的库存为1。线程1执行到第3步后，还没执行到第四步，但线程2已执行到第2步。此时mysql的数据库为0，Redis的数据库为1。线程2可以执行第3步，执行完后，mysql数据库的库存为-1。因而，这里就出现了并发带来第二超卖问题，针对这个问题，线程1需要锁定2、3、4操作，然后线程2才能执行第2步操作。这里可以使用synchronized和ReentrantLock加锁，如图所示：  
![加锁下单][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x2b2VsaWZl_size_16_color_FFFFFF_t_70_pic_center 1]

## 分布式无锁 ##

在项目的初期很好，但随着用户量的增加，一台服务器的Tomcat无法支撑起很大的并发量，毕竟每个Tomcat最多并发量是500，也很容易出现单点故障。于是，需要多台服务器，来支撑较高的并发量，于是，使用nginx做负载均衡，采用轮训的机制，并采用epoll进行多路复用， 架构如下：

![分布式无锁][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x2b2VsaWZl_size_16_color_FFFFFF_t_70_pic_center 2]

用户1和用户2被负载到服务器1，在同一个JVM当中，lock加锁有效，同理，用户3和用户4的lock也成立。假如Redis和mysql存储的库存都是1，用户1执行完第3步后，mysql数据库库存为0，此时，用户3才执行完第二步，从Redis中取出的数据为1，再执行第3步，此时mysql的数据库库存为-1，这就减了两次，造成了超卖的脏数据，因而，这里需要加分布式锁，锁住当前库存。

分布式锁的思路：在整个系统提供一个全局、唯一的获取锁的“变量”，然后每个系统在需要加锁时，都去问这个“变量”拿到一把锁，这样不同的系统拿到的就可以认为是同一把锁。

# 分布式锁 #

能够实现分布式锁的两种途径

1.  基于 Redis 实现分布式锁
2.  基于zookeeper实现分布式锁

## redis分布式锁 ##

基于 Redis原生 API 来实现，容易造成锁失效。假如线程1设置超时时间为30s,但在30秒之内还未完成业务处理，线程2便获取了锁，我们需要额外去维护这个过期时间。因而，我们常使用开源框架【Redission】，

Redisson 所有指令都通过 Lua 原子性的脚本执行，它提供简单易用的API，如下加锁代码：

Config config = new Config();
    config.useClusterServers()
    		.addNodeAddress("redis://192.168.31.101:7001")
    		.addNodeAddress("redis://192.168.31.101:7002")
    		.addNodeAddress("redis://192.168.31.101:7003")
    		.addNodeAddress("redis://192.168.31.102:7001")
    		.addNodeAddress("redis://192.168.31.102:7002")
    		.addNodeAddress("redis://192.168.31.102:7003");
    RedissonClient redisson = Redisson.create(config);
    RLock lock = redisson.getLock("anyLock");
    lock.lock();
    lock.unlock();

Redisson 设置一个Key 的默认过期时间为30s，它会在你获取锁后，使用 Watchdog 【线程池，定时执行】，也就是看门狗，每隔10s帮你把Key的超时时间设为 30s。

## 基于zookeeper实现分布式锁 ##

Zookeeper是分布式协调组件，它有有序节点、临时节点、事件监听的特性，因而，可以根据这些特性创建分布式锁：

1.  使用 zookeeper的临时节点和有序节点，每个线程获取锁就是在zookeeper创建一个临时有序的节点，比如在 /lock/ 目录下。
2.  创建节点成功后，获取 /lock目录下的所有临时节点，再判断当前线程创建的节点是否是所有的节点的序号最小的节点。
3.  如果当前线程创建的节点是所有节点序号最小的节点，则认为获取锁成功。
4.  如果当前线程创建的节点不是所有节点序号最小的节点，则对节点序号的前一个节点添加一个事件监听。比如当前线程获取到的节点序号为 /lock/003，然后所有的节点列表为\[/lock/001，/lock/002，/lock/003\]，则对 /lock/002这个节点添加一事件监听器.
5.  如果锁释放了，会唤醒下一个序号的节点，然后重新执行第 3 步，判断是否自己的节点序号是最小。比如 /lock/001释放了，/lock/002监听到时间，此时节点集合为\[/lock/002，/lock/003\]，则 /lock/002为最小序号节点，获取到锁。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x2b2VsaWZl_size_16_color_FFFFFF_t_70_pic_center 3]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x2b2VsaWZl_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/1e6137f92c0740a0b146d983f5876777.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x2b2VsaWZl_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/6265d6b3e15c444d8c318f5c06fb60d8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x2b2VsaWZl_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/2036bc8ac2fc490eaa416e2eeff1a42e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x2b2VsaWZl_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221123/2d2d42bca2f94ad391a0484ecf9bc3fb.png