1. 概述

分布式锁：满足分布式系统或集群模式下多进程可见并且互斥的锁。

分布式锁的核心思想就是让大家都使用同一把锁，只要大家使用的是同一把锁，那么我们就能锁住线程，不让线程进行，让程序串行执行，这就是分布式锁的核心思路

1.1. 前提条件

需要满足一些条件：

可见性：多个线程都能看到相同的结果（此处可见性区别于并发编程的内存可见性）

互斥：互斥是分布式锁的最基本的条件，使得程序串行执行

高可用：程序不易崩溃，时时刻刻都保证较高的可用性

高性能：由于加锁本身就让性能降低，所有对于分布式锁本身需要他就较高的加锁性能和释放锁性能

安全性：安全也是程序中必不可少的一环

1.2. 分类

常见的分布式锁有三种

Mysql：mysql本身就带有锁机制，但是由于mysql性能本身一般，所以采用分布式锁的情况下，其实使用mysql作为分布式锁比较少见

Redis：redis作为分布式锁是非常常见的一种使用方式，现在企业级开发中基本都使用redis或者zookeeper作为分布式锁，利用setnx这个方法，如果插入key成功，则表示获得到了锁，如果有人插入成功，其他人插入失败则表示无法获得到锁，利用这套逻辑来实现分布式锁

Zookeeper：zookeeper也是企业级开发中较好的一个实现分布式锁的方案，由于本套视频并不讲解zookeeper的原理和分布式锁的实现，所以不过多阐述

技术	MySQL	Redis	Zookeeper
互斥	利用mysql本身的互斥锁机制	利用setnx这样的互斥命令	利用节点的唯一性和有序性实现互斥
高可用	好	好	好
高性能	一般	好	一般
安全性	断开连接，自动释放锁	利用锁超时时间，到期释放	临时节点，断开连接自动释放

1.3. 核心思路

实现分布式锁时需要实现的两个基本方法：获取锁以及释放锁

//获取锁:
//互斥:确保只能有一个线程获取锁O
//添加锁，利用setnx的互斥特性
SETNX lock thread1
//非阻塞:尝试一次，成功返回true，失败返回false
//添加锁，NX是互斥、EX是设置超时时间
SET lock threadl NX EX 1
//释放锁:
//手动释放
//释放锁,删除即可
DEL key
//超时释放:获取锁时添加一个超时时间
//添加锁过期时间，避免服务宕机引起的死锁
EXPIRE lock 10

我们利用redis 的setNx 方法，当有多个线程进入时，我们就利用该方法，第一个线程进入时，redis 中就有这个key 了，返回了1，如果结果是1，则表示他抢到了锁，那么他去执行业务，然后再删除锁，退出锁逻辑，没有抢到锁的哥们，等待一定时间后重试即可

2. Redis分布式锁

资料链接：https://pan.baidu.com/s/1BnQ08PlhJ0WLoBcMMHp75A

提取码：gv96

2.1. Redis实现分布式锁的初级版本

在order-service服务中导入redis的依赖坐标

<!--redis依赖-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

在配置文件中添加redis的配置

spring:
redis:
host: 192.168.200.128
port: 6379

在order-service服务中创建一个setNx锁

package cn.itcast.order.lock;
import org.springframework.data.redis.core.StringRedisTemplate;
import java.util.concurrent.TimeUnit;
/**
 * 创建setnx锁
 */
public class RedisSetNxLock {
    //定义锁的名称
    private final String name;
    private final StringRedisTemplate stringRedisTemplate;
    public RedisSetNxLock(String name, StringRedisTemplate stringRedisTemplate) {
        this.name = name;
        this.stringRedisTemplate = stringRedisTemplate;
    }
    //定义当前key的前缀
    private static  final String KEY_PREFIX="lock:";
    /**
     * 获取锁
     *
     * @param timeoutsec key的过期时间
     * @return true表示拿到锁，false表示没有拿到锁
     */
    public boolean tryLock(Long timeoutsec) {
        //2.获取当前线程的id作为value值，保证唯一性
        long threadId = Thread.currentThread().getId();
        /**
         * 1.获取锁
         * setIfAbsent(K key, V value, long timeout, TimeUnit unit)
         *
         key –参数1表示redis中的key
         value – 参数2表示redis中存储的值
         timeout – 参数3表示key的过期时间
         unit – 参数4表示时间单位
         */
        Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name, threadId + "", timeoutsec, TimeUnit.MINUTES);
        /**
         * 这个是为了防止类型的拆箱，如果返回值为null的话，boolean类型会报错
         * 意思：如果相等于则返回true,不想等于返回false,如果flag=null的话，也是返回false;
         */
        return Boolean.TRUE.equals(flag);
    }
    /**
     * 释放锁
     *
     */
    public void unLock() {
        //通过手动释放锁
        stringRedisTemplate.delete(KEY_PREFIX+name);
    }
}

改造orderService接口中创建订单的业务代码

package cn.itcast.order.service.impl;
import cn.itcast.feign.client.AccountFeignClient;
import cn.itcast.feign.client.StorageFeginClient;
import cn.itcast.order.entity.Order;
import cn.itcast.order.lock.RedisSetNxLock;
import cn.itcast.order.mapper.OrderMapper;
import cn.itcast.order.service.OrderService;
import feign.FeignException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import javax.annotation.Resource;
/**
 *订单业务相关
 */
@Slf4j
@Service
public class OrderServiceImpl implements OrderService {
    @Resource
    private AccountFeignClient accountClient;
    @Resource
    private StorageFeginClient storageClient;
    @Resource
    private  OrderMapper orderMapper;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    /**
     * 下单业务
     * @param order
     * @return
     */
    @Override
    @Transactional
    public Long create(Order order) {
        //获取锁
        RedisSetNxLock redisSetNxLock=new RedisSetNxLock("order:"+order.getUserId(),stringRedisTemplate);
        boolean flag = redisSetNxLock.tryLock( 1200L);
        if(!flag){
            log.error("不允许重复下单");
            return 0L;
        }
        // 创建订单
        orderMapper.insert(order);
        try {
            // 扣用户余额
            accountClient.deduct(order.getUserId(), order.getMoney());
            // 扣库存
            storageClient.deduct(order.getCommodityCode(), order.getCount());
            return order.getId();
        } catch (FeignException e) {
            log.error("下单失败，原因:{}", e.contentUTF8(), e);
            throw new RuntimeException(e.contentUTF8(), e);
        }finally {
            //释放锁
            redisSetNxLock.unLock();
        }
    }
}

测试效果

我们起两个order-service服务，并且端口号分别为8881和8801
然后在创建订单的业务方法中的第一行打上断点
启动方式都为debug方式
在postman中设置两个请求

form表单的参数

userId=user202103032042012
commodityCode=100202003032041
count=2
money=200

线程1：

线程2：

测试结果：debug发现，两个线程只有一个线程能够下单成功，另外一个获取锁失败，则下单失败，并在控制台打印不可重复下单异常信息。

2.2. 基于Redis实现分布式锁误删情况

2.2.1. 误删逻辑说明：

持有锁的线程在锁的内部出现了阻塞，导致他的锁自动释放，这时其他线程，线程2来尝试获得锁，就拿到了这把锁，然后线程2在持有锁执行过程中，线程1反应过来，继续执行，而线程1执行过程中，走到了删除锁逻辑，此时就会把本应该属于线程2的锁进行删除，这就是误删别人锁的情况说明

解决方案：解决方案就是在每个线程释放锁的时候，去判断一下当前这把锁是否属于自己，如果属于自己，则不进行锁的删除，假设还是上边的情况，线程1卡顿，锁自动释放，线程2进入到锁的内部执行逻辑，此时线程1反应过来，然后删除锁，但是线程1，一看当前这把锁不是属于自己，于是不进行删除锁逻辑，当线程2走到删除锁逻辑时，如果没有卡过自动释放锁的时间点，则判断当前这把锁是属于自己的，于是删除这把锁。

2.2.2. 具体实现：

需求：修改之前的分布式锁实现，满足：在获取锁时存入线程标示（可以用UUID表示）在释放锁时先获取锁中的线程标示，判断是否与当前线程标示一致

如果一致则释放锁
如果不一致则不释放锁

核心逻辑：在存入锁时，放入自己线程的标识，在删除锁时，判断当前这把锁的标识是不是自己存入的，如果是，则进行删除，如果不是，则不进行删除。

本次，只需要更改RedisSetNxLock类中的获取锁和释放锁中的代码逻辑，无须更改下单业务逻辑

修改后代码

//定义一个uuid作为前缀
private static  final String ID_PREFIX=UUID.randomUUID().toString().replace("-","");
/**
     * 获取锁
     * @param timeoutsec key的过期时间
     * @return true表示拿到锁，false表示没有拿到锁
     */
public boolean tryLock(Long timeoutsec) {
//2.获取当前线程的id作为value值，保证唯一性
String threadId = ID_PREFIX+"-"+Thread.currentThread().getId();
/**
         * 1.获取锁
         * setIfAbsent(K key, V value, long timeout, TimeUnit unit)
         *
         key –参数1表示redis中的key
         value – 参数2表示redis中存储的值
         timeout – 参数3表示key的过期时间
         unit – 参数4表示时间单位
         */
Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name, threadId , timeoutsec, TimeUnit.MINUTES);
/**
         * 这个是为了防止类型的拆箱，如果返回值为null的话，boolean类型会报错
         * 意思：如果相等于则返回true,不想等于返回false,如果flag=null的话，也是返回false;
         */
return Boolean.TRUE.equals(flag);
}
/**
     * 释放锁
     *
     */
public void unLock() {
    //获取线程id
    String threadId = ID_PREFIX+Thread.currentThread().getId();
    //获取redis中的value=线程id
    String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);
    //如果两个线程id相等的话，则释放锁
    if(threadId.equals(id)){
        //通过手动释放锁
        stringRedisTemplate.delete(KEY_PREFIX+name);
    }
}

测试结果

debug重新启动order-service两个服务，8881和8801
然后在下单业务create方法中的获取锁和释放锁那分别打断点
在unlock释放锁的代码中的第一行打个断点
在postman中启动发起两个服务的请求
第一个线程持有锁后，手动释放锁，第二个线程此时进入到锁内部，再放行第一个线程，此时第一个线程由于锁的value值并非是自己，所以不能释放锁，也就无法删除别人的锁，此时第二个线程能够正确释放锁，通过这个案例初步说明我们解决了锁误删的问题。

2.3. 基于Redis实现分布式锁的原子性情况

2.3.1. 更为极端的误删逻辑说明：

线程1现在持有锁之后，在执行业务逻辑过程中，他正准备删除锁，而且已经走到了条件判断的过程中，比如他已经拿到了当前这把锁确实是属于他自己的，正准备删除锁，但是此时他的锁到期了，那么此时线程2进来，但是线程1他会接着往后执行，当他卡顿结束后，他直接就会执行删除锁那行代码，相当于条件判断并没有起到作用，这就是删锁时的原子性问题，之所以有这个问题，是因为线程1的拿锁，比锁，删锁，实际上并不是原子性的，我们要防止刚才的情况发生。

2.3.2. 使用Lua语言解决原子性问题

了解Lua语言

Redis提供了Lua脚本功能，在一个脚本中编写多条Redis命令，确保多条命令执行时的原子性。Lua是一种编程语言，它的基本语法大家可以参考网站：https://www.runoob.com/lua/lua-tutorial.html，这里重点介绍Redis提供的调用函数，我们可以使用lua去操作redis，又能保证他的原子性，这样就可以实现拿锁比锁删锁是一个原子性动作了，作为Java程序员这一块并不作一个简单要求，并不需要大家过于精通，只需要知道他有什么作用即可。

这里重点介绍Redis提供的调用函数，语法如下：

redis.call('命令名称', 'key', '其它参数', ...)

例如，我们要执行set name jack，则脚本是这样：

# 执行 set name jack
redis.call('set', 'name', 'jack')

例如，我们要先执行set name Rose，再执行get name，则脚本如下：

# 先执行 set name jack
redis.call('set', 'name', 'Rose')
# 再执行 get name
local name = redis.call('get', 'name')
# 返回
return name

写好脚本以后，需要用Redis命令来调用脚本，调用脚本的常见命令如下：

例如，我们要执行 redis.call('set', 'name', 'jack') 这个脚本，语法如下：

# 调用脚本
EVAL "return redis.call('set' ,"name", "jack')" 0
后面的0为脚本需要的key类型的参数个数

如果脚本中的key、value不想写死，可以作为参数传递。key类型参数会放入KEYS数组，其它参数会放入ARGV数组，在脚本中可以从KEYS和ARGV数组获取这些参数：

#调用脚本
EVAL "return redis.call('set',KEYS[1],ARGV[1])" 1 name Rose

代码实现原子性

1、在idea中可以下载lua的插件Luanalysis

2、在order-service服务中的resource下创建一个unlock.lua脚本

编写以下内容：

-- 这里的 KEYS[1] 就是锁的key，这里的ARGV[1] 就是当前线程标示
-- 获取锁中的标示，判断是否与当前线程标示一致
if (redis.call('GET', KEYS[1]) == ARGV[1]) then
    -- 一致，则删除锁
    return redis.call('DEL', KEYS[1])
end
-- 不一致，则直接返回
return 0

3、更改释放锁的代码

//定义静态化常量类
private static final DefaultRedisScript<Long> redisScript;
//执行静态快代码
static {
    redisScript=new DefaultRedisScript<>();
    redisScript.setLocation(new ClassPathResource("unlock.lua"));
    redisScript.setResultType(Long.class);
}
/**
     * 释放锁-利用lua脚本实现原子性
     *
     */
public void unLock() {
    //调用lua脚本
    stringRedisTemplate.execute(redisScript,
                                Collections.singletonList(KEY_PREFIX+name),
                                ID_PREFIX+Thread.currentThread().getId());
}

4、解析execute方法

我们的RedisTemplate中，可以利用execute方法去执行lua脚本，关系如下：

测试效果

debug重新启动order-service两个服务，8881和8801
然后在下单业务create方法中的获取锁和释放锁那分别打断点
在unlock释放锁的代码中的第一行打个断点
在postman中启动发起两个服务的请求

第一个线程进来，得到了锁，手动删除锁，模拟锁超时了，其他线程会执行lua来抢锁，当第一个线程利用lua删除锁时，lua能保证他不能删除他的锁，第二个线程删除锁时，利用lua同样可以保证不会删除别人的锁，同时还能保证原子性。

3. Redisson分布式锁

3.1. SetNx实现分布式锁存在以下问题

基于setnx实现的分布式锁存在下面的问题：

重入问题：重入问题是指获得锁的线程可以再次进入到相同的锁的代码块中，可重入锁的意义在于防止死锁，比如HashTable这样的代码中，他的方法都是使用synchronized修饰的，假如他在一个方法内，调用另一个方法，那么此时如果是不可重入的，不就死锁了吗？所以可重入锁他的主要意义是防止死锁，我们的synchronized和Lock锁都是可重入的。

不可重试：是指目前的分布式只能尝试一次，我们认为合理的情况是：当线程在获得锁失败后，他应该能再次尝试获得锁。

超时释放：我们在加锁时增加了过期时间，这样的我们可以防止死锁，但是如果卡顿的时间超长，虽然我们采用了lua表达式防止删锁的时候，误删别人的锁，但是毕竟没有锁住，有安全隐患

主从一致性： 如果Redis提供了主从集群，当我们向集群写数据时，主机需要异步的将数据同步给从机，而万一在同步过去之前，主机宕机了，就会出现死锁问题。

3.2. Redisson介绍

Redisson底层采用的是Netty 框架。支持Redis 2.8以上版本，支持Java1.6+以上版本。Redisson是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务。其中包括(BitSet, Set, Multimap, SortedSet, Map, List, Queue, BlockingQueue, Deque, BlockingDeque, Semaphore, Lock, AtomicLong, CountDownLatch, Publish / Subscribe, Bloom filter, Remote service, Spring cache, Executor service, Live Object service, Scheduler service) 。

使用Redisson可以非常方便将Java本地内存中的常用数据结构的对象搬到分布式缓存redis中。

也可以将常用的并发编程工具如：AtomicLong、CountDownLatch、Semaphore等支持分布式。

使用RScheduledExecutorService 实现分布式调度服务。

支持数据分片，将数据分片存储到不同的redis实例中。

支持分布式锁，基于Java的Lock接口实现分布式锁，方便开发。

简单来说：

一个基于Redis实现的分布式工具，有基本分布式对象和高级又抽象的分布式服务，为每个试图再造分布式轮子的程序员带来了大部分分布式问题的解决办法。

提供了使用Redis的最简单和最便捷的方法。促进使用者对Redis的关注分离，从而让使用者能够将精力更集中地放在处理业务逻辑上。

3.3. Redisson和Jedis、Lettuce区别

Jedis：Redis 官方推出的用于通过 Java 连接 Redis 客户端的一个工具包，它提供了全面的类似于 Redis 原生命令的支持，是目前使用最广的一款 java 客户端。
Lettuce：一个可扩展的线程安全的 Redis 客户端，通讯框架基于 Netty 开发，支持高级的 Redis 特性，比如哨兵，集群，管道，自动重新连接等特性。从 Spring Boot 2.x 开始， Lettuce 已取代 Jedis 成为首选 Redis 的客户端。
Redisson：一款架设在 Redis 基础上，通讯基于 Netty 的综合的、新型的中间件，是企业级开发中使用 Redis 的最佳范本。

总结下来，Jedis 把 Redis 命令封装的非常全面，Lettuce 则进一步丰富了 Api，支持 Redis 各种高级特性。但是两者并没有进一步深化，只给了你操作 Redis 数据库的工具，而 Redisson 则是基于 Redis、Lua 和 Netty 建立起了一套的分布式解决方案，比如分布式锁的实现，分布式对象的操作等等。

在实际使用过程中，Lettuce + Redisson组合使用的比较多，两者相铺相成。

3.4. 具体实现

导入依赖

<dependency>
  <groupId>org.redisson</groupId>
  <artifactId>redisson</artifactId>
  <version>3.13.6</version>
</dependency>

配置文件

@Configuration
public class RedisConfig {
    @Bean
    public RedissonClient redissonClient(){
        Config config = new Config();
        config.useSingleServer().setAddress("redis://192.168.200.196:6379");
        return Redisson.create(config);
    }
}

在下单业务中添加分布式锁

@Resource
private RedissonClient redissonClient;
/**
     * 下单业务
     * @param order
     * @return
     */
@Override
@Transactional
public Long create(Order order) {
//redisson
RLock lock = redissonClient.getLock("lock:" + order.getUserId());
boolean flag = false;
try {
    flag = lock.tryLock(10, 60, TimeUnit.SECONDS);
} catch (InterruptedException e) {
    e.printStackTrace();
}
if(!flag){//获取锁失败
    log.error("获取锁失败，不允许执行业务");
    return 0L;
}
// 创建订单
orderMapper.insert(order);
try {
    // 扣用户余额
    accountClient.deduct(order.getUserId(), order.getMoney());
    // 扣库存
    storageClient.deduct(order.getCommodityCode(), order.getCount());
} catch (FeignException e) {
    log.error("下单失败，原因:{}", e.contentUTF8(), e);
    throw new RuntimeException(e.contentUTF8(), e);
}finally {
    //释放锁
    lock.unlock();
}
return order.getId();
}

【Redis系列笔记】分布式锁