利用consul在spring boot中实现分布式锁场景分析

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;

     /**

      * 构造函数

      * @param consulHost 注册consul的client或服务端的Ip或主机名，或域名

      * @param consulPort 端口号

      */

     public DistributedLock(String consulHost, int consulPort){

         consulClient = new ConsulClient(consulHost,consulPort);

     }

     /**

      * 获得锁的方法

      * @param lockName 竞争的资源名

      * @param ttlSeconds 锁的超时时间，超过该时间自动释放

      * @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共同决定决定锁定时长

         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共同决定锁时长

         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;

     }

     /**

      * 竞争锁时返回的对象

      */

     @Data

     public class LockContext{

         /**

          * 获得锁成功返回该值，比便后面用该值来释放锁

          */

         private String session;

         /**

          * 是否获得到锁

          */

         private boolean isGetLock;

     }

}

<?xml version= "1.0" encoding= "UTF-8" ?>

<project xmlns= "http://maven.apache.org/POM/4.0.0" xmlns:xsi= "http://HdhCmsTestw3.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>

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() {

         //针对相同资源在同一时刻只有一个线程会获得锁

         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(){

         //针对不通的资源所有线程都应该能获得锁

         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();

         }

     }

}