2 分布式锁实现方案

2.1 数据库实现分布式锁

基于数据库实现分布式锁的核心思想是：在数据库中创建一个表，表中包含方法名等字段，并在方法名字段上创建唯一索引。想要执行某个方法，首先需要将这个方法名插入表中，成功插入则获取锁，执行完成后删除对应的行数据释放锁。此种方式就是建立在数据库唯一索引的特性基础上的。

表结构如下：

具体实现过程如下（在前面lock-test工程基础上进行改造）：

第一步：在pom.xml中导入maven坐标

<dependency>
    <groupId>com.baomidou</groupId>
    <artifactId>mybatis-plus-boot-starter</artifactId>
    <version>3.2.0</version>
</dependency>
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
</dependency>

第二步：配置文件application.yml中配置mybatis-plus相关配置

server:
  port: 8002
spring:
  redis:
    host: 68.79.63.42
    port: 26379
    password: itheima123
  application:
    name: lockTest
  datasource:
      driver-class-name: com.mysql.cj.jdbc.Driver
      url: jdbc:mysql://127.0.0.1:3306/dlock
      username: root
      password: root
mybatis-plus:
  configuration:
    map-underscore-to-camel-case: false
    auto-mapping-behavior: full
    #log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
  mapper-locations: classpath*:mapper/**/*Mapper.xml

第三步：创建实体类

package com.itheima.entity;
import com.baomidou.mybatisplus.annotation.TableName;
import java.io.Serializable;
@TableName("mylock")
public class MyLock implements Serializable {
    private int id;
    private String methodName;
    public int getId() {
        return id;
    }
    public void setId(int id) {
        this.id = id;
    }
    public String getMethodName() {
        return methodName;
    }
    public void setMethodName(String methodName) {
        this.methodName = methodName;
    }
}

第四步：创建Mapper接口

package com.itheima.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.itheima.entity.MyLock;
import org.apache.ibatis.annotations.Mapper;
@Mapper
public interface MyLockMapper extends BaseMapper<MyLock> {
    public void deleteByMethodName(String methodName);
}

第五步：在resources/mapper目录下创建Mapper映射文件MyLockMapper.xml

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.itheima.mapper.MyLockMapper">
    <delete id="deleteByMethodName" parameterType="string">
        delete from mylock where methodName = #{value}
    </delete>
</mapper>

第六步：改造StockController

@Autowired
private MyLockMapper myLockMapper;
@GetMapping("/stock")
public String stock(){
    MyLock entity = new MyLock();
    entity.setMethodName("stock");
    try {
        //插入数据，如果不抛异常则表示插入成功，即获得锁
        myLockMapper.insert(entity);
        int stock = Integer.parseInt(redisTemplate.opsForValue().get("stock"));
        if(stock > 0){
            stock --;
            redisTemplate.opsForValue().set("stock",stock+"");
            System.out.println("库存扣减成功，剩余库存：" + stock);
        }else {
            System.out.println("库存不足！！！");
        }
        //释放锁
        myLockMapper.deleteByMethodName("stock");
    }catch (Exception ex){
        System.out.println("没有获取锁，不能执行减库存操作！！！");
    }
    return "OK";
}

通过观察控制台输出可以看到，使用此种方式已经解决了线程并发问题。

注意，虽然使用数据库方式可以实现分布式锁，但是这种实现方式还存在如下一些问题：

1、因为是基于数据库实现的，数据库的可用性和性能将直接影响分布式锁的可用性及性能，所以，数据库需要双机部署、数据同步、主备切换；

2、不具备可重入的特性，因为同一个线程在释放锁之前，行数据一直存在，无法再次成功插入数据，所以，需要在表中新增一列，用于记录当前获取到锁的机器和线程信息，在再次获取锁的时候，先查询表中机器和线程信息是否和当前机器和线程相同，若相同则直接获取锁；

3、没有锁失效机制，因为有可能出现成功插入数据后，服务器宕机了，对应的数据没有被删除，当服务恢复后一直获取不到锁，所以，需要在表中新增一列，用于记录失效时间，并且需要有定时任务清除这些失效的数据；

4、不具备阻塞锁特性，获取不到锁直接返回失败，所以需要优化获取逻辑，循环多次去获取。

5、在实施的过程中会遇到各种不同的问题，为了解决这些问题，实现方式将会越来越复杂；依赖数据库需要一定的资源开销，性能问题需要考虑。

2.2 ZooKeeper实现分布式锁

ZooKeeper实现分布式锁

2.3 Redis实现分布式锁

redis实现分布式锁比较简单，就是调用redis的set命令设置值，能够设置成功则表示加锁成功，即获得锁，通过调用del命令来删除设置的键值，即释放锁。

2.3.1 版本一

加锁命令：set lock_key lock_value NX

解锁命令：del lock_key

Java程序：

@GetMapping("/stock")
public String stock() {
    try {
        //尝试加锁
        Boolean locked = redisTemplate.opsForValue().setIfAbsent("mylock", "mylock");
        if(locked){//加锁成功
            int stock = Integer.parseInt(redisTemplate.opsForValue().get("stock"));
            if(stock > 0){
                stock --;
                redisTemplate.opsForValue().set("stock",stock+"");
                System.out.println("库存扣减成功，剩余库存：" + stock);
            }else {
                System.out.println("库存不足！！！");
            }
            //释放锁
            redisTemplate.delete("mylock");
        }else{
            System.out.println("没有获取锁，不能执行减库存操作！！！");
        }
    }catch (Exception ex){
        System.out.println("出现异常！！！");
    }
    return "OK";
}

2.3.2 版本二

上面版本一的实现中存在一个问题，就是当某个线程获得锁后程序挂掉，此时还没来得及释放锁，这样后面所有的线程都无法获得锁了。为了解决这个问题可以在加锁时设置一个过期时间防止死锁。

加锁命令：set lock_key lock_value NX PX 5000

解锁命令：del lock_key

Java程序：

@GetMapping("/stock")
public String stock() {
    try {
        //尝试加锁
        Boolean locked = redisTemplate.opsForValue().setIfAbsent("mylock", "mylock",5000,TimeUnit.MILLISECONDS);
        if(locked){//加锁成功
            int stock = Integer.parseInt(redisTemplate.opsForValue().get("stock"));
            if(stock > 0){
                stock --;
                redisTemplate.opsForValue().set("stock",stock+"");
                System.out.println("库存扣减成功，剩余库存：" + stock);
            }else {
                System.out.println("库存不足！！！");
            }
            //释放锁
            redisTemplate.delete("mylock");
        }else{
            System.out.println("没有获取锁，不能执行减库存操作！！！");
        }
    }catch (Exception ex){
        System.out.println("出现异常！！！");
    }
    return "OK";
}

2.3.3 版本三

针对前面版本二还有一点需要优化，就是加锁和解锁必须是同一个客户端，所以在加锁时可以设置当前线程id，在释放锁时判断是否为当前线程加的锁，如果是再释放锁即可。

Java程序：

@GetMapping("/stock")
public String stock() {
    try {
        String threadId = Thread.currentThread().getId()+"";
        //尝试加锁
        Boolean locked = redisTemplate.opsForValue().setIfAbsent("mylock",threadId,5000,TimeUnit.MILLISECONDS);
        if(locked){//加锁成功
            int stock = Integer.parseInt(redisTemplate.opsForValue().get("stock"));
            if(stock > 0){
                stock --;
                redisTemplate.opsForValue().set("stock",stock+"");
                System.out.println("库存扣减成功，剩余库存：" + stock);
            }else {
                System.out.println("库存不足！！！");
            }
            String myValue = redisTemplate.opsForValue().get("mylock");
            if(threadId.equals(myValue)){
                //释放锁
                redisTemplate.delete("mylock");
            }
        }else{
            System.out.println("没有获取锁，不能执行减库存操作！！！");
        }
    }catch (Exception ex){
        System.out.println("出现异常！！！");
    }
    return "OK";
}

3. Redisson

3.1 Redisson介绍

Redisson是架设在Redis基础上的一个Java驻内存数据网格（In-Memory Data Grid）。充分的利用了Redis键值数据库提供的一系列优势，基于Java实用工具包中常用接口，为使用者提供了一系列具有分布式特性的常用工具类。使得原本作为协调单机多线程并发程序的工具包获得了协调分布式多机多线程并发系统的能力，大大降低了设计和研发大规模分布式系统的难度。同时结合各富特色的分布式服务，更进一步简化了分布式环境中程序相互之间的协作。

Redisson已经内置提供了基于Redis的分布式锁实现，此种方式是我们推荐的分布式锁使用方式。

Redisson的maven坐标如下：

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

3.2 Redisson分布式锁使用方式

第一步：在pom.xml中导入redisson的maven坐标

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

第二步：编写配置类

package com.itheima.config;
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RedissonConfig {
    @Value("${spring.redis.host}")
    private String host;
    @Value("${spring.redis.port}")
    private String port;
    @Value("${spring.redis.password}")
    private String password;
    @Bean
    public RedissonClient redissonClient(){
        Config config = new Config();
        config.useSingleServer().setAddress("redis://" + host + ":" + port);
        config.useSingleServer().setPassword(password);
        final RedissonClient client = Redisson.create(config);
        return client;
    }
}

第三步：改造Controller

@Autowired
private RedissonClient redissonClient;
@GetMapping("/stock")
public String stock() {
    //获得分布式锁对象，注意，此时还没有加锁成功
    RLock lock = redissonClient.getLock("mylock");
    try {
        //尝试加锁，如果加锁成功则后续程序继续执行，如果加锁不成功则阻塞等待
        lock.lock(5000,TimeUnit.MILLISECONDS);
        int stock = Integer.parseInt(redisTemplate.opsForValue().get("stock"));
        if(stock > 0){
            stock --;
            redisTemplate.opsForValue().set("stock",stock+"");
            System.out.println("库存扣减成功，剩余库存：" + stock);
        }else {
            System.out.println("库存不足！！！");
        }
    }catch (Exception ex){
        System.out.println("出现异常！！！");
    }finally {
        //解锁
        lock.unlock();
    }
    return "OK";
}

3.3 Lua脚本

3.3.1 Lua简介

Lua 是一种轻量小巧的脚本语言，用标准C语言编写并以源代码形式开放，其设计目的是为了嵌入应用程序中，从而为应用程序提供灵活的扩展和定制功能。

从Redis2.6.0版本开始提供了EVAL 和 EVALSHA 命令，这两个命令可以执行 Lua 脚本。

3.3.2 Redis中使用Lua的好处

Redis中使用Lua的好处：

减少网络开销。可以将多个请求通过脚本的形式一次发送，减少网络时延

原子操作。redis会将整个脚本作为一个整体执行，中间不会被其他命令插入。因此在编写脚本的过程中无需担心会出现竞态条件，无需使用事务

复用。客户端发送的脚本会永久存在redis中，这样，其他客户端可以复用这一脚本而不需要使用代码完成相同的逻辑

3.3.3 如何在Redis中使用Lua

在redis中使用Lua脚本主要有三个命令

eval

evalsha

script load

eval用来直接执行lua脚本，使用方式如下：

EVAL script numkeys key [key ...] arg [arg ...]

key代表要操作的redis key

arg可以传自定义的参数

numkeys用来确定key有几个

script就是你写的lua脚本

lua脚本里面使用KEYS[1]和ARGV[1]来获取传递的第一个key和第一个arg，后面以此类推

举例：

eval "return redis.call('set',KEYS[1],ARGV[1])" 1 city beijing
eval "return redis.call('set','name','xiaoming')" 0
eval "return redis.call('del',KEYS[1])" 1 city
eval "return redis.call('get',KEYS[1])" 1 name
eval "if (redis.call('exists', KEYS[1]) == 0) then redis.call('set', KEYS[2], ARGV[1]); redis.call('expire', KEYS[2], ARGV[2]);return nil;end;" 2 citys city beijing 5000

在用eval命令的时候，可以注意到每次都要把执行的脚本发送过去，这样势必会有一定的网络开销，所以redis对lua脚本做了缓存，通过script load 和 evalsha实现：

SCRIPT LOAD script
EVALSHA sha1 numkeys key [key ...] arg [arg ...]

举例：

script load "return redis.call('get',KEYS[1]);"
evalsha 0e11c9f252fd76115c38403ce6095872b8c70580 1 name

4 zk锁和redis锁比较

4.1 setnx + lua脚本

优点：redis基于[内存]，读写性能很高，因此基于redis的分布式锁效率比较高

缺点：分布式环境下可能会有节点数据同步问题，可靠性有一定的影响。比如现在有一个3主3丛的Redis集群，客户端发生的命令写入了机器1的master 节点，数据正准备主丛同步的时候，master 结点挂了，slave 结点没有接收到最新的数据，此时 slave结点竞选为master, 导致之前加的分布式锁失效。

4.2 Redission

优点：解决了Redis集群的同步可用性问题

缺点：网上是说：发布时间短，稳定性和可靠性有待验证。个人觉得，目前市面上已经稳定了，算是比较成熟的比较完善的分布式锁了。

4.3 Redis分布式锁的缺点

如果你对某个redis master实例，写入了myLock这种锁key的value，此时会异步复制给对应的master slave实例。

但是这个过程中一旦发生redis master宕机，主备切换，redis slave变为了redis master。接着就会导致，客户端2来尝试加锁的时候，在新的redis master上完成了加锁，而客户端1也以为自己成功加了锁。

此时就会导致多个客户端对一个分布式锁完成了加锁。这时系统在业务语义上一定会出现问题，导致各种脏数据的产生。

所以这个就是redis cluster，或者是redis master-slave架构的主从异步复制导致的redis分布式锁的最大缺陷：*在redis master实例宕机的时候，可能导致多个客户端同时完成加锁。*

4.4 ZK分布式锁

优点：不存在redis的数据同步（zookeeper是同步完以后才返回）、主从切换（zookeeper主从切换的过程中服务是不可用的）的问题，可靠性很高

缺点：保证了可靠性的同时牺牲了一部分效率（但是依然很高）。性能不如redis。

注:实际开发过程中，可以 curator 工具包封装的API帮助我们实现分布式锁。

实现分布式锁的解决方案 2

2 分布式锁实现方案

2.1 数据库实现分布式锁

2.2 ZooKeeper实现分布式锁

2.3 Redis实现分布式锁

2.3.1 版本一

2.3.2 版本二

2.3.3 版本三

3. Redisson

3.1 Redisson介绍

3.2 Redisson分布式锁使用方式

3.3 Lua脚本

3.3.1 Lua简介

3.3.2 Redis中使用Lua的好处

3.3.3 如何在Redis中使用Lua

4 zk锁和redis锁比较

4.1 setnx + lua脚本

4.2 Redission

4.3 Redis分布式锁的缺点

4.4 ZK分布式锁

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

实现分布式锁的解决方案 2

2 分布式锁实现方案

2.1 数据库实现分布式锁

2.2 ZooKeeper实现分布式锁

2.3 Redis实现分布式锁

2.3.1 版本一

2.3.2 版本二

2.3.3 版本三

3. Redisson

3.1 Redisson介绍

3.2 Redisson分布式锁使用方式

3.3 Lua脚本

3.3.1 Lua简介

3.3.2 Redis中使用Lua的好处

3.3.3 如何在Redis中使用Lua

4 zk锁和redis锁比较

4.1 setnx + lua脚本

4.2 Redission

4.3 Redis分布式锁的缺点

4.4 ZK分布式锁

热门文章

最新文章

相关课程

相关电子书

相关实验场景