目前幾乎所有的大型網站及應用都是採用分散式部署的方式，分散式系統開發帶來的優點很多，高可用，高併發，水平擴充套件，分開部署等。但分散式的開發也帶來了一些新問題，有的時候，我們需要保證一個方法在同一時間內只能被同一個執行緒執行。在單機環境中，Java中其實提供了很多併發處理相關的API  ，也就是我們常說的“鎖”(如synchronized,lock)，但是這些API在分散式場景中就無能為力了，也就是說Java沒有提供分散式鎖的功能。

基於分散式鎖的實現有多種方案，常見的有基於資料庫本身的鎖來實現，或者基於zookeeper的API實現，或者是基於快取來實現分散式鎖等等，這些方案都各有可取之處，今天我們介紹的是基於redis的快取實現分散式鎖的方案，大家如果對其他方案有興趣的可以上網搜尋研究。

redis是基於key-value的一種NoSql資料庫，廣泛應用於分散式的應用中，一般用於放置快取資料。安裝的方法也比較簡單，樓主安裝的是windows版本的，選擇最新的zip版，下載完之後直接解壓即可。下載地址：https://github.com/MicrosoftArchive/redis/tags

redis中有一個命令setnx (SET IF NOT EXISTS) , 如果不存在，就設定key，將 key 的值設為 value，當且僅當 key 不存在。若給定的 key 已經存在，則 SETNX 不做任何動作。基於這個特性，我們可以對需要鎖住的物件加上key，這樣，同一時間就只能有一個執行緒擁有這把鎖，從而達到分散式鎖的效果。

Java操作redis需要用到第三方的庫類，所以先在pom.xml中引入依賴。

加入依賴後，做一個redis的工具方法，分別實現的是加鎖和解鎖的功能。

public class RedisLock {

    @Autowired
private StringRedisTemplate redisTemplate;
/**
     * 加鎖
     *
     * @param key
* @param value 當前時間+超時時間
     * @return
*/
public boolean lock(String key, 
 
String value) {
        //相當於setnx命令
if (redisTemplate.opsForValue().setIfAbsent(key, value)) {
            return true;
}
        //下面的這段程式碼是判斷之前加的鎖是否超時，是的話就更新，一定要加這段程式碼
        //不然就有可能出現死鎖。
String currentValue = redisTemplate.opsForValue().get(key);
//如果鎖過期
if (!StringUtils.isEmpty(currentValue)
                && Long.parseLong(currentValue) < System.currentTimeMillis()) {
            //獲取上一個鎖的時間，這段程式碼的判斷是防止多執行緒進入這裡，只會有一個執行緒拿到鎖
String oldValue = redisTemplate.opsForValue().getAndSet(key, value);
            if (!StringUtils.isEmpty(oldValue)
                    && oldValue.equals(currentValue)) {
                return true;
}
        }
        return false;
}

    /**
     * 解鎖
     *
     * @param key
* @param value
*/
public void unLock(String key, String value) {
        try {
            String currentValue = redisTemplate.opsForValue().get(key);
            if (!StringUtils.isEmpty(currentValue)
                    && currentValue.equals(value)) {
                redisTemplate.opsForValue().getOperations().delete(key);
}
        } catch (Exception e) {
            log.error("【redis分散式鎖】 解鎖異常，{}", e);
}
    }
}

現在，我們模擬一個下單的場景，假設有一個秒殺的活動，同一時間有多個執行緒對同一個產品進行訪問，然後分別看看加鎖和沒加鎖的結果來做對比。下面是秒殺的模擬程式碼：

public class SecKillController {

    @Autowired
private SecKillService secKillService;
/**
     * 查詢秒殺活動特價商品的資訊
     * @param productId
* @return
* @throws Exception
     */
@GetMapping("/query/{productId}")
    public String query(@PathVariable String productId) throws Exception{
        return secKillService.querySecKillProductInfo(productId);
}

    /**
     * 秒殺的方法
     * @param productId
* @return
* @throws Exception
     */
@GetMapping("/order/{productId}")
    public String skill(@PathVariable String productId) throws Exception{
        log.info("@skill request ,productId:" +productId);
secKillService.orderProductKill(productId);
        return secKillService.querySecKillProductInfo(productId);
}
}

public class SecKillServiceImpl implements SecKillService {

    private static final int TIME_OUT = 1 * 1000;
@Autowired
private RedisLock redisLock;
    static Map<String, Integer> products;
    static Map<String, Integer> stock;
    static Map<String, String> orders;
    static {
        /**
         * 模擬多個表，商品資訊表，庫存表，秒殺成功訂單表
         */
products = new HashMap<>();
stock = new HashMap<>();
orders = new HashMap<>();
products.put("123", 100000);
stock.put("123", 100000);
}

    /**
     * @param productId  訂單id
     * @return
*/
private String queryMap(String productId) {
        return "限量份數" + products.get(productId)
                + "還剩:" + stock.get(productId) + "份"
+ "該商品成功下單使用者數目："
+ orders.size() + "人";
}

    @Override
public String querySecKillProductInfo(String productId) {
        return this.queryMap(productId);
}

    @Override
public void orderProductKill(String productId) {

        //1.查詢該商品庫存，為0則活動結束
int stockNum = stock.get(productId);
        if (stockNum == 0) {
            throw new RuntimeException("活動結束");
} else {
            //2.下單(模擬不同使用者openid不同)
orders.put(KeyUtil.getUniqueKey(), productId);
//3.減庫存
stockNum = stockNum - 1;
            try {
                Thread.sleep(100);
} catch (InterruptedException e) {
                e.printStackTrace();
}
            stock.put(productId, stockNum);
}
    }
}

先模擬沒加鎖的下單狀態，我們開啟工程後，用Apache ab作為壓測工具來模擬高併發訪問過程

在瀏覽器上訪問查詢後的訂單數量，結果顯示如下：

可以看到，再高併發的訪問環境下，如果我們沒有對訂單做鎖的處理，那麼就可能出現數據的紊亂，導致結果不對應，這顯然不符合我們的需求，下面我們來看看加上redis鎖之後的訪問情況，先把service中的秒殺程式碼加上鎖。

@Override
public void orderProductKill(String productId) {

    //加鎖，保證下面的程式碼單執行緒的訪問
long time = System.currentTimeMillis() + TIME_OUT;
    if (!redisLock.lock(productId, String.valueOf(time))) {
        throw new RuntimeException( "下單失敗");
}

    //1.查詢該商品庫存，為0則活動結束
int stockNum = stock.get(productId);
    if (stockNum == 0) {
        throw new RuntimeException("活動結束");
} else {
        //2.下單(模擬不同使用者openid不同)
orders.put(KeyUtil.getUniqueKey(), productId);
//3.減庫存
stockNum = stockNum - 1;
        try {
            Thread.sleep(100);
} catch (InterruptedException e) {
            e.printStackTrace();
}
        stock.put(productId, stockNum);
}
    //解鎖
redisLock.unLock(productId, String.valueOf(time));
}

 然後再進行同樣的操作

我們可以看到，加上鎖之後的訂單處理數量是正確的，也就是redis鎖是起到了作用的，這是符合我們的需求的。

上面的例子相對比較簡單，因為精力能力有限，樓主沒法給大家展示真正的分散式鎖的實現效果，但從原理上其實是一樣的，都是用redis的setnx命令來加上鎖，保證分散式環境下鎖住的物件只能被一個執行緒訪問，而且從實現方式上來說也比較簡單  (只需要一個命令就行，很深入人心 ) ，因此，redis在分散式鎖的應用中也被廣泛使用。