MySQL做分布式锁

这篇具有很好参考价值的文章主要介绍了MySQL做分布式锁。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

分布式锁mysql实现方式
方式1:唯一索引

  • 创建锁表,内部存在字段表示资源名及资源描述,同一资源名使用数据库唯一性限制。
  • 多个进程同时往数据库锁表中写入对某个资源的占有记录,当某个进程成功写入时则表示其获取锁成功
  • 其他进程由于资源字段唯一性限制插入失败陷入自旋并且失败重试。
  • 当执行完业务后持有该锁的进程则删除该表内的记录,此时回到步骤一。
    MySQL做分布式锁,学习笔记,mysql,分布式,数据库
    表数据
create table `database_lock`(
	`id` BIGINT NOT NULL AUTO_INCREMENT,
	`resource` INT NOT NULL COMMENT '锁资源',
	`description` varchar(1024) NOT NULL DEFAULT "" COMMENT '描述',
	PRIMARY KEY (`id`),
	UNIQUE KEY `resource` (`resource`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='数据库分布式锁表';

db.properties

driver=com.mysql.cj.jdbc.Driver
url=jdbc:mysql://localhost:3306/distribute_lock?useUnicode=true&characterEncoding=utf-8&useSSL=true&serverTimezone=Asia/Shanghai
user=root
password=123456
  • PropertiesReader
@Slf4j
public class PropertiesReader {

    // Properties缓存文件
    private static final Map<String, Properties> propertiesCache = new HashMap<String, Properties>();

    public static Properties getProperties(String propertiesName) throws IOException {
        if (propertiesCache.containsKey(propertiesName)) {
            return propertiesCache.get(propertiesName);
        }
        loadProperties(propertiesName);
        return propertiesCache.get(propertiesName);
    }

    private synchronized static void loadProperties(String propertiesName) throws IOException {
        FileReader fileReader = null;
        try {
            // 创建Properties集合类
            Properties pro = new Properties();
            // 获取src路径下的文件--->ClassLoader类加载器
            ClassLoader classLoader = PropertiesReader.class.getClassLoader();
            URL resource = classLoader.getResource(propertiesName);
            // 获取配置路径
            String path = resource.getPath();
            // 读取文件
            fileReader = new FileReader(path);
            // 加载文件
            pro.load(fileReader);
            // 初始化
            propertiesCache.put(propertiesName, pro);
        } catch (IOException e) {
            log.error("读取Properties文件失败,Properties名为:" + propertiesName);
            throw e;
        } finally {
            try {
                if (fileReader != null) {
                    fileReader.close();
                }
            } catch (IOException e) {
                log.error("fileReader关闭失败!", e);
            }
        }
    }
}
  • JDBCUtils
@Slf4j
public class JDBCUtils {
    private static String url;
    private static String user;
    private static String password;

    static {
        //读取文件,获取值
        try {
            Properties properties = PropertiesReader.getProperties("db.properties");
            url = properties.getProperty("url");
            user = properties.getProperty("user");
            password = properties.getProperty("password");
            String driver = properties.getProperty("driver");
            //4.注册驱动
            Class.forName(driver);
        } catch (IOException | ClassNotFoundException e) {
            log.error("初始化jdbc连接失败!", e);
        }
    }

    /**
     * 获取连接
     * @return 连接对象
     */
    public static Connection getConnection() throws SQLException {
        return DriverManager.getConnection(url, user, password);
    }

    /**
     * 释放资源
     * @param rs
     * @param st
     * @param conn
     */
    public static void close(ResultSet rs, Statement st, Connection conn) {
        if (rs != null) {
            try {
                rs.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        if (st != null) {
            try {
                st.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        if (conn != null) {
            try {
                conn.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
    }
}

数据库操作类

/**
 * MySQL 锁操作类(加锁+释放锁)
 */
@Slf4j
public class MySQLDistributedLockService {

    private static Connection connection;
    private static Statement statement;
    private static ResultSet resultSet;

    static{
        try {
            connection = JDBCUtils.getConnection();
            statement = connection.createStatement();
            resultSet = null;
        } catch (SQLException e) {
            log.error("数据库连接失败!");
        }
    }

    /**
     * 锁表 - 获取锁
     * @param resource      资源
     * @param description   锁描述
     * @return  是否操作成功
     */
    public static boolean tryLock(int resource,String description){
        String sql = "insert into database_lock (resource,description) values (" + resource + ", '" + description + "');";
        //获取数据库连接
        try {
            int stat = statement.executeUpdate(sql);
            return stat == 1;
        } catch (SQLException e) {
            return false;
        }
    }
    /**
     * 锁表-释放锁
     * @return
     */
    public static boolean releaseLock(int resource) throws SQLException {
        String sql = "delete from database_lock where resource = " + resource;
        //获取数据库连接
        int stat = statement.executeUpdate(sql);
        return stat == 1;
    }
    /**
     * 关闭连接
     */
    public static void close(){
        log.info("当前线程: " + ManagementFactory.getRuntimeMXBean().getName().split("@")[0] +
                ",关闭了数据库连接!");
        JDBCUtils.close(resultSet,statement,connection);
    }
}

LockTable

/**
 * mysql分布式锁
 *      执行流程: 多进程抢占数据库某个资源,然后执行业务,执行完释放资源
 *      锁机制: 单一进程获取锁时,则其他进程提交失败
 */
@Slf4j
public class LockTable extends Thread {
    @Override
    public void run() {
        super.run();
        //获取Java虚拟机的进程ID
        String pid = ManagementFactory.getRuntimeMXBean().getName().split("@")[0];
        try{
            while(true){
                log.info("当前进程PID:" + pid + ",尝试获取锁资源!");
                if(MySQLDistributedLockService.tryLock(1,"测试锁")){
                    log.info("当前进程PID:" + pid + ",获取锁资源成功!");
                    //sleep模拟业务处理过程
                    log.info("开始处理业务!");
                    Thread.sleep(10*1000);
                    log.info("业务处理完成!");
                    MySQLDistributedLockService.releaseLock(1);
                    log.info("当前进程PID: " + pid + ",释放了锁资源!");
                    break;
                }else{
                    log.info("当前进程PID: " + pid + ",获取锁资源失败!");
                    Thread.sleep(2000);
                }
            }
        }catch (Exception e){
            log.error("抢占锁发生错误!",e);
        }finally {
            MySQLDistributedLockService.close();
        }
    }
    // 程序入口
    public static void main(String[] args) {
        new LockTable().start();
    }
}

测试
运行时开启并行执行选项,每次运行三个或三个以上进程. Allow parallel run 运行并行执行
MySQL做分布式锁,学习笔记,mysql,分布式,数据库
MySQL做分布式锁,学习笔记,mysql,分布式,数据库
注意事项:

  • 该锁为非阻塞的
  • 当某进程持有锁并且挂死时候会造成资源一直不释放的情况,造成死锁,因此需要维护一个定时清理任务去清理持有过久的锁
  • 要注意数据库的单点问题,最好设置备库,进一步提高可靠性
  • 该锁为非可重入锁,如果要设置成可重入锁需要添加数据库字段记录持有该锁的设备信息以及加锁次数

方式二:基于乐观锁

  • 每次执行业务前首先进行数据库查询,查询当前的需要修改的资源值(或版本号)。
  • 进行资源的修改操作,并且修改前进行资源(或版本号)的比对操作,比较此时数据库中的值是否和上一步查询结果相同。
  • 查询结果相同则修改对应资源值,不同则回到第一步。
    MySQL做分布式锁,学习笔记,mysql,分布式,数据库

例子:数据库中设定某商品基本信息(名为外科口罩,数量为10),多进程对该商品进行抢购,当商品数量为0时结束抢购。
MySQL做分布式锁,学习笔记,mysql,分布式,数据库
代码实现

    /**
     * 乐观锁-获取资源
     * @param id 资源ID
     * @return result
     */
    public static ResultSet getGoodCount(int id) throws SQLException {

        String sql = "select  * from  database_lock_2 where id = " + id;

        //查询数据
        resultSet = statement.executeQuery(sql);
        return resultSet;
    }


    /**
     * 乐观锁-修改资源
     * @param id        资源ID
     * @param goodCount 资源
     * @return  修改状态
     */
    public static boolean setGoodCount(int id, int goodCount) throws SQLException {

        String sql = "update database_lock_2 set good_count = good_count - 1 where id =" + id +"  and good_count = " + goodCount;

        int stat = statement.executeUpdate(sql);
        return stat == 1;
    }

    /**
     * 乐观锁-开启事务自动提交
     */
    public static void AutoCommit(){
        try {
            connection.setAutoCommit(true);
        } catch (SQLException e) {
            log.error("开启自动提交!",e);
        }
    }

OptimisticLock测试类

/**
 * mysql分布式锁-乐观锁
 *  执行流程: 多进程抢购同一商品,每次抢购成功商品数量-1,商品数据量为0时退出
 *  锁机制: 单一进程获取锁时,则其他进程提交失败
 */
@Slf4j
public class OptimisticLock extends Thread{

    @Override
    public void run() {
        super.run();
        String pid = ManagementFactory.getRuntimeMXBean().getName().split("@")[0];

        ResultSet resultSet = null;
        String goodName = null;
        int goodCount = 0;
        try {
            while(true){
                log.info("当前线程:" + pid + ",开始抢购商品!");
                //获取当前商品信息
                resultSet = MySQLDistributedLockService.getGoodCount(1);
                while (resultSet.next()){
                    goodName = resultSet.getString("good_name");
                    goodCount = resultSet.getInt("good_count");
                }
                log.info("获取库存成功,当前商品名为:" + goodName + ",当前库存剩余量为:" + goodCount);
                //模拟执行业务操作
                Thread.sleep(2*3000);
                if(0 == goodCount){
                    log.info("抢购失败,当前库存为0! ");
                    break;
                }
                //修改库存信息,库存量-1
                if(MySQLDistributedLockService.setGoodCount(1,goodCount)){
                    log.info("当前线程:" + pid + " 抢购商品:" + goodName + "成功,剩余库存为:" + (goodCount -1));
                    //模拟延迟,防止锁每次被同一进程获取
                    Thread.sleep(2 * 1000);
                }else{
                    log.error("抢购商品:" + goodName +"失败,商品数量已被修改");
                }
            }
        }catch (Exception e){
            log.error("抢购商品发生错误!",e);
        }finally {
            if(resultSet != null){
                try {
                    resultSet.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                    log.error("关闭Result失败!" , e);
                }
            }
            MySQLDistributedLockService.close();
        }
    }

    public static void main(String[] args) {
        new OptimisticLock().start();
    }
}

代码测试
开启三个进程,查看执行情况
MySQL做分布式锁,学习笔记,mysql,分布式,数据库
MySQL做分布式锁,学习笔记,mysql,分布式,数据库
MySQL做分布式锁,学习笔记,mysql,分布式,数据库
注意事项:

  • 该锁为非阻塞的
  • 该锁对于业务具有侵入式,如果设置版本号校验则增加的额外的字段,增加了数据库冗余
  • 当并发量过高时会有大量请求访问数据库的某行记录,对数据库造成很大的写压力
  • 因此乐观锁适用于并发量不高,并且写操作不频繁的场景

方式三:悲观锁实现方式(利用事务加上行/表锁)
MySQL做分布式锁,学习笔记,mysql,分布式,数据库
实现思路

  • 关闭jdbc连接自动commit属性
  • 每次执行业务前先使用查询语句后接for update表示锁定该行数据(注意查询条件如果未命中主键或索引,此时将会从行锁变为表锁)
  • 执行业务流程修改表资源
  • 执行commit操作

代码实现
MySQLDistributedLockService


    /**
     * 悲观锁-获取资源
     * @param id 资源ID
     * @return result
     */
    public static ResultSet getGoodCount2(int id) throws SQLException {
        String sql = "select  * from  database_lock_2 where id = " + id + "for update";
        //查询数据
        resultSet = statement.executeQuery(sql);
        return resultSet;
    }

    /**
     * 悲观锁-修改资源
     * @param id        资源ID
     * @return  修改状态
     */
    public static boolean setGoodCount2(int id) throws SQLException {

        String sql = "update database_lock_2 set good_count = good_count - 1 where id =" + id;
        int stat = statement.executeUpdate(sql);
        return stat == 1;
    }

    /**
     * 悲观锁-关闭事务自动提交
     */
    public static void closeAutoCommit(){
        try {
            connection.setAutoCommit(false);
        } catch (SQLException e) {
            log.error("关闭自动提交失败!",e);
        }
    }

    /**
     * 悲观锁-提交事务
     */
    public static void commit(String pid,String goodName,int goodCount) throws SQLException {
        connection.commit();
        log.info("当前线程:" + pid + "抢购商品: " + goodName + "成功,剩余库存为:" + (goodCount-1));
    }

    /**
     * 悲观锁-回滚
     */
    public static void rollBack() throws SQLException {
        connection.rollback();
    }

PessimisticLock

/**
 * mysql 分布式锁-悲观锁
 *     执行流程:多个进程抢占同一个商品,执行业务完毕则通过connection.commit() 释放锁
 *     锁机制:单一进程获取锁时,则其他进程将阻塞等待
 */
@Slf4j
public class PessimisticLock extends Thread {
    @Override
    public void run() {
        super.run();
        ResultSet resultSet = null;
        String goodName = null;
        int goodCount = 0;
        String pid = ManagementFactory.getRuntimeMXBean().getName().split("@")[0];
        //关闭自动提交
        MySQLDistributedLockService.closeAutoCommit();
        try{
            while(true){
                log.info("当前线程:" + pid + "");
                //获取库存
                resultSet = MySQLDistributedLockService.getGoodCount2(1);
                while (resultSet.next()) {
                    goodName = resultSet.getString("good_name");
                    goodCount = resultSet.getInt("good_count");
                }
                log.info("获取库存成功,当前商品名称为:" + goodName + ",当前库存剩余量为:" + goodCount);
                // 模拟执行业务事件
                Thread.sleep(2 * 1000);
                if (0 == goodCount) {
                    log.info("抢购失败,当前库存为0!");
                    break;
                }
                // 抢购商品
                if (MySQLDistributedLockService.setGoodCount2(1)) {
                    // 模拟延时,防止锁每次被同一进程获取
                    MySQLDistributedLockService.commit(pid, goodName, goodCount);
                    Thread.sleep(2 * 1000);
                } else {
                    log.error("抢购商品:" + goodName + "失败!");
                }
            }
        }catch (Exception e){
            //抢购失败
            log.error("抢购商品发生错误!",e);
            try {
                MySQLDistributedLockService.rollBack();
            } catch (SQLException ex) {
                log.error("回滚失败! ",e);
            }
        }finally {
            if(resultSet != null){
                try {
                    resultSet.close();
                } catch (SQLException e) {
                    log.error("Result关闭失败!",e);
                }
            }
            MySQLDistributedLockService.close();
        }
    }

    public static void main(String[] args) {
        new PessimisticLock().start();
    }
}

测试结果
MySQL做分布式锁,学习笔记,mysql,分布式,数据库
MySQL做分布式锁,学习笔记,mysql,分布式,数据库
MySQL做分布式锁,学习笔记,mysql,分布式,数据库
注意事项:

  • 该锁为阻塞锁
  • 每次请求存在额外加锁的开销
  • 在并发量很高的情况下会造成系统中存在大量阻塞的请求,影响系统的可用性
  • 因此悲观锁适用于并发量不高,读操作不频繁的写场景

总结:文章来源地址https://www.toymoban.com/news/detail-623717.html

  • 在实际使用中,由于受到性能以及稳定性约束,对于关系型数据库实现的分布式锁一般很少被用到。但是对于一些并发量不高、系统仅提供给内部人员使用的单一业务场景可以考虑使用关系型数据库分布式锁,因为其复杂度较低,可靠性也能够得到保证。

到了这里,关于MySQL做分布式锁的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 如何使用mysql实现分布式锁

    目录 什么是分布式锁? 如何实现分布式锁? 定义分布式表结构 定义锁统一接口 使用mysql来实现分布式锁 ① 生成线程标记ID ② 加锁 ③ 解锁 ④ 重置锁 写在最后 1. 什么是分布式锁? 百度百科:分布式锁是控制分布式系统之间同步访问共享资源的一种方式。 ㅤ如引用所述,

    2024年02月05日
    浏览(42)
  • 只用Mysql搞一个分布式锁

    在web开发中,分布式的锁的应用场景甚多,我们可以通过分布式锁来进行一些仅依赖于数据库的事务很难直接保证原子性的操作,比如多种不同数据源的访问,网络通信等等。多数情况下我们会使用memcache的add, redis中在set中指定nx参数等来完成。 下面介绍一个仅依赖Mysql来完

    2024年01月22日
    浏览(45)
  • 分布式锁实现(mysql,以及redis)以及分布式的概念(续)redsync包使用

    这张尽量结合上一章进行使用:上一章 这章主要是讲如何通过 redis 实现分布式锁的 这里我用 redis 去实现: 技术: golang , redis , 数据结构 这里是有一个大体的实现思路:主要是使用 redis 中这些语法 redis 命令说明: setnx 命令: set if not exists ,当且仅当 key 不存在时,将 ke

    2024年01月22日
    浏览(64)
  • 分布式学习笔记

    Consistency(一致性):用户访问分布式系统中的任意节点,得到的数据必须一致。 Availability(可用性):用户访问集群中的任意健康节点,必须得到相应,而不是超时或拒绝。 Partition tolerance (分区容忍性):因为网络故障或其他原因导致分布式系统中的部分节点与其他节点

    2024年02月20日
    浏览(34)
  • 分布式锁,学习笔记

    什么是分布式锁 1.1 作用: 保证数据的正确性: 比如:秒杀的时候防止商品超卖,接口幂等性。 避免重复处理数据: 比如:1避免调度任务在多台机器重复执行,2避免缓存过期所有请求都去加载数据库。 一个分布式锁需要考虑的问题: 1互斥阻塞。2锁需要可重入。3过期时间

    2024年02月17日
    浏览(42)
  • 分别使用Redis、MySQL、ZooKeeper构建分布式锁

    本文使用Java构建三种中间件的分布式锁,下面介绍下三种分布式锁的优缺点, 使用MySQL构建分布式锁 ,因为数据库数据存储在磁盘中,所以IO速率相对较慢,因此构建出来的分布式锁不适合用在高并发场景,对于一些对并发要求不高的系统中可以使用,进一步提高系统的安全

    2024年02月06日
    浏览(46)
  • ​【五一创作】基于mysql关系型实现分布式锁

    看完该文预计用时:15分钟 看之前应具体的技术栈:springboot mysql nginx(了解即可) 目录 0.写在前面 1. 从减库存聊起 1.1. 环境准备   1.2. 简单实现减库存  1.3. 演示超卖现象 1.4. jvm锁问题演示  1.4.2. 原理 1.5. 多服务问题  1.5.1. 安装配置nginx 1.5.2. 压力测试  1.6. mysql锁演示 1

    2024年02月05日
    浏览(45)
  • MySQL 按日期流水号 条码 分布式流水号

    有这样一个场景,有多台终端,要获取唯一的流水号,流水号格式是 日期+0001形式,使用MySQL的存储过程+全局锁实现这个需求。 以下是代码示例。 注:所有的终端连接到MySQL服务器获取流水号,如果获取到的是 “-1”,需要重新获取。

    2024年02月02日
    浏览(37)
  • 大模型学习笔记08——分布式训练

    模型规模的扩大,对硬件(算力、内存)的发展提出要求。然而,因为内存墙的存在,单一设备的算力及容量,受限于物理定律,持续提高芯片的集成越来越困难,难以跟上模型扩大的需求。 为了解决算力增速不足的问题,人们考虑用多节点集群进行分布式训练,以提升算力

    2024年01月23日
    浏览(46)
  • 分布式Erlang/OTP(学习笔记)(一)

    假设你在机器A和机器B上各跑着一个Simple Cache应用的实例。要是在机器A的缓存上插人一个键/值对之后,从机器B上也可以访问,那可就好了。显然,要达到这个目的,机器A必须以某种方式将相关信息告知给机器B。传递该信息的方式有很多,有些方式简单,有些方式复杂。但无

    2024年01月19日
    浏览(61)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包