SpringCloud 分佈式鎖

A 簡單實現：

1、簡介 通常來講，對數據進行加鎖時，程序先經過acquire獲取鎖來對數據進行排他訪問，而後對數據進行一些列的操做，最後須要釋放鎖。 Redis 自己用 watch命令進行了加鎖，這個鎖是樂觀鎖。使用 watch命令對於頻繁訪問的鍵會引發性能的問題。

2、redis命令介紹 SETNX命令（SET if Not eXists） 當且僅當 key 不存在，將 key 的值設爲 value ，並返回1；若給定的 key 已經存在，則 SETNX 不作任何動做，並返回0。

SETEX命令 設置超時時間

GET命令 返回 key 所關聯的字符串值，若是 key 不存在那麼返回特殊值 null 。

DEL命令 刪除給定的一個或多個 key ,不存在的 key 會被忽略。

3、實現思路 因爲redis的setnx命令天生就適合用來實現鎖的功能，這個命令只有在鍵不存在的狀況下爲鍵設置值。獲取鎖以後，其餘程序再設置值就會失敗，即獲取不到鎖。獲取鎖失敗。只需不斷的嘗試獲取鎖，直到成功獲取鎖，或者到設置的超時時間爲止。

另外爲了防治死鎖，即某個程序獲取鎖以後，程序出錯，沒有釋放，其餘程序沒法獲取鎖，從而致使整個分佈式系統沒法獲取鎖而致使一系列問題，甚至致使系統沒法正常運行。這時須要給鎖設置一個超時時間，即setex命令，鎖超時後，從而其它程序就能夠獲取鎖了。

4、編碼實現 本文采用springboot結合redis 取實現的，因此你須要裝一個redis。

1. 首先pom.xml引入建立springboot工程，引入redis 。

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

        <!-- 開啓web-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <!-- redis-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>

2.建立一個鎖類

/**
 * 全局鎖，包括鎖的名稱
 */
public class Lock {
    private String name;
    private String value;

    public Lock(String name, String value) {
        this.name = name;
        this.value = value;
    }

    public String getName() {
        return name;
    }

    public String getValue() {
        return value;
    }

}

3.建立分佈式鎖的具體方法,思路已經說清楚了，代碼註釋也寫好了，就不講解了。

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 */
@Component
public class DistributedLockHandler {

    private static final Logger logger = LoggerFactory.getLogger(DistributedLockHandler.class);
    private final static long LOCK_EXPIRE = 30 * 1000L;//單個業務持有鎖的時間30s，防止死鎖
    private final static long LOCK_TRY_INTERVAL = 30L;//默認30ms嘗試一次
    private final static long LOCK_TRY_TIMEOUT = 20 * 1000L;//默認嘗試20s

    @Autowired
    private StringRedisTemplate template;

    /**
     * 嘗試獲取全局鎖
     *
     * @param lock 鎖的名稱
     * @return true 獲取成功，false獲取失敗
     */
    public boolean tryLock(Lock lock) {
        return getLock(lock, LOCK_TRY_TIMEOUT, LOCK_TRY_INTERVAL, LOCK_EXPIRE);
    }

    /**
     * 嘗試獲取全局鎖
     *
     * @param lock    鎖的名稱
     * @param timeout 獲取超時時間 單位ms
     * @return true 獲取成功，false獲取失敗
     */
    public boolean tryLock(Lock lock, long timeout) {
        return getLock(lock, timeout, LOCK_TRY_INTERVAL, LOCK_EXPIRE);
    }

    /**
     * 嘗試獲取全局鎖
     *
     * @param lock        鎖的名稱
     * @param timeout     獲取鎖的超時時間
     * @param tryInterval 多少毫秒嘗試獲取一次
     * @return true 獲取成功，false獲取失敗
     */
    public boolean tryLock(Lock lock, long timeout, long tryInterval) {
        return getLock(lock, timeout, tryInterval, LOCK_EXPIRE);
    }

    /**
     * 嘗試獲取全局鎖
     *
     * @param lock           鎖的名稱
     * @param timeout        獲取鎖的超時時間
     * @param tryInterval    多少毫秒嘗試獲取一次
     * @param lockExpireTime 鎖的過時
     * @return true 獲取成功，false獲取失敗
     */
    public boolean tryLock(Lock lock, long timeout, long tryInterval, long lockExpireTime) {
        return getLock(lock, timeout, tryInterval, lockExpireTime);
    }


    /**
     * 操做redis獲取全局鎖
     *
     * @param lock           鎖的名稱
     * @param timeout        獲取的超時時間
     * @param tryInterval    多少ms嘗試一次
     * @param lockExpireTime 獲取成功後鎖的過時時間
     * @return true 獲取成功，false獲取失敗
     */
    public boolean getLock(Lock lock, long timeout, long tryInterval, long lockExpireTime) {
        try {
            if (StringUtils.isEmpty(lock.getName()) || StringUtils.isEmpty(lock.getValue())) {
                return false;
            }
            long startTime = System.currentTimeMillis();
            do{
                if (!template.hasKey(lock.getName())) {
                    ValueOperations<String, String> ops = template.opsForValue();
                    ops.set(lock.getName(), lock.getValue(), lockExpireTime, TimeUnit.MILLISECONDS);
                    return true;
                } else {//存在鎖
                    logger.debug("lock is exist!！！");
                }
                if (System.currentTimeMillis() - startTime > timeout) {//嘗試超過了設定值以後直接跳出循環
                    return false;
                }
                Thread.sleep(tryInterval);
            }
            while (template.hasKey(lock.getName())) ;
        } catch (InterruptedException e) {
            logger.error(e.getMessage());
            return false;
        }
        return false;
    }

    /**
     * 釋放鎖
     */
    public void releaseLock(Lock lock) {
        if (!StringUtils.isEmpty(lock.getName())) {
            template.delete(lock.getName());
        }
    }

}

4.用法：

@Autowired
DistributedLockHandler distributedLockHandler;
Lock lock=new Lock("lockk","sssssssss);
if(distributedLockHandler.tryLock(lock){
    doSomething();
    distributedLockHandler.releaseLock();
}

5、注意點

在使用全局鎖時爲了防止死鎖採用 setex命令，這種命令須要根據具體的業務具體設置鎖的超時時間。另一個就是鎖的粒度性。好比在redis實戰中有個案列，爲了實現買賣市場交易的功能，把整個交易市場都鎖住了，致使了性能不足的狀況，改進方案只對買賣的商品進行加鎖而不是整個市場。

B 優化實現:

setNx是一個耗時操做，由於它須要查詢這個鍵是否存在，就算redis的百萬的qps，在高併發的場景下，這種操做也是有問題的。關於redis實現分佈式鎖，redis官方推薦使用redlock。

1、redlock簡介 在不一樣進程須要互斥地訪問共享資源時，分佈式鎖是一種很是有用的技術手段。實現高效的分佈式鎖有三個屬性須要考慮：

安全屬性：互斥，無論何時，只有一個客戶端持有鎖 效率屬性A:不會死鎖 效率屬性B：容錯，只要大多數redis節點可以正常工做，客戶端端都能獲取和釋放鎖。

Redlock是redis官方提出的實現分佈式鎖管理器的算法，這個算法會比通常的普通方法更加安全可靠。

2、怎麼用java使用 redlock 在pom文件引入redis和redisson依賴：

<!-- redis-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>
        <!-- redisson-->
        <dependency>
            <groupId>org.redisson</groupId>
            <artifactId>redisson</artifactId>
            <version>3.3.2</version>
        </dependency>

AquiredLockWorker接口類，，主要是用於獲取鎖後須要處理的邏輯：

/**
 * 獲取鎖後須要處理的邏輯
 */
public interface AquiredLockWorker<T> {
     T invokeAfterLockAquire() throws Exception;
}

DistributedLocker 獲取鎖管理類：

/**
 * 獲取鎖管理類
 */
public interface DistributedLocker {

     /**
      * 獲取鎖
      * @param resourceName  鎖的名稱
      * @param worker 獲取鎖後的處理類
      * @param <T>
      * @return 處理完具體的業務邏輯要返回的數據
      * @throws UnableToAquireLockException
      * @throws Exception
      */
     <T> T lock(String resourceName, AquiredLockWorker<T> worker) throws UnableToAquireLockException, Exception;

     <T> T lock(String resourceName, AquiredLockWorker<T> worker, int lockTime) throws UnableToAquireLockException, Exception;

}

UnableToAquireLockException ，不能獲取鎖的異常類：

/**
 * 異常類
 */
public class UnableToAquireLockException extends RuntimeException {

    public UnableToAquireLockException() {
    }

    public UnableToAquireLockException(String message) {
        super(message);
    }

    public UnableToAquireLockException(String message, Throwable cause) {
        super(message, cause);
    }
}

RedissonConnector 鏈接類：

/**
 * 獲取RedissonClient鏈接類
 */
@Component
public class RedissonConnector {
    RedissonClient redisson;
    @PostConstruct
    public void init(){
        redisson = Redisson.create();
    }

    public RedissonClient getClient(){
        return redisson;
    }

}

RedisLocker 類，實現了DistributedLocker：

import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.concurrent.TimeUnit;

@Component
public class RedisLocker  implements DistributedLocker{

    private final static String LOCKER_PREFIX = "lock:";

    @Autowired
    RedissonConnector redissonConnector;
    @Override
    public <T> T lock(String resourceName, AquiredLockWorker<T> worker) throws InterruptedException, UnableToAquireLockException, Exception {

        return lock(resourceName, worker, 100);
    }

    @Override
    public <T> T lock(String resourceName, AquiredLockWorker<T> worker, int lockTime) throws UnableToAquireLockException, Exception {
        RedissonClient redisson= redissonConnector.getClient();
        RLock lock = redisson.getLock(LOCKER_PREFIX + resourceName);
      // Wait for 100 seconds seconds and automatically unlock it after lockTime seconds
        boolean success = lock.tryLock(100, lockTime, TimeUnit.SECONDS);
        if (success) {
            try {
                return worker.invokeAfterLockAquire();
            } finally {
                lock.unlock();
            }
        }
        throw new UnableToAquireLockException();
    }
}

測試類：

@Autowired
    RedisLocker distributedLocker;
    @RequestMapping(value = "/redlock")
    public String testRedlock() throws Exception{

        CountDownLatch startSignal = new CountDownLatch(1);
        CountDownLatch doneSignal = new CountDownLatch(5);
        for (int i = 0; i < 5; ++i) { // create and start threads
            new Thread(new Worker(startSignal, doneSignal)).start();
        }
        startSignal.countDown(); // let all threads proceed
        doneSignal.await();
        System.out.println("All processors done. Shutdown connection");
        return "redlock";
    }

     class Worker implements Runnable {
        private final CountDownLatch startSignal;
        private final CountDownLatch doneSignal;

        Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {
            this.startSignal = startSignal;
            this.doneSignal = doneSignal;
        }

        public void run() {
            try {
                startSignal.await();
                distributedLocker.lock("test",new AquiredLockWorker<Object>() {

                    @Override
                    public Object invokeAfterLockAquire() {
                        doTask();
                        return null;
                    }

                });
            }catch (Exception e){

            }
        }

        void doTask() {
            System.out.println(Thread.currentThread().getName() + " start");
            Random random = new Random();
            int _int = random.nextInt(200);
            System.out.println(Thread.currentThread().getName() + " sleep " + _int + "millis");
            try {
                Thread.sleep(_int);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName() + " end");
            doneSignal.countDown();
        }
    }