死磕-java同步系列之mysql分布式锁

共计 4626 个字符，预计需要花费 12 分钟才能阅读完成。

（1）什么是分布式锁？

（2）为什么需要分布式锁？

（3）mysql 如何实现分布式锁？

（4）mysql 分布式锁的优点和缺点？

随着并发量的不断增加，单机的服务迟早要向多节点或者微服务进化，这时候原来单机模式下使用的 synchronized 或者 ReentrantLock 将不再适用，我们迫切地需要一种分布式环境下保证线程安全的解决方案，今天我们一起来学习一下 mysql 分布式锁如何实现分布式线程安全。

mysql 中提供了两个函数——get_lock('key', timeout)和 release_lock('key')——来实现分布式锁，可以根据key 来加锁，这是一个字符串，可以设置超时时间（单位：秒），当调用 release_lock('key') 或者 客户端断线 的时候释放锁。

它们的使用方法如下：

mysql> select get_lock('user_1', 10);
    -> 1
mysql> select release_lock('user_1');
    -> 1

get_lock('user_1', 10)如果 10 秒之内获取到锁则返回 1，否则返回 0；

release_lock('user_1')如果该锁是当前客户端持有的则返回 1，如果该锁被其它客户端持有着则返回 0，如果该锁没有被任何客户端持有则返回 null；

为了便于举例【本篇文章由“彤哥读源码”原创，请支持原创，谢谢！】，这里的超时时间全部设置为 0，也就是立即返回。

时刻	客户端 A	客户端 B
1	get_lock(‘user_1’, 0) -> 1	–
2	–	get_lock(‘user_1’, 0) -> 0
3	–	release_lock(‘user_1’, 0) -> 0
4	release_lock(‘user_1’, 0) -> 1	–
5	release_lock(‘user_2’, 0) -> null	–
6	–	get_lock(‘user_1’, 0) -> 1
7	–	release_lock(‘user_1’, 0) -> 1

为了方便快速实现，这里使用 springboot2.1 + mybatis 实现，并且省略 spring 的配置，只列举主要的几个类。

接口中只有一个方法，入参 1 为加锁的 key，入参 2 为执行的命令。

public interface Locker {void lock(String key, Runnable command);
}

mysql 的实现中要注意以下两点：

（1）加锁、释放锁必须在同一个 session（同一个客户端）中，所以这里不能使用 Mapper 接口的方式调用，因为 Mapper 接口有可能会导致不在同一个 session。

（2）可重入性是通过 ThreadLocal 保证的；

@Slf4j
@Component
public class MysqlLocker implements Locker {private static final ThreadLocal<SqlSessionWrapper> localSession = new ThreadLocal<>();

    @Autowired
    private SqlSessionFactory sqlSessionFactory;

    @Override
    public void lock(String key, Runnable command) {
        // 加锁、释放锁必须使用同一个 session
        SqlSessionWrapper sqlSessionWrapper = localSession.get();
        if (sqlSessionWrapper == null) {
            // 第一次获取锁
            localSession.set(new SqlSessionWrapper(sqlSessionFactory.openSession()));
        }
        try {
            //【本篇文章由“彤哥读源码”原创，请支持原创，谢谢！】// - 1 表示没获取到锁一直等待
            if (getLock(key, -1)) {command.run();
            }
        } catch (Exception e) {log.error("lock error", e);
        } finally {releaseLock(key);
        }
    }

    private boolean getLock(String key, long timeout) {Map<String, Object> param = new HashMap<>();
        param.put("key", key);
        param.put("timeout", timeout);
        SqlSessionWrapper sqlSessionWrapper = localSession.get();
        Integer result = sqlSessionWrapper.sqlSession.selectOne("LockerMapper.getLock", param);
        if (result != null && result.intValue() == 1) {
            // 获取到了锁，state 加 1
            sqlSessionWrapper.state++;
            return true;
        }
        return false;
    }

    private boolean releaseLock(String key) {SqlSessionWrapper sqlSessionWrapper = localSession.get();
        Integer result = sqlSessionWrapper.sqlSession.selectOne("LockerMapper.releaseLock", key);
        if (result != null && result.intValue() == 1) {
            // 释放锁成功，state 减 1
            sqlSessionWrapper.state--;
            // 当 state 减为 0 的时候说明当前线程获取的锁全部释放了，则关闭 session 并从 ThreadLocal 中移除
            if (sqlSessionWrapper.state == 0) {sqlSessionWrapper.sqlSession.close();
                localSession.remove();}
            return true;
        }
        return false;
    }

    private static class SqlSessionWrapper {
        int state;
        SqlSession sqlSession;

        public SqlSessionWrapper(SqlSession sqlSession) {
            this.state = 0;
            this.sqlSession = sqlSession;
        }
    }
}

定义 get_lock()、release_lock()的语句。

<?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="LockerMapper">
    <select id="getLock" resultType="integer">
        select get_lock(#{key}, #{timeout});
    </select>

    <select id="releaseLock" resultType="integer">
        select release_lock(#{key})
    </select>
</mapper>

这里启动 1000 个线程，每个线程打印一句话并睡眠 2 秒钟。

@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class MysqlLockerTest {

    @Autowired
    private Locker locker;

    @Test
    public void testMysqlLocker() throws IOException {for (int i = 0; i < 1000; i++) {
            // 多节点测试
            try {Thread.sleep(2000);
            } catch (InterruptedException e) {e.printStackTrace();
            }
            new Thread(()->{locker.lock("lock", ()-> {
                    // 可重入性测试
                    locker.lock("lock", ()-> {System.out.println(String.format("time: %d, threadName: %s", System.currentTimeMillis(), Thread.currentThread().getName()));
                        try {Thread.sleep(2000);
                        } catch (InterruptedException e) {e.printStackTrace();
                        }
                    });
                });
            }, "Thread-"+i).start();}

        System.in.read();}
}

查看运行结果发现每隔 2 秒打印一个线程的信息，说明这个锁是有效的，至于分布式环境下面的验证也很简单，起多个 MysqlLockerTest 实例即可。

time: 1568715905952, threadName: Thread-3
time: 1568715907955, threadName: Thread-4
time: 1568715909966, threadName: Thread-8
time: 1568715911967, threadName: Thread-0
time: 1568715913969, threadName: Thread-1
time: 1568715915972, threadName: Thread-9
time: 1568715917975, threadName: Thread-6
time: 1568715919997, threadName: Thread-5
time: 1568715921999, threadName: Thread-7
time: 1568715924001, threadName: Thread-2

（1）分布式环境下需要使用分布式锁，单机的锁将无法保证线程安全；

（2）mysql 分布式锁是基于 get_lock('key', timeout) 和release_lock('key')两个函数实现的；

（3）mysql 分布式锁是可重入锁；

使用 mysql 分布式锁需要注意些什么呢？

答：必须保证多个服务节点使用的是同一个 mysql 库【本篇文章由“彤哥读源码”原创，请支持原创，谢谢！】。

mysql 分布式锁具有哪些优点？

答：1）方便快捷，因为基本每个服务都会连接数据库，但是不是每个服务都会使用 redis 或者 zookeeper；

2）如果客户端断线了会自动释放锁，不会造成锁一直被占用；

3）mysql 分布式锁是可重入锁，对于旧代码的改造成本低；

mysql 分布式锁具有哪些缺点？

答：1）加锁直接打到数据库，增加了数据库的压力；

2）加锁的线程会占用一个 session，也就是一个连接数，如果并发量大可能会导致正常执行的 sql 语句获取不到连接；

3）服务拆分后如果每个服务使用自己的数据库，则不合适；

4）相对于 redis 或者 zookeeper 分布式锁，效率相对要低一些；

【本篇文章由“彤哥读源码”原创，请支持原创，谢谢！】

问题

简介

基础知识

多客户端案例

Java 实现

定义 Locker 接口

mysql 分布式锁实现

LockerMapper.xml

测试类

运行结果

总结

彩蛋

Just My Socks（注册教程内含优惠码）