[[357636]]

目前的項目單體結構的基本上已經沒有了，大多是分布式集群或者是微服務這些。既然是多臺服務器。就免不了資源的共享問題。既然是資源共享就免不了并發的問題。針對這些問題，redis也給出了一個很好的解決方案，那就是分布式鎖。這篇文章主要是針對為什么需要使用分布式鎖這個話題來展開討論的。

前一段時間在群里有個兄弟問，既然分布式鎖能解決大部分生產問題，那么java為我們提供的那些鎖有什么用呢?直接使用分布式鎖不就結了嘛。針對這個問題我想了很多，一開始是在網上找找看看有沒有類似的回答。后來想了想。想要解決這個問題，還需要從本質上來分析。

OK，開始上車出發。

一、前言

既然是分布式鎖，這就說明服務器不是一臺，可能是很多臺。我們使用一個案例，來一步一步說明。假設某網站有一個秒殺商品，一看還有100件，于是陜西、江蘇、西藏等地的人都看到了這個活動，于是開始進行瘋狂秒殺。假設這個秒殺商品的數量值保存在一個redis數據庫中。

但是不同地區的用戶使用不同的服務器進行秒殺。這樣就形成了一個集群訪問的方式。

方式我們使用Springboot來整合redis。

二、項目搭建準備

(1)添加pom依賴

<dependency> 
            <groupId>org.springframework.boot</groupId> 
            <artifactId>spring-boot-starter-web</artifactId> 
        </dependency> 
        <dependency> 
            <groupId>org.springframework.boot</groupId> 
            <artifactId>spring-boot-starter-test</artifactId> 
            <scope>test</scope> 
        </dependency> 
        <dependency> 
            <groupId>org.springframework.boot</groupId> 
            <artifactId>spring-boot-starter-data-redis</artifactId> 
        </dependency> 
        <dependency> 
            <groupId>org.apache.commons</groupId> 
            <artifactId>commons-pool2</artifactId> 
        </dependency> 

(2)添加屬性配置

# Redis數據庫索引（默認為0） 
spring.redis.database=0   
# Redis服務器地址 
spring.redis.host=localhost 
# Redis服務器連接端口 
spring.redis.port=6379   
# Redis服務器連接密碼（默認為空） 
spring.redis.password= 
# 連接池最大連接數（使用負值表示沒有限制） 默認 8 
spring.redis.lettuce.pool.max-active=8 
# 連接池最大阻塞等待時間（使用負值表示沒有限制） 默認 -1 
spring.redis.lettuce.pool.max-wait=-1 
# 連接池中的最大空閑連接 默認 8 
spring.redis.lettuce.pool.max-idle=8 
# 連接池中的最小空閑連接 默認 0 
spring.redis.lettuce.pool.min-idle=0 

(3)新建config包，創建RedisConfig類

@Configuration 
public class RedisConfig { 
    @Bean 
    public RedisTemplate<String, Serializable>  
            redisTemplate(LettuceConnectionFactory connectionFactory) { 
        RedisTemplate<String, Serializable> redisTemplate = new RedisTemplate<>(); 
        redisTemplate.setKeySerializer(new StringRedisSerializer()); 
        redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer()); 
        redisTemplate.setConnectionFactory(connectionFactory); 
        return redisTemplate; 
    } 
} 

(4)新建controller，創建Mycontroller類

@RestController 
public class MyController { 
    @Autowired 
    private StringRedisTemplate stringRedisTemplate; 
    @GetMapping("/test") 
    public String deduceGoods(){ 
        int goods =Integer.parseInt(stringRedisTemplate.opsForValue().get("goods")); 
        int realGoods = goods-1; 
        if(goods>0){ 
            stringRedisTemplate.opsForValue().set("goods",realGoods+""); 
            return "你已經成功秒殺商品，此時還剩余：" + realGoods + "件"; 
        }else{ 
            return "商品已經售罄，歡迎下次活動"; 
        } 
    } 
} 

很簡單的一個整合教程。這個端口是8080，我們復制一份這個項目，把端口改成8090，并且以nginx作負載均衡搭建集群。現在環境我們已經整理好了。下面我們就開始進行分析。

三、為什么需要分布式鎖

階段一：采用原生方式

我們使用多個線程訪問8080這個端口。因為沒有加鎖，此時肯定會出現并發問題。因此我們可能會想到，既然這個goods是一個共享資源，而且是多線程訪問的，就立馬能想到java中的各種鎖了，最有名的就是synchronized。所以我們不如對上面的代碼進行優化。

階段二：使用synchronized加鎖

此時我們對代碼修改一下：

@RestController 
public class MyController { 
    @Autowired 
    private StringRedisTemplate stringRedisTemplate; 
    @GetMapping("/test") 
    public String deduceGoods(){ 
        synchronized (this){ 
            int goods =Integer.parseInt(stringRedisTemplate.opsForValue().get("goods")); 
            int realGoods = goods-1; 
            if(goods>0){ 
                stringRedisTemplate.opsForValue().set("goods",realGoods+""); 
                return "你已經成功秒殺商品，此時還剩余：" + realGoods + "件"; 
            }else{ 
                return "商品已經售罄，歡迎下次活動"; 
            } 
        } 
 
    } 
} 

看到沒，現在我們使用synchronized關鍵字加上鎖，這樣多個線程并發訪問的時候就不會出現數據不一致等各種問題了。這種方式在單體結構下的確有用。目前的項目單體結構的很少，一般都是集群方式的。此時的synchronized就不再起作用了。為什么synchronized不起作用了呢?

我們采用集群的方式去訪問秒殺商品(nginx為我們做了負載均衡)。就會看到數據不一致的現象。也就是說synchronized關鍵字的作用域其實是一個進程，在這個進程下面的所有線程都能夠進行加鎖。但是多進程就不行了。對于秒殺商品來說，這個值是固定的。但是每個地區都可能有一臺服務器。這樣不同地區服務器不一樣，地址不一樣，進程也不一樣。因此synchronized無法保證數據的一致性。

階段三：分布式鎖

上面synchronized關鍵字無法保證多進程的鎖機制，為了解決這個問題，我們可以使用redis分布式鎖。現在我們把代碼再進行修改一下：

@RestController 
public class MyController { 
    @Autowired 
    private StringRedisTemplate stringRedisTemplate; 
    @GetMapping("/test") 
    public String deduceGoods(){ 
      Boolean result = stringRedisTemplate.opsForValue().setIfAbsent("lock","馮冬冬"); 
      if(!result){ 
           return "其他人正在秒殺，無法進入"; 
      } 
      int goods =Integer.parseInt(stringRedisTemplate.opsForValue().get("goods")); 
      int realGoods = goods-1; 
      if(goods>0){ 
          stringRedisTemplate.opsForValue().set("goods",realGoods+""); 
          System.out.println("你已經成功秒殺商品，此時還剩余：" + realGoods + "件"); 
      }else{ 
          System.out.println("商品已經售罄，歡迎下次活動"); 
      } 
      stringRedisTemplate.delete("lock"); 
      return "success"; 
    } 
} 

就是這么簡單，我們只是加了一句話，然后進行判斷了一下。其實setIfAbsent方法的作用就是redis中的setnx。意思是如果當前key已經存在了，就不做任何操作了，返回false。如果當前key不存在，那我們就可以操作。最后別忘了釋放這個key，這樣別人就可以再進來實時秒殺操作。

當然這里只是給出一個最基本的案例，其實分布式鎖實現起來步驟還是比較多的，而且里面很多坑也沒有給出。我們隨便解決幾個：

階段四：分布式鎖優化

(1)第一個坑：秒殺商品出現異常，最終無法釋放lock分布式鎖

  public String deduceGoods() throws Exception{ 
     Boolean result = stringRedisTemplate.opsForValue().setIfAbsent("lock","馮冬冬"); 
     if(!result){ 
        return "其他人正在秒殺，無法進入"; 
     } 
     try { 
        int goods =Integer.parseInt(stringRedisTemplate.opsForValue().get("goods")); 
        int realGoods = goods-1; 
        if(goods>0){ 
           stringRedisTemplate.opsForValue().set("goods",realGoods+""); 
           System.out.println("你已經成功秒殺商品，此時還剩余：" + realGoods + "件"); 
       }else{ 
           System.out.println("商品已經售罄，歡迎下次活動"); 
       } 
    }finally { 
       stringRedisTemplate.delete("lock"); 
    } 
    return "success"; 
} 

此時我們加一個try和finally語句就可以了。最終一定要刪除lock。

(2)第二個坑：秒殺商品時間太久，其他用戶等不及

public String deduceGoods() throws Exception{ 
    stringRedisTemplate.expire("lock",10, TimeUnit.MILLISECONDS); 
    Boolean result = stringRedisTemplate.opsForValue().setIfAbsent("lock","馮冬冬"); 
    if(!result){ 
       return "其他人正在秒殺，無法進入"; 
    } 
    try { 
       int goods =Integer.parseInt(stringRedisTemplate.opsForValue().get("goods")); 
       int realGoods = goods-1; 
       if(goods>0){ 
           stringRedisTemplate.opsForValue().set("goods",realGoods+""); 
           System.out.println("你已經成功秒殺商品，此時還剩余：" + realGoods + "件"); 
       }else{ 
           System.out.println("商品已經售罄，歡迎下次活動"); 
       } 
    }finally { 
       stringRedisTemplate.delete("lock"); 
    } 
    return "success"; 
} 

給其添加一個過期時間，也就是說如果10毫秒內沒有秒殺成功，就表示秒殺失敗，換下一個用戶。

(3)第三個坑：高并發場景下，秒殺時間太久，鎖永久失效問題

我們剛剛設置的鎖過期時間是10毫秒，如果一個用戶秒殺時間是15毫秒，這也就意味著他可能還沒秒殺成功，就有其他用戶進來了。當這種情況過多時，就可能有大量用戶還沒秒殺成功其他大量用戶就進來了。有可能其他用戶提前刪除了lock，但是當前用戶還沒有秒殺成功。最終造成數據的不一致。看看如何解決：

public String deduceGoods() throws Exception{ 
    String user = UUID.randomUUID().toString(); 
    stringRedisTemplate.expire("lock",10, TimeUnit.MILLISECONDS); 
    Boolean result = stringRedisTemplate.opsForValue().setIfAbsent("lock",user); 
    if(!result){ 
       return "其他人正在秒殺，無法進入"; 
    } 
    try { 
       int goods =Integer.parseInt(stringRedisTemplate.opsForValue().get("goods")); 
       int realGoods = goods-1; 
       if(goods>0){ 
           stringRedisTemplate.opsForValue().set("goods",realGoods+""); 
           System.out.println("你已經成功秒殺商品，此時還剩余：" + realGoods + "件"); 
       }else{ 
           System.out.println("商品已經售罄，歡迎下次活動"); 
       } 
    }finally { 
       if(user.equals(stringRedisTemplate.opsForValue().get("lock"))){ 
                 stringRedisTemplate.delete("lock"); 
       } 
    } 
    return "success"; 
} 

也就是說，我們在刪除lock的時候判斷是不是當前的線程，如果是那就刪除，如果不是那就不刪除，這樣就算別的線程進來也不會亂刪lock，造成混亂。

OK，到目前為止基本上把分布式鎖的緣由介紹了一遍。對于分布式鎖redisson完成的相當出色，下篇文章也將圍著繞Redisson來介紹一下分布式如何實現，以及其中的原理。

本文轉載自微信公眾號「愚公要移山」，可以通過以下二維碼關注。轉載本文請聯系愚公要移山公眾號。