文章/答案/技术大牛

发布

社区首页 >专栏 >【五一创作】基于mysql关系型实现分布式锁

【五一创作】基于mysql关系型实现分布式锁

一个风轻云淡

发布于 2023-10-15 02:37:31

37500

代码可运行

文章被收录于专栏：java学习javajava学习java

运行总次数：0

代码可运行

0.写在前面

在多线程高并发场景下，为了保证资源的线程安全问题，jdk为我们提供了synchronized关键字和 ReentrantLock可重入锁，但是它们只能保证一个jvm内的线程安全。在分布式集群、微服务、云原生横行的当下，如何保证不同进程、不同服务、不同机器的线程安全问题，jdk并没有给我们提供既有的解决方案。此时，我们就必须借助于相关技术手动实现了。目前主流的实现有三种方式： 1. 基于mysql关系型实现 2. 基于redis非关系型数据实现 3. 基于zookeeper实现

这篇文章主要讲解的是基于基于mysql关系型实现分布式锁

1. 从减库存聊起

库存在并发量较大情况下很容易发生超卖现象，一旦发生超卖现象，就会出现多成交了订单而发不了货的情况。

场景：商品S库存余量为5时，用户A和B同时来购买一个商品S，此时查询库存数都为5，库存充足则开始减库存：用户A：update db_stock set stock = stock - 1 where id = 1 用户B：update db_stock set stock = stock - 1 where id = 1 并发情况下，更新后的结果可能是4，而实际的最终库存量应该是3才对

1.1. 环境准备

为了模拟具体场景我们需要准备开发环境

首先需要在mysql数据库中准备一张表：

CREATE TABLE `db_stock` (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`product_code` varchar(255) DEFAULT NULL COMMENT '商品编号',
`stock_code` varchar(255) DEFAULT NULL COMMENT '仓库编号',
`count` int(11) DEFAULT NULL COMMENT '库存量',
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;

表中数据如下：

创建分布式锁demo工程：

建立以下工具目录结构：

pom依赖文件：

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>5.1.46</version>
        </dependency>
        <dependency>
            <groupId>com.baomidou</groupId>
            <artifactId>mybatis-plus-boot-starter</artifactId>
            <version>3.4.0</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.16</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

application.yml配置文件：

server:
  port: 6000
spring:
  datasource:
    driver-class-name: com.mysql.jdbc.Driver
    url: jdbc:mysql://172.16.116.100:3306/test
    username: root
    password: root

DistributedLockApplication启动类：

@SpringBootApplication
@MapperScan("com.atguigu.distributedlock.mapper")

public class DistributedLockApplication {
    public static void main(String[] args) {
        SpringApplication.run(DistributedLockApplication.class, args);
   }
}

Stock实体类：

@Data
@TableName("db_stock")
public class Stock {
   @TableId
   private Long id;
   private String productCode;
   private String stockCode;
   private Integer count;
}

StockMapper接口：

public interface StockMapper extends BaseMapper<Stock> {
}

1.2. 简单实现减库存

接下来咱们代码实操一下

StockController：

@RestController
public class StockController {
    @Autowired
    private StockService stockService;
    @GetMapping("check/lock")
    public String checkAndLock(){
        this.stockService.checkAndLock();
        return "验库存并锁库存成功！";
    }
}

StockService：

@Service
public class StockService {
    @Autowired
    private StockMapper stockMapper;

    public void checkAndLock() {
// 先查询库存是否充足
        Stock stock = this.stockMapper.selectById(1L);
// 再减库存
        if (stock != null && stock.getCount() > 0) {
            stock.setCount(stock.getCount() - 1);
            this.stockMapper.updateById(stock);
        }
    }
}

测试：

查看数据库：

在浏览器中一个一个访问时，每访问一次，库存量减1，没有任何问题。

1.3. 演示超卖现象

接下来咱们使用jmeter压力测试工具，高并发下压测一下，添加线程组：并发100循环50次，即5000次请求。

给线程组添加HTTP Request请求：

填写测试接口路径如下：

再选择你想要的测试报表，例如这里选择聚合报告：

启动测试，查看压力测试报告：

测试结果：请求总数5000次，平均请求时间202ms，中位数（50%）请求是在173ms内完成的，90%请求是在344ms内完成的，最小耗时12ms，最大耗时1125ms，错误率0%，每秒钟平均473.8次。

查看mysql数据库剩余库存数：还有4870

此时如果还有人来下单，就会出现超卖现象（别人购买成功，而无货可发）。

1.4. jvm锁问题演示

使用jvm锁（synchronized关键字或者ReetrantLock）试试：

重启tomcat服务，再次使用jmeter压力测试，效果如下：

查看mysql数据库：

并没有发生超卖现象，完美解决。

1.4.2. 原理

添加synchronized关键字之后，StockService就具备了对象锁，由于添加了独占的排他锁，同一时刻只有一个请求能够获取到锁，并减库存。此时，所有请求只会one-by-one执行下去，也就不会发生超卖现象。

1.5. 多服务问题

使用jvm锁在单工程单服务情况下确实没有问题，但是在集群情况下会怎样？接下启动多个服务并使用nginx负载均衡，结构如下：

启动三个服务（端口号分别8000 8100 8200），如下：

1.5.1. 安装配置nginx

基于安装nginx：

# 拉取镜像

docker pull nginx:latest

# 创建nginx对应资源、日志及配置目录

mkdir -p /opt/nginx/logs /opt/nginx/conf /opt/nginx/html

# 先在conf目录下创建nginx.conf文件，配置内容参照下方

# 再运行容器

docker run -d -p 80:80 --name nginx -v /opt/nginx/html:/usr/share/nginx/html 

-v /opt/nginx/conf/nginx.conf:/etc/nginx/nginx.conf -v 
/opt/nginx/logs:/var/log/nginx nginx

user  nginx;
worker_processes  1;

error_log  /var/log/nginx/error.log warn;
pid        /var/run/nginx.pid;

events {
    worker_connections  1024;
}

http {
    include       /etc/nginx/mime.types;
    default_type  application/octet-stream;

    log_format  main  '$remote_addr - $remote_user [$time_local] "$request" '
                      '$status $body_bytes_sent "$http_referer" '
                      '"$http_user_agent" "$http_x_forwarded_for"';

    access_log  /var/log/nginx/access.log  main;

    sendfile        on;
    #tcp_nopush     on;

    keepalive_timeout  65;

    #gzip  on;

    #include /etc/nginx/conf.d/*.conf;
	
	upstream distributed {
		server 172.16.116.1:8000;
		server 172.16.116.1:8100;
		server 172.16.116.1:8200;
	}
	
	server {
		listen       80;
        server_name  172.16.116.100;
		location / {
			proxy_pass http://distributed;
		}
	}
	
}

在浏览器中测试：172.16.116.100是我的nginx服务器地址

经过测试，通过nginx访问服务一切正常。

1.5.2. 压力测试

注意：先把数据库库存量还原到5000。

参照之前的测试用例，再创建一个新的测试组：参数给之前一样

配置nginx的地址及服务的访问路径如下：

测试结果：性能只是略有提升。

数据库库存剩余量如下：

又出现了并发问题，即出现了超卖现象。

1.6. mysql锁演示

除了使用jvm锁之外，还可以使用数据锁：悲观锁或者乐观锁

悲观锁：在读取数据时锁住那几行，其他对这几行的更新需要等到悲观锁结束时才能继续。乐观所：读取数据时不锁，更新时检查是否数据已经被更新过，如果是则取消当前更新，一般在悲观锁的等待时间过长而不能接受时我们才会选择乐观锁。

1.6.1. mysql悲观锁

在MySQL的InnoDB中，预设的Tansaction isolation level 为REPEATABLE READ（可重读）

在SELECT 的读取锁定主要分为两种方式：

SELECT ... LOCK IN SHARE MODE （共享锁）
SELECT ... FOR UPDATE （悲观锁）

这两种方式在事务(Transaction) 进行当中SELECT 到同一个数据表时，都必须等待其它事务数据被提交(Commit)后才会执行。而主要的不同在于LOCK IN SHARE MODE 在有一方事务要Update 同一个表单时很容易造成死锁。简单的说，如果SELECT 后面若要UPDATE 同一个表单，最好使用SELECT ... FOR UPDATE。

代码实现改造StockService：

在StockeMapper中定义selectStockForUpdate方法：

public interface StockMapper extends BaseMapper<Stock> {
    public Stock selectStockForUpdate(Long id);
}

在StockMapper.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.atguigu.distributedlock.mapper.StockMapper">
    <select id="selectStockForUpdate" 
resultType="com.atguigu.distributedlock.pojo.Stock">
       select * from db_stock where id = #{id} for update
    </select>
</mapper>

压力测试

注意：测试之前，需要把库存量改成5000。压测数据如下：比jvm性能高很多，比无锁要低将近1倍

mysql数据库存：

1.6.2. mysql乐观锁

乐观锁（ Optimistic Locking ）相对悲观锁而言，乐观锁假设认为数据一般情况下不会造成冲突，所以在数据进行提交更新的时候，才会正式对数据的冲突与否进行检测，如果发现冲突了，则重试。那么我们如何实现乐观锁呢？

使用数据版本（Version）记录机制实现，这是乐观锁最常用的实现方式。一般是通过为数据库表增加一个数字类型的 “version” 字段来实现。当读取数据时，将version字段的值一同读出，数据每更新一次，对此version值加一。当我们提交更新的时候，判断数据库表对应记录的当前版本信息与第一次取出来的version值进行比对，如果数据库表当前版本号与第一次取出来的version值相等，则予以更新。

给db_stock表添加version字段：

对应也需要给Stock实体类添加version属性。此处略。

代码实现

public void checkAndLock() {
    // 先查询库存是否充足
    Stock stock = this.stockMapper.selectById(1L);
    // 再减库存
    if (stock != null && stock.getCount() > 0){
        // 获取版本号
        Long version = stock.getVersion();
        stock.setCount(stock.getCount() - 1);
        // 每次更新 版本号 + 1
        stock.setVersion(stock.getVersion() + 1);
        // 更新之前先判断是否是之前查询的那个版本，如果不是重试
        if (this.stockMapper.update(stock, new UpdateWrapper<Stock>
().eq("id", stock.getId()).eq("version", version)) == 0) {
            checkAndLock();
       }
   }
}

重启后使用jmeter压力测试工具结果如下：

修改测试参数如下：

测试结果如下：

说明乐观锁在并发量越大的情况下，性能越低（因为需要大量的重试）；并发量越小，性能越高。

1.6.3. mysql锁缺陷

在数据库集群情况下会导致数据库锁失效，并且很多数据库集群的中间件压根就不支持悲观锁。例如：mycat，在读写分离的场景下可能会导致乐观锁不可靠。这把锁强依赖数据库的可用性，数据库是一个单点，一旦数据库挂掉，会导致业务系统不可用。

2. 基于mysql实现分布式锁

不管是jvm锁还是mysql锁，为了保证线程的并发安全，都提供了悲观独占排他锁。所以独占排他也是分布式锁的基本要求。可以利用唯一键索引不能重复插入的特点实现。设计表如下：

CREATE TABLE `db_lock` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `lock_name` varchar(50) NOT NULL COMMENT '锁名',
  `class_name` varchar(100) DEFAULT NULL COMMENT '类名',
  `method_name` varchar(50) DEFAULT NULL COMMENT '方法名',
  `server_name` varchar(50) DEFAULT NULL COMMENT '服务器ip',
  `thread_name` varchar(50) DEFAULT NULL COMMENT '线程名',
  `create_time` timestamp NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP 

COMMENT '获取锁时间',
  `desc` varchar(100) DEFAULT NULL COMMENT '描述',
  PRIMARY KEY (`id`),
  UNIQUE KEY `idx_unique` (`lock_name`)
) ENGINE=InnoDB AUTO_INCREMENT=1332899824461455363 DEFAULT CHARSET=utf8;

Lock实体类：

@Data
@AllArgsConstructor
@NoArgsConstructor
@TableName("db_lock")

public class Lock {
    private Long id;
    private String lockName;
    private String className;
    private String methodName;
    private String serverName;
    private String threadName;
    private Date createTime;
    private String desc;
}

LockMapper接口：

public interface LockMapper extends BaseMapper<Lock> {
}

2.1. 基本思路

synchronized关键字和ReetrantLock锁都是独占排他锁，即多个线程争抢一个资源时，同一时刻只有一个线程可以抢占该资源，其他线程只能阻塞等待，直到占有资源的线程释放该资源。

线程同时获取锁（insert）
获取成功，执行业务逻辑，执行完成释放锁（delete）
其他线程等待重试

2.2. 代码实现

改造StockService：

@Service

public class StockService {
    @Autowired
    private StockMapper stockMapper;
    @Autowired
    private LockMapper lockMapper;
    /**
     * 数据库分布式锁
     */
    public void checkAndLock() {
        // 加锁
        Lock lock = new Lock(null, "lock", this.getClass().getName(), new 
Date(), null);
        try {
            this.lockMapper.insert(lock);
       } catch (Exception ex) {
            // 获取锁失败，则重试
            try {
                Thread.sleep(50);
                this.checkAndLock();
           } catch (InterruptedException e) {
                e.printStackTrace();
           }
       }
        // 先查询库存是否充足
        Stock stock = this.stockMapper.selectById(1L);
        // 再减库存
        if (stock != null && stock.getCount() > 0){
            stock.setCount(stock.getCount() - 1);
            this.stockMapper.updateById(stock);
       }
        // 释放锁
        this.lockMapper.deleteById(lock.getId());
   }
}

加锁：

// 加锁
Lock lock = new Lock(null, "lock", this.getClass().getName(), new Date(), null);
try {
    this.lockMapper.insert(lock);
} catch (Exception ex) {
    // 获取锁失败，则重试
    try {
        Thread.sleep(50);
        this.checkAndLock();
   } catch (InterruptedException e) {
        e.printStackTrace();
   }
}

解锁：

// 释放锁
this.lockMapper.deleteById(lock.getId());

使用Jmeter压力测试结果：

可以看到性能感人。mysql数据库库存余量为0，可以保证线程安全。

2.3. 缺陷及解决方案

1. 这把锁强依赖数据库的可用性，数据库是一个单点，一旦数据库挂掉，会导致业务系统不可用。解决方案：给锁数据库搭建主备
2. 这把锁没有失效时间，一旦解锁操作失败，就会导致锁记录一直在数据库中，其他线程无法再获得到锁。解决方案：只要做一个定时任务，每隔一定时间把数据库中的超时数据清理一遍。
3. 这把锁是非重入的，同一个线程在没有释放锁之前无法再次获得该锁。因为数据中数据已经存在了。解决方案：记录获取锁的主机信息和线程信息，如果相同线程要获取锁，直接重入。
4. 受制于数据库性能，并发能力有限。解决方案：无法解决。

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2023-09-15，如有侵权请联系 cloudcommunity@tencent.com 删除

数据库