首页 / MYSQL / Mysql高手系列 - 第26篇：聊聊如何使用mysql实现分布式锁

Mysql高手系列 - 第26篇：聊聊如何使用mysql实现分布式锁

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了Mysql高手系列 - 第26篇：聊聊如何使用mysql实现分布式锁，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含10390字，纯文字阅读大概需要15分钟。

内容图文

Mysql系列的目标是：通过这个系列从入门到全面掌握一个高级开发所需要的全部技能。

欢迎大家加我微信itsoku一起交流java、算法、数据库相关技术。

这是Mysql系列第26篇。

本篇我们使用mysql实现一个分布式锁。

分布式锁的功能

分布式锁使用者位于不同的机器中，锁获取成功之后，才可以对共享资源进行操作
锁具有重入的功能：即一个使用者可以多次获取某个锁
获取锁有超时的功能：即在指定的时间内去尝试获取锁，超过了超时时间，如果还未获取成功，则返回获取失败
能够自动容错，比如：A机器获取锁lock1之后，在释放锁lock1之前，A机器挂了，导致锁lock1未释放，结果会lock1一直被A机器占有着，遇到这种情况时，分布式锁要能够自动解决，可以这么做：持有锁的时候可以加个持有超时时间，超过了这个时间还未释放的，其他机器将有机会获取锁

预备技能：乐观锁

通常我们修改表中一条数据过程如下：

t1：select获取记录R1
t2：对R1进行编辑
t3：update R1

我们来看一下上面的过程存在的问题：

如果A、B两个线程同时执行到t1，他们俩看到的R1的数据一样，然后都对R1进行编辑，然后去执行t3，最终2个线程都会更新成功，后面一个线程会把前面一个线程update的结果给覆盖掉，这就是并发修改数据存在的问题。

我们可以在表中新增一个版本号，每次更新数据时候将版本号作为条件，并且每次更新时候版本号+1，过程优化一下，如下：

t1：打开事务start transaction
t2：select获取记录R1,声明变量v=R1.version
t3：对R1进行编辑
t4：执行更新操作
    update R1 set version = version + 1 where user_id=#user_id# and version = #v#;
t5：t4中的update会返回影响的行数，我们将其记录在count中，然后根据count来判断提交还是回滚
    if(count==1){
        //提交事务
        commit;
    }else{
        //回滚事务
        rollback;
    }

上面重点在于步骤t4，当多个线程同时执行到t1，他们看到的R1是一样的，但是当他们执行到t4的时候，数据库会对update的这行记录加锁，确保并发情况下排队执行，所以只有第一个的update会返回1，其他的update结果会返回0，然后后面会判断count是否为1，进而对事务进行提交或者回滚。可以通过count的值知道修改数据是否成功了。

上面这种方式就乐观锁。我们可以通过乐观锁的方式确保数据并发修改过程中的正确性。

使用mysql实现分布式锁

建表

我们创建一个分布式锁表，如下

DROP DATABASE IF EXISTS javacode2018;
CREATE DATABASE javacode2018;
USE javacode2018;
DROP TABLE IF EXISTS t_lock;
create table t_lock(
  lock_key varchar(32) PRIMARY KEY NOT NULL COMMENT '锁唯一标志',
  request_id varchar(64) NOT NULL DEFAULT '' COMMENT '用来标识请求对象的',
  lock_count INT NOT NULL DEFAULT 0 COMMENT '当前上锁次数',
  timeout BIGINT NOT NULL DEFAULT 0 COMMENT '锁超时时间',
  version INT NOT NULL DEFAULT 0 COMMENT '版本号，每次更新+1'
)COMMENT '锁信息表';

分布式锁工具类：

package com.itsoku.sql;

import lombok.Builder;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.sql.*;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.TimeUnit;


/**
 * 工作10年的前阿里P7分享Java、算法、数据库方面的技术干货！坚信用技术改变命运，让家人过上更体面的生活！
 * 喜欢的请关注公众号：路人甲Java
 */
@Slf4j
public class LockUtils {

    //将requestid保存在该变量中
    static ThreadLocal<String> requestIdTL = new ThreadLocal<>();

    /**
     * 获取当前线程requestid
     *
     * @return
     */
    public static String getRequestId() {
        String requestId = requestIdTL.get();
        if (requestId == null || "".equals(requestId)) {
            requestId = UUID.randomUUID().toString();
            requestIdTL.set(requestId);
        }
        log.info("requestId:{}", requestId);
        return requestId;
    }

    /**
     * 获取锁
     *
     * @param lock_key        锁key
     * @param locktimeout(毫秒) 持有锁的有效时间，防止死锁
     * @param gettimeout(毫秒)  获取锁的超时时间，这个时间内获取不到将重试
     * @return
     */
    public static boolean lock(String lock_key, long locktimeout, int gettimeout) throws Exception {
        log.info("start");
        boolean lockResult = false;
        String request_id = getRequestId();
        long starttime = System.currentTimeMillis();
        while (true) {
            LockModel lockModel = LockUtils.get(lock_key);
            if (Objects.isNull(lockModel)) {
                //插入一条记录,重新尝试获取锁
                LockUtils.insert(LockModel.builder().lock_key(lock_key).request_id("").lock_count(0).timeout(0L).version(0).build());
            } else {
                String reqid = lockModel.getRequest_id();
                //如果reqid为空字符，表示锁未被占用
                if ("".equals(reqid)) {
                    lockModel.setRequest_id(request_id);
                    lockModel.setLock_count(1);
                    lockModel.setTimeout(System.currentTimeMillis() + locktimeout);
                    if (LockUtils.update(lockModel) == 1) {
                        lockResult = true;
                        break;
                    }
                } else if (request_id.equals(reqid)) {
                    //如果request_id和表中request_id一样表示锁被当前线程持有者，此时需要加重入锁
                    lockModel.setTimeout(System.currentTimeMillis() + locktimeout);
                    lockModel.setLock_count(lockModel.getLock_count() + 1);
                    if (LockUtils.update(lockModel) == 1) {
                        lockResult = true;
                        break;
                    }
                } else {
                    //锁不是自己的，并且已经超时了，则重置锁，继续重试
                    if (lockModel.getTimeout() < System.currentTimeMillis()) {
                        LockUtils.resetLock(lockModel);
                    } else {
                        //如果未超时，休眠100毫秒，继续重试
                        if (starttime + gettimeout > System.currentTimeMillis()) {
                            TimeUnit.MILLISECONDS.sleep(100);
                        } else {
                            break;
                        }
                    }
                }
            }
        }
        log.info("end");
        return lockResult;
    }

    /**
     * 释放锁
     *
     * @param lock_key
     * @throws Exception
     */
    public static void unlock(String lock_key) throws Exception {
        //获取当前线程requestId
        String requestId = getRequestId();
        LockModel lockModel = LockUtils.get(lock_key);
        //当前线程requestId和库中request_id一致 && lock_count>0，表示可以释放锁
        if (Objects.nonNull(lockModel) && requestId.equals(lockModel.getRequest_id()) && lockModel.getLock_count() > 0) {
            if (lockModel.getLock_count() == 1) {
                //重置锁
                resetLock(lockModel);
            } else {
                lockModel.setLock_count(lockModel.getLock_count() - 1);
                LockUtils.update(lockModel);
            }
        }
    }

    /**
     * 重置锁
     *
     * @param lockModel
     * @return
     * @throws Exception
     */
    public static int resetLock(LockModel lockModel) throws Exception {
        lockModel.setRequest_id("");
        lockModel.setLock_count(0);
        lockModel.setTimeout(0L);
        return LockUtils.update(lockModel);
    }

    /**
     * 更新lockModel信息，内部采用乐观锁来更新
     *
     * @param lockModel
     * @return
     * @throws Exception
     */
    public static int update(LockModel lockModel) throws Exception {
        return exec(conn -> {
            String sql = "UPDATE t_lock SET request_id = ?,lock_count = ?,timeout = ?,version = version + 1 WHERE lock_key = ? AND  version = ?";
            PreparedStatement ps = conn.prepareStatement(sql);
            int colIndex = 1;
            ps.setString(colIndex++, lockModel.getRequest_id());
            ps.setInt(colIndex++, lockModel.getLock_count());
            ps.setLong(colIndex++, lockModel.getTimeout());
            ps.setString(colIndex++, lockModel.getLock_key());
            ps.setInt(colIndex++, lockModel.getVersion());
            return ps.executeUpdate();
        });
    }

    public static LockModel get(String lock_key) throws Exception {
        return exec(conn -> {
            String sql = "select * from t_lock t WHERE t.lock_key=?";
            PreparedStatement ps = conn.prepareStatement(sql);
            int colIndex = 1;
            ps.setString(colIndex++, lock_key);
            ResultSet rs = ps.executeQuery();
            if (rs.next()) {
                return LockModel.builder().
                        lock_key(lock_key).
                        request_id(rs.getString("request_id")).
                        lock_count(rs.getInt("lock_count")).
                        timeout(rs.getLong("timeout")).
                        version(rs.getInt("version")).build();
            }
            return null;
        });
    }

    public static int insert(LockModel lockModel) throws Exception {
        return exec(conn -> {
            String sql = "insert into t_lock (lock_key, request_id, lock_count, timeout, version) VALUES (?,?,?,?,?)";
            PreparedStatement ps = conn.prepareStatement(sql);
            int colIndex = 1;
            ps.setString(colIndex++, lockModel.getLock_key());
            ps.setString(colIndex++, lockModel.getRequest_id());
            ps.setInt(colIndex++, lockModel.getLock_count());
            ps.setLong(colIndex++, lockModel.getTimeout());
            ps.setInt(colIndex++, lockModel.getVersion());
            return ps.executeUpdate();
        });
    }

    public static <T> T exec(SqlExec<T> sqlExec) throws Exception {
        Connection conn = getConn();
        try {
            return sqlExec.exec(conn);
        } finally {
            closeConn(conn);
        }
    }

    @FunctionalInterface
    public interface SqlExec<T> {
        T exec(Connection conn) throws Exception;
    }

    @Getter
    @Setter
    @Builder
    public static class LockModel {
        private String lock_key;
        private String request_id;
        private Integer lock_count;
        private Long timeout;
        private Integer version;
    }

    private static final String url = "jdbc:mysql://localhost:3306/javacode2018?useSSL=false";        //数据库地址
    private static final String username = "root";        //数据库用户名
    private static final String password = "root123";        //数据库密码
    private static final String driver = "com.mysql.jdbc.Driver";        //mysql驱动

    /**
     * 连接数据库
     *
     * @return
     */
    public static Connection getConn() {
        Connection conn = null;
        try {
            Class.forName(driver);  //加载数据库驱动
            try {
                conn = DriverManager.getConnection(url, username, password);  //连接数据库
            } catch (SQLException e) {
                e.printStackTrace();
            }
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
        return conn;
    }

    /**
     * 关闭数据库链接
     *
     * @return
     */
    public static void closeConn(Connection conn) {
        if (conn != null) {
            try {
                conn.close();  //关闭数据库链接
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
    }
}

上面代码中实现了文章开头列的分布式锁的所有功能，大家可以认真研究下获取锁的方法：lock，释放锁的方法：unlock。

测试用例

package com.itsoku.sql;

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import static com.itsoku.sql.LockUtils.lock;
import static com.itsoku.sql.LockUtils.unlock;

/**
 * 工作10年的前阿里P7分享Java、算法、数据库方面的技术干货！坚信用技术改变命运，让家人过上更体面的生活！
 * 喜欢的请关注公众号：路人甲Java
 */
@Slf4j
public class LockUtilsTest {

    //测试重复获取和重复释放
    @Test
    public void test1() throws Exception {
        String lock_key = "key1";
        for (int i = 0; i < 10; i++) {
            lock(lock_key, 10000L, 1000);
        }
        for (int i = 0; i < 9; i++) {
            unlock(lock_key);
        }
    }

    //获取之后不释放，超时之后被thread1获取
    @Test
    public void test2() throws Exception {
        String lock_key = "key2";
        lock(lock_key, 5000L, 1000);
        Thread thread1 = new Thread(() -> {
            try {
                try {
                    lock(lock_key, 5000L, 7000);
                } finally {
                    unlock(lock_key);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
        thread1.setName("thread1");
        thread1.start();
        thread1.join();
    }
}

test1方法测试了重入锁的效果。

test2测试了主线程获取锁之后一直未释放，持有锁超时之后被thread1获取到了。

留给大家一个问题

上面分布式锁还需要考虑一个问题：比如A机会获取了key1的锁，并设置持有锁的超时时间为10秒，但是获取锁之后，执行了一段业务操作，业务操作耗时超过10秒了，此时机器B去获取锁时可以获取成功的，此时会导致A、B两个机器都获取锁成功了，都在执行业务操作，这种情况应该怎么处理？大家可以思考一下然后留言，我们一起讨论一下。

MySQL-MyCAT分布式架构【代码】【图】

MySQL-MyCAT分布式架构MyCAT基础架构准备主机名ip地址sql0110.0.1.110sql0210.0.1.120#环境准备：两台虚拟机 sql01 sql02 每台创建四个mysql实例：3307 3308 3309 3310#创建相关目录初始化数据 [root@sql01 ~]# mkdir /data/33{07..10}/data -p [root@sql01 ~]# mysqld --initialize-insecure --user=mysql --datadir=/data/3307/data --basedir=/application/mysql [root@sql01 ~]# mysqld --initialize-insecure --user=mysql...

spring与quartz整合实现分布式动态创建，删除，改变执行时间定时任务(mysql数据库)【代码】【图】

背景：因为在项目中用到了定时任务，当时想到了spring的quartz，写完发现费了很大功夫，光是整合就花了一上午，其中最大的问题就是版本问题，项目中用的是spring3.2.8的版本，查阅发现，3.0以上的版本需要使用quartz2.X以上版本，我就去官网下载了2.1.7的quartz，结果发现jar包与spring冲突，最后使用了quartz1.6.0版本。spring与quartz整合第一步需要导jar包，这个在百度搜下quartz的jar，下载一个第二步：分布式定时任务，是基于...

mysql分布式中间件cobar

Cobar的分布式主要是通过将表放入不同的库来实现: 1.Cobar支持将一张表水平拆分成多份分别放入不同的库来实现表的水平拆分 2.Cobar也支持将不同的表放入不同的库 3.多数情况下，用户会将以上两种方式混合使用 4.Cobar不支持将一张表，例如test表拆分成test_1, test_2, test_3.....放在同一个库中，必须将拆分后的表分别放入不同的库来实现分布式环境规划:IP 数据库表192.168.1.247...

解决mysql 允许执行 XA RECOVER语句(atomikos 解决分布式事务报错)

当前访问mysql的账号root缺少系统权限，执行以下sql语句即可GRANT XA_RECOVER_ADMIN ON *.* TO root@‘%‘ ; 查看mysql当前访问用户的权限 show grants for 用户名@‘主机地址（%代表任意ip）‘ 参考：https://blog.csdn.net/qq_43028054/article/details/93138503mysql权限介绍链接：https://blog.csdn.net/qq_20387013/article/details/89875332 原文：https://www.cnblogs.com/jsfh/p/13404223.html

mycat分布式mysql中间件（数据库切分概述）

mysql数据库切分前言通过MySQLReplication功能所实现的扩展总是会受到数据库大小的限制，一旦数据库过于庞大，尤其是当写入过于频繁，很难由一台主机支撑的时候，我们还是会面临到扩展瓶颈。这时候，我们就必须许找其他技术手段来解决这个瓶颈，那就是我们这一章所要介绍恶的数据切分技术。何谓数据切分可能很多读者朋友在网上或者杂志上面都已经多次见到关于数据切分的相关文章了，只不过在有些文章中称之为数据的Sharding。其实...

【源码】java 高并发、高性能、分布式 java后台框架 springmvc整合mybatis框架源码 bootstrap html5 mysql oracl【图】

获取【下载地址】? ?QQ: 313596790? ?【免费支持更新】A 代码生成器(开发利器);全部是源码??? ?增删改查的处理类，service层，mybatis的xml，SQL( mysql? ?和oracle)脚本,? ?jsp页面都生成? ?就不用写搬砖的代码了,生成的放到项目里，可以直接运行B 阿里巴巴数据库连接池druid;??数据库连接池??阿里巴巴的 druid。Druid在监控、可扩展性、稳定性和性能方面都有明显的优势C 安全权限框架shiro ;??Shiro 是一个用 Java 语言实现的框...

【分布式锁的演化】“超卖场景”，MySQL分布式锁篇【代码】【图】

前言之前的文章中通过电商场景中秒杀的例子和大家分享了单体架构中锁的使用方式，但是现在很多应用系统都是相当庞大的，很多应用系统都是微服务的架构体系，那么在这种跨jvm的场景下，我们又该如何去解决并发。单体应用锁的局限性在进入实战之前简单和大家粗略聊一下互联网系统中的架构演进。在互联网系统发展之初，消耗资源比较小，用户量也比较小，我们只部署一个tomcat应用就可以满足需求。一个tomcat我们可以看做是一个jvm的进...

MySql数据库实现分布式的主从结构【图】

最近学习了关于使用MySql数据的实现主动结构的原理，在以前的并发访问低的场景一下，一般一台性能高的服务器作为一个MySql数据，就可以满足业务的增删改查场景，但是随着网络用户的增加当出现高并发，高QPS的情况下，一台MySql就很难支撑这种场景了，根据现在的分布式处理架构，处理在使用Redis这种高效的缓存数据库外，其实也可以针对数据库端进行分布式处理，也就是原来和Redis相同，使用分布式主从架构，通过Master 和 Slave 实...

Spring整合Quartz定时任务在集群、分布式系统中的应用(Mysql数据库环境)【代码】【图】

转载：http://www.cnblogs.com/jiafuwei/p/6145280.html单个Quartz实例能给予你很好的Job调度能力，但它不能满足典型的企业需求，如可伸缩性、高可靠性满足。假如你需要故障转移的能力并能运行日益增多的 Job，Quartz集群势必成为你应用的一部分了。使用 Quartz 的集群能力可以更好的支持你的业务需求，并且即使是其中一台机器在最糟的时间崩溃了也能确保所有的 Job 得到执行。Quartz 中集群如何工作一个 Quartz 集群中的每个节点是...

有生之年系列----MySQL分布式集群之MyCAT调优初探（四）

这是有生之年系列的填坑_(:з」∠)_ 前作第一篇:http://blog.itpub.net/29510932/viewspace-1664499/ 前作第二篇:http://blog.itpub.net/29510932/viewspace-1667814/ 前作第三篇:http://blog.itpub.net/29510932/viewspace-1678591/ MyCAT基准测试:http://blog.itpub.net/29510932/viewspace-1726924/和http://blog.itpub.net/29510932/viewspace-1717783/ --------------------------------------------------------------------...

MYSQL - 技术教程分类

MySQL 教程 MySQL 安装 MySQL 管理 MySQL PHP 语法 MySQL 连接 MySQL 创建数据库 MySQL 删除数据库 MySQL 选择数据库 MySQL 数据类型 MySQL 创建数据表 MySQL 删除数据表 MySQL 插入数据 MySQL 查询数据 MySQL WHERE 子句 MySQL UPDATE 更新 MySQL DELETE 语句 MySQL LIKE 子句 MySQL UNION MySQL 排序 MySQL 分组 MySQL 连接的使用 MySQL NULL 值处理 MySQL 事务 MySQL ALTER命令 MySQL 索引 MySQL 临时表 MySQL 复制表 MySQL 元数据 MySQL 序列使用 MySQL 处理重复数据 MySQL 及 SQL 注入 MySQL 导出数据 MySQL 导入数据 MySQL 函数 MySQL 运算符 mysql 全部

MYSQL - 最热教程

sql分组取最大记录方法 mysql如何设置默认值 mysql创建数据表时指定默认值教程 MySQL 5.7 的初始化操作（root初始密码...Oracle中合并数据集(多行变一行)mysql 相同内容的字段合并为一条的方法解决mysql设置时区时的错误Unknown or ...解决ubuntu下mysql的'Access denied fo...ubuntu系统中MysqlERROR1045(28000)报错...mysql数据库设置不区分大小写

首页 / MYSQL / Mysql高手系列 - 第26篇：聊聊如何使用mysql实现分布式锁

Mysql高手系列 - 第26篇：聊聊如何使用mysql实现分布式锁

内容导读

内容图文

分布式锁的功能

预备技能：乐观锁

使用mysql实现分布式锁

建表

分布式锁工具类：

测试用例

留给大家一个问题

更多优质文章

内容总结

内容备注

内容手机端

【Mysql高手系列 - 第26篇：聊聊如何使用mysql实现分布式锁】教程文章相关的互联网学习教程文章

MySQL-MyCAT分布式架构【代码】【图】

spring与quartz整合实现分布式动态创建，删除，改变执行时间定时任务(mysql数据库)【代码】【图】

mysql分布式中间件cobar

解决mysql 允许执行 XA RECOVER语句(atomikos 解决分布式事务报错)

mycat分布式mysql中间件（数据库切分概述）

【源码】java 高并发、高性能、分布式 java后台框架 springmvc整合mybatis框架源码 bootstrap html5 mysql oracl【图】

【分布式锁的演化】“超卖场景”，MySQL分布式锁篇【代码】【图】

MySql数据库实现分布式的主从结构【图】

Spring整合Quartz定时任务在集群、分布式系统中的应用(Mysql数据库环境)【代码】【图】

有生之年系列----MySQL分布式集群之MyCAT调优初探（四）

年薪30W+面试清单：JVM\MySQL\设计模式\分布式\微服务【图】

入手30w年薪的面试清单：JVM\MySQL\设计模式\分布式\微服务【图】

BATJ年薪30W+面试清单：JVM\MySQL\设计模式\分布式\微服务【图】

基于Hadoop-0.20.2的完全分布式集群搭建详细过程

Hadoop伪分布式运行【图】

MYSQL - 相关标签

分布式 - 相关标签

MYSQL - 技术教程分类

MYSQL - 最新教程

MYSQL - 最热教程