最近在看分散式鎖的資料，看了 Josial L的《Redis in Action》的分散式鎖的章節。實現思路是利用springcloud結合redis實現分散式鎖。

注意：這篇文章有問題，請看這一篇http://blog.csdn.net/forezp/article/details/70305336

一、簡介

一般來說，對資料進行加鎖時，程式先通過acquire獲取鎖來對資料進行排他訪問，然後對資料進行一些列的操作，最後需要釋放鎖。Redis 本身用 watch命令進行了加鎖，這個鎖是樂觀鎖。使用 watch命令對於頻繁訪問的鍵會引起效能的問題。

二、redis命令介紹

• SETNX命令（SET if Not eXists）

當且僅當 key 不存在，將 key 的值設為 value ，並返回1；若給定的 key 已經存在，則 SETNX 不做任何動作，並返回0。

• SETEX命令

設定超時時間

• GET命令

返回 key 所關聯的字串值，如果 key 不存在那麼返回特殊值 nil 。

• DEL命令

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

三、實現思路

由於redis的setnx命令天生就適合用來實現鎖的功能，這個命令只有在鍵不存在的情況下為鍵設定值。獲取鎖之後，其他程式再設定值就會失敗，即獲取不到鎖。獲取鎖失敗。只需不斷的嘗試獲取鎖，直到成功獲取鎖，或者到設定的超時時間為止。

另外為了防治死鎖，即某個程式獲取鎖之後，程式出錯，沒有釋放，其他程式無法獲取鎖，從而導致整個分散式系統無法獲取鎖而導致一系列問題，甚至導致系統無法正常執行。這時需要給鎖設定一個超時時間，即setex命令，鎖超時後，從而其它程式就可以獲取鎖了。

四、編碼實現

本文采用springboot結合redis 取實現的，所以你需要裝一個redis。

1. 首先引入建立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.建立一個鎖類

/**
 * 全域性鎖，包括鎖的名稱
 * Created by fangzhipeng on 2017/4/1.
 */
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;

/**
 * Created by fangzhipeng on 2017/4/1.
 */
@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();
}

五、注意點

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

六、參考資料

Josiah.L 《reids in action》

掃碼關注公眾號有驚喜

（轉載本站文章請註明作者和出處 方誌朋的部落格）