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

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

问题

（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

Java实现

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

定义Locker接口

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

public interface Locker {

    void lock(String key, Runnable command);

}

mysql分布式锁实现

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;

        }

    }

}

LockerMapper.xml

定义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