快捷導(dǎo)航

利用consul在spring boot中實(shí)現(xiàn)分布式鎖場(chǎng)景分析

更新時(shí)間：2021年09月15日 16:39:21 作者：wenwuxianren

這篇文章通過場(chǎng)景分析給大家介紹如何利用consul在spring boot中實(shí)現(xiàn)簡(jiǎn)單的分布式鎖功能，代碼簡(jiǎn)單易懂，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友參考下吧

因?yàn)樵陧?xiàng)目實(shí)際過程中所采用的是微服務(wù)架構(gòu)，考慮到承載量基本每個(gè)相同業(yè)務(wù)的服務(wù)都是多節(jié)點(diǎn)部署，所以針對(duì)某些資源的訪問就不得不用到用到分布式鎖了。

這里列舉一個(gè)最簡(jiǎn)單的場(chǎng)景，假如有一個(gè)智能售貨機(jī)，由于機(jī)器本身的原因不能同一臺(tái)機(jī)器不能同時(shí)出兩個(gè)商品，這就要求在在出貨流程前針對(duì)同一臺(tái)機(jī)器在同一時(shí)刻出現(xiàn)并發(fā)創(chuàng)建訂單時(shí)只能有一筆訂單創(chuàng)建成功，但是訂單服務(wù)是多節(jié)點(diǎn)部署的，所以就不得不用到分布式鎖了。

以上只是一種簡(jiǎn)單的業(yè)務(wù)場(chǎng)景，在各種大型互聯(lián)網(wǎng)實(shí)際應(yīng)用中，需要分布式鎖的業(yè)務(wù)場(chǎng)景會(huì)更多，綜合比較了業(yè)界基于各種中間件來實(shí)現(xiàn)的分布式鎖方案，然后結(jié)合實(shí)際業(yè)務(wù)最終決定采用consul來實(shí)現(xiàn)，因?yàn)槲覀兊捻?xiàng)目中采用了consul做注冊(cè)中心，并且consul天生可以保證一致性(這點(diǎn)類似zk)，當(dāng)然zk也能實(shí)現(xiàn)分布式鎖，但是這里不對(duì)這點(diǎn)做過多討論。

redis雖然也能實(shí)現(xiàn)分布式鎖，但是可能因?yàn)閳?chǎng)景比較復(fù)雜，如果redis采用cluster部署的話，如果某一主節(jié)點(diǎn)出現(xiàn)故障的話，有一定幾率會(huì)出現(xiàn)腦裂現(xiàn)象，這樣就可能會(huì)讓競(jìng)爭(zhēng)者在并發(fā)時(shí)同時(shí)獲得到鎖，這樣可能會(huì)破壞掉后面的業(yè)務(wù)，當(dāng)然出現(xiàn)這種情況的概率很低，但是也不能完全排除，因?yàn)閞edis的根本不能保證強(qiáng)一致性導(dǎo)致的。

好了，這里說的最簡(jiǎn)單的分布式鎖的意思是，多個(gè)競(jìng)爭(zhēng)者同一時(shí)間并發(fā)去獲得鎖時(shí)，獲取失敗的就直接返回了，獲取成功的繼續(xù)后續(xù)的流程，然后在合適的時(shí)間釋放鎖，并且為鎖加了超時(shí)時(shí)間，防止獲得到鎖的進(jìn)程或線程在未來得及釋放鎖時(shí)自己掛掉了，導(dǎo)致資源處于一直被鎖定的狀態(tài)無(wú)法得到釋放。主要的實(shí)現(xiàn)邏輯就是這樣，如果有人想實(shí)現(xiàn)獲得鎖失

敗的競(jìng)爭(zhēng)者一直繼續(xù)嘗試獲得，可以基于該示例進(jìn)行修改，加上自旋邏輯就OK。

以下是鎖實(shí)現(xiàn)代碼：

package com.lyb.consullock;

import com.ecwid.consul.v1.ConsulClient;
import com.ecwid.consul.v1.agent.model.NewCheck;
import com.ecwid.consul.v1.kv.model.PutParams;
import com.ecwid.consul.v1.session.model.NewSession;
import com.ecwid.consul.v1.session.model.Session;
import lombok.Data;


import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;


public class DistributedLock{
    private ConsulClient consulClient;

    /**
     * 構(gòu)造函數(shù)
     * @param consulHost 注冊(cè)consul的client或服務(wù)端的Ip或主機(jī)名，或域名
     * @param consulPort 端口號(hào)
     */
    public DistributedLock(String consulHost,int consulPort){
        consulClient = new ConsulClient(consulHost,consulPort);
    }

    /**
     * 獲得鎖的方法
     * @param lockName 競(jìng)爭(zhēng)的資源名
     * @param ttlSeconds 鎖的超時(shí)時(shí)間，超過該時(shí)間自動(dòng)釋放
     * @return
     */
    public LockContext getLock(String lockName,int ttlSeconds){
        LockContext lockContext = new LockContext();
        if(ttlSeconds<10 || ttlSeconds > 86400) ttlSeconds = 60;
        String sessionId = createSession(lockName,ttlSeconds);
        boolean success = lock(lockName,sessionId);
        if(success == false){
            consulClient.sessionDestroy(sessionId,null);
            lockContext.setGetLock(false);

            return lockContext;
        }

        lockContext.setSession(sessionId);
        lockContext.setGetLock(true);

        return lockContext;
    }

    /**
     * 釋放鎖
     * @param sessionID
     */
    public void releaseLock(String sessionID){
        consulClient.sessionDestroy(sessionID,null);
    }

    private String createSession(String lockName,int ttlSeconds){
        NewCheck check = new NewCheck();
        check.setId("check "+lockName);
        check.setName(check.getId());
        check.setTtl(ttlSeconds+"s"); //該值和session ttl共同決定決定鎖定時(shí)長(zhǎng)
        check.setTimeout("10s");
        consulClient.agentCheckRegister(check);
        consulClient.agentCheckPass(check.getId());

        NewSession session = new NewSession();
        session.setBehavior(Session.Behavior.RELEASE);
        session.setName("session "+lockName);
        session.setLockDelay(1);
        session.setTtl(ttlSeconds + "s"); //和check ttl共同決定鎖時(shí)長(zhǎng)
        List<String> checks = new ArrayList<>();
        checks.add(check.getId());
        session.setChecks(checks);
        String sessionId = consulClient.sessionCreate(session,null).getValue();

        return sessionId;
    }

    private boolean lock(String lockName,String sessionId){
        PutParams putParams = new PutParams();
        putParams.setAcquireSession(sessionId);

        boolean isSuccess = consulClient.setKVValue(lockName,"lock:"+ LocalDateTime.now(),putParams).getValue();

        return isSuccess;
    }

    /**
     * 競(jìng)爭(zhēng)鎖時(shí)返回的對(duì)象
     */
    @Data
    public class LockContext{
        /**
         * 獲得鎖成功返回該值，比便后面用該值來釋放鎖
         */
        private String session;
        /**
         * 是否獲得到鎖
         */
        private boolean isGetLock;
    }
}

pom文件

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.6.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.lyb</groupId>
    <artifactId>consul-lock</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>consul-lock</name>
    <description>Demo project for Spring Boot</description>

    <properties>
        <java.version>1.8</java.version>
        <spring-cloud.version>Greenwich.SR2</spring-cloud.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-consul-discovery</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.8</version>
            <optional>true</optional>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>${spring-cloud.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

測(cè)試代碼：

package com.lyb.consullock;

import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

@RunWith(SpringRunner.class)
@SpringBootTest
public class ConsulLockApplicationTests {
    @Autowired
    private ServiceConfig serviceConfig;
    @Test
    public void lockSameResourer() {
        //針對(duì)相同資源在同一時(shí)刻只有一個(gè)線程會(huì)獲得鎖
        ExecutorService threadPool = Executors.newFixedThreadPool(10);
        for (int a=0;a<20;a++){
            threadPool.submit(
                    () -> {
                        for (int i = 0;i < 100; i++) {
                            DistributedLock lock = new DistributedLock(
                                    serviceConfig.getConsulRegisterHost(),
                                    serviceConfig.getConsulRegisterPort());

                            DistributedLock.LockContext lockContext = lock.getLock("test lock", 10);
                            if (lockContext.isGetLock()) {
                                System.out.println(Thread.currentThread().getName() + "獲得了鎖");
                                try {
                                    TimeUnit.SECONDS.sleep(1);
                                    lock.releaseLock(lockContext.getSession());
                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                }
                            }else {
                                //System.out.println(Thread.currentThread().getName() + "沒有獲得鎖");
                            }
                        }
                    });
        }

        try {
            TimeUnit.MINUTES.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void lockDiffResource(){
        //針對(duì)不通的資源所有線程都應(yīng)該能獲得鎖
        ExecutorService threadPool = Executors.newFixedThreadPool(10);
        for (int a=0;a<20;a++){
            threadPool.submit(
                    () -> {
                        for (int i = 0;i < 100; i++) {
                            DistributedLock lock = new DistributedLock(
                                    serviceConfig.getConsulRegisterHost(),
                                    serviceConfig.getConsulRegisterPort());

                            DistributedLock.LockContext lockContext = lock.getLock("test lock"+Thread.currentThread().getName(), 10);
                            if (lockContext.isGetLock()) {
                                System.out.println(Thread.currentThread().getName() + "獲得了鎖");
                                try {
                                    TimeUnit.SECONDS.sleep(1);
                                    lock.releaseLock(lockContext.getSession());
                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                }
                            }else {
                                //System.out.println(Thread.currentThread().getName() + "沒有獲得鎖");
                                Assert.assertTrue(lockContext.isGetLock());
                            }
                        }
                    });
        }

        try {
            TimeUnit.MINUTES.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

項(xiàng)目路徑：

https://github.com/wenwuxianren/consul-lock

到此這篇關(guān)于利用consul在spring boot中實(shí)現(xiàn)最簡(jiǎn)單的分布式鎖的文章就介紹到這了,更多相關(guān)spring boot分布式鎖內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: