死磕 java同步系列之mysql分佈式鎖

時間 2019-12-22

標籤 java 同步系列 mysql 分佈式欄目 Java 简体版

原文原文鏈接

（手機橫屏看源碼更方便）java

問題

（1）什麼是分佈式鎖？mysql

（2）爲何須要分佈式鎖？redis

（3）mysql如何實現分佈式鎖？spring

（4）mysql分佈式鎖的優勢和缺點？sql

簡介

隨着併發量的不斷增長，單機的服務早晚要向多節點或者微服務進化，這時候原來單機模式下使用的synchronized或者ReentrantLock將再也不適用，咱們迫切地須要一種分佈式環境下保證線程安全的解決方案，今天咱們一塊兒來學習一下mysql分佈式鎖如何實現分佈式線程安全。數據庫

基礎知識

mysql中提供了兩個函數——get_lock('key', timeout)和release_lock('key')——來實現分佈式鎖，能夠根據key來加鎖，這是一個字符串，能夠設置超時時間（單位：秒），當調用release_lock('key')或者客戶端斷線的時候釋放鎖。安全

它們的使用方法以下：springboot

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

get_lock('user_1', 10)若是10秒以內獲取到鎖則返回1，不然返回0；session

release_lock('user_1')若是該鎖是當前客戶端持有的則返回1，若是該鎖被其它客戶端持有着則返回0，若是該鎖沒有被任何客戶端持有則返回null；mybatis

多客戶端案例

爲了便於舉例【本篇文章由公衆號「彤哥讀源碼」原創，請支持原創，謝謝！】，這裏的超時時間所有設置爲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