基于SpringBoot與數據庫表記錄的方式實現分布式鎖

作者：java小當家 2021-03-11 09:53:07

數據庫分布式

今天演示一下最簡單的分布式鎖方案——基于數據庫表記錄的分布式鎖.

[[386855]]

同一進程內的不同線程操作共享資源時，我們只需要對資源加鎖，比如利用JUC下的工具，就可以保證操作的正確性。對JUC不熟悉的同學，可以看看以下的幾篇文章：

淺說Synchronized
Synchronized的優化
JUC基石——Unsafe類

但是，為了高可用，我們的系統總是多副本的，分布在不同的機器上，以上同進程內的鎖機制就不再起作用。為了保證多副本系統對共享資源的訪問，我們引入了分布式鎖。

分布式鎖主要的實現方式有以下幾種：

基于數據庫的，其中又細分為基于數據庫的表記錄、悲觀鎖、樂觀鎖
基于緩存的，比如Redis
基于Zookeeper的

今天演示一下最簡單的分布式鎖方案——基于數據庫表記錄的分布式鎖

主要的原理就是利用數據庫的唯一索引(對數據庫的索引不了解的同學，可以參考我的另外一篇文章mysql索引簡談)

例如，有以下的一張表：

CREATE TABLE `test`.`Untitled`  ( 
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '自增序號', 
  `name` varchar(255) NOT NULL COMMENT '鎖名稱', 
  `survival_time` int(11) NOT NULL COMMENT '存活時間，單位ms', 
  `create_time` timestamp(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '創建時間', 
  `thread_name` varchar(255) NOT NULL COMMENT '線程名稱', 
  PRIMARY KEY (`id`) USING BTREE, 
  UNIQUE INDEX `uk_name`(`name`) USING BTREE 
) ENGINE = InnoDB ROW_FORMAT = Dynamic;

其中name字段加上了唯一索引，多條含有同樣name值的新增操作，數據庫只能保證僅有一個操作成功，其他操作都會被拒絕掉，并且拋出“重復鍵”的錯誤。

那么，當系統1準備獲取分布式鎖時，就嘗試往數據庫中插入一條name="key"的記錄，如果插入成功，則代表獲取鎖成功。其他系統想要獲取分布式鎖，同樣需要往數據庫插入相同name的記錄，當然數據庫會報錯，插入失敗，也就代表著這些系統獲取鎖失敗。當系統1想要釋放掉鎖時，刪除掉此記錄即可。thread_name列可以用來保證只能主動釋放自己創建的鎖。

我們希望實現的分布式鎖有以下的效果：

獲取鎖是阻塞的，獲取不到會一直阻塞
鎖會失效，超過鎖的生存時間后，會自動釋放掉。這一點可以避免某些系統因為宕機而無法主動釋放鎖的問題

大致的流程圖如下：

使用到了以下依賴：

SpringBoot
MyBatis-plus
Lombok

項目的工程目錄為：

其中pom文件用到的依賴：

<dependencies> 
    <dependency> 
        <groupId>org.springframework.boot</groupId> 
        <artifactId>spring-boot-starter-web</artifactId> 
    </dependency> 
 
    <dependency> 
        <groupId>org.projectlombok</groupId> 
        <artifactId>lombok</artifactId> 
        <version>1.18.6</version> 
    </dependency> 
 
    <dependency> 
        <groupId>mysql</groupId> 
        <artifactId>mysql-connector-java</artifactId> 
    </dependency> 
 
    <dependency> 
        <groupId>com.baomidou</groupId> 
        <artifactId>mybatis-plus-boot-starter</artifactId> 
        <version>3.3.1</version> 
    </dependency> 
 
    <dependency> 
        <groupId>com.baomidou</groupId> 
        <artifactId>mybatis-plus-extension</artifactId> 
        <version>3.3.1</version> 
    </dependency> 
 
    <dependency> 
        <groupId>org.springframework.boot</groupId> 
        <artifactId>spring-boot-starter-test</artifactId> 
        <scope>test</scope> 
    </dependency> 
</dependencies>

配置項為：

server: 
  port: 9091 
 
 
spring: 
  datasource: 
    driver-class-name: com.mysql.cj.jdbc.Driver 
    url: jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=UTF-8&useSSL=false&serverTimezone=Asia/Shanghai 
    username: root 
    password: a123 
 
logging: 
  level: 
    root: info

用于映射數據庫字段的實體類為：

package com.yang.lock1.entity; 
 
import com.baomidou.mybatisplus.annotation.IdType; 
import com.baomidou.mybatisplus.annotation.TableField; 
import com.baomidou.mybatisplus.annotation.TableId; 
import com.baomidou.mybatisplus.annotation.TableName; 
import lombok.AllArgsConstructor; 
import lombok.Data; 
import lombok.NoArgsConstructor; 
 
import java.util.Date; 
 
/** 
 * @author qcy 
 * @create 2020/08/25 15:03:47 
 */ 
@Data 
@NoArgsConstructor 
@TableName(value = "t_lock") 
public class Lock { 
 
    /** 
     * 自增序號 
     */ 
    @TableId(value = "id", type = IdType.AUTO) 
    private Integer id; 
 
    /** 
     * 鎖名稱 
     */ 
    private String name; 
 
    /** 
     * 存活時間，單位ms 
     */ 
    private int survivalTime; 
 
    /** 
     * 鎖創建的時間 
     */ 
    private Date createTime; 
 
    /** 
     * 線程名稱 
     */ 
    private String ThreadName; 
}

Dao層：

package com.yang.lock1.dao; 
 
import com.baomidou.mybatisplus.core.mapper.BaseMapper; 
import com.yang.lock1.entity.Lock; 
import org.apache.ibatis.annotations.Mapper; 
 
/** 
 * @author qcy 
 * @create 2020/08/25 15:06:24 
 */ 
@Mapper 
public interface LockDao extends BaseMapper<Lock> { 
}

Service接口層：

package com.yang.lock1.service; 
 
import com.baomidou.mybatisplus.extension.service.IService; 
import com.yang.lock1.entity.Lock; 
 
/** 
 * @author qcy 
 * @create 2020/08/25 15:07:44 
 */ 
public interface LockService extends IService<Lock> { 
 
    /** 
     * 阻塞獲取分布式鎖 
     * 
     * @param name         鎖名稱 
     * @param survivalTime 存活時間 
     */ 
    void lock(String name, int survivalTime); 
 
    /** 
     * 釋放鎖 
     * 
     * @param name 鎖名稱 
     */ 
    public void unLock(String name); 
}

Service實現層：

package com.yang.lock1.service.impl; 
 
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; 
import com.yang.lock1.dao.LockDao; 
import com.yang.lock1.entity.Lock; 
import com.yang.lock1.service.LockService; 
import lombok.extern.slf4j.Slf4j; 
import org.springframework.dao.DuplicateKeyException; 
import org.springframework.stereotype.Service; 
 
import java.util.Date; 
 
/** 
 * @author qcy 
 * @create 2020/08/25 15:08:25 
 */ 
@Slf4j 
@Service 
public class LockServiceImpl extends ServiceImpl<LockDao, Lock> implements LockService { 
 
    @Override 
    public void lock(String name, int survivalTime) { 
        String threadName = "system1-" + Thread.currentThread().getName(); 
        while (true) { 
            Lock lock = this.lambdaQuery().eq(Lock::getName, name).one(); 
            if (lock == null) { 
                //說明無鎖 
                Lock lk = new Lock(); 
                lk.setName(name); 
                lk.setSurvivalTime(survivalTime); 
                lk.setThreadName(threadName); 
                try { 
                    save(lk); 
                    log.info(threadName + "獲取鎖成功"); 
                    return; 
                } catch (DuplicateKeyException e) { 
                    //繼續重試 
                    log.info(threadName + "獲取鎖失敗"); 
                    continue; 
                } 
            } 
 
            //此時有鎖,判斷鎖是否過期 
            Date now = new Date(); 
            Date expireDate = new Date(lock.getCreateTime().getTime() + lock.getSurvivalTime()); 
            if (expireDate.before(now)) { 
                //鎖已經過期 
                boolean result = removeById(lock.getId()); 
                if (result) { 
                    log.info(threadName + "刪除了過期鎖"); 
                } 
 
                //嘗試獲取鎖 
                Lock lk = new Lock(); 
                lk.setName(name); 
                lk.setSurvivalTime(survivalTime); 
                lk.setThreadName(threadName); 
                try { 
                    save(lk); 
                    log.info(threadName + "獲取鎖成功"); 
                    return; 
                } catch (DuplicateKeyException e) { 
                    log.info(threadName + "獲取鎖失敗"); 
                } 
            } 
        } 
 
    } 
 
    @Override 
    public void unLock(String name) { 
        //釋放鎖的時候，需要注意只能釋放自己創建的鎖 
        String threadName = "system1-" + Thread.currentThread().getName(); 
        Lock lock = lambdaQuery().eq(Lock::getName, name).eq(Lock::getThreadName, threadName).one(); 
        if (lock != null) { 
            boolean b = removeById(lock.getId()); 
            if (b) { 
                log.info(threadName + "釋放了鎖"); 
            } else { 
                log.info(threadName + "準備釋放鎖,但鎖過期了,被其他客戶端強制釋放掉了"); 
            } 
        } else { 
            log.info(threadName + "準備釋放鎖,但鎖過期了,被其他客戶端強制釋放掉了"); 
        } 
    } 
 
}

測試類如下：

package com.yang.lock1; 
 
import com.yang.lock1.service.LockService; 
import lombok.extern.slf4j.Slf4j; 
import org.junit.Test; 
import org.junit.runner.RunWith; 
import org.springframework.boot.test.context.SpringBootTest; 
import org.springframework.test.context.junit4.SpringRunner; 
 
import javax.annotation.Resource; 
 
/** 
 * @author qcy 
 * @create 2020/08/25 15:10:54 
 */ 
@Slf4j 
@RunWith(SpringRunner.class) 
@SpringBootTest 
public class Lock1ApplicationTest { 
 
    @Resource 
    LockService lockService; 
 
    @Test 
    public void testLock() { 
        log.info("system1準備獲取鎖"); 
        lockService.lock("key", 6 * 1000); 
        try { 
            //模擬業務耗時 
            Thread.sleep(4 * 1000); 
        } catch (Exception e) { 
            e.printStackTrace(); 
        } finally { 
            lockService.unLock("key"); 
        } 
    } 
 
}