Redis實現分散式鎖的五種方法總結

這篇文章為大家帶來了關於Redis的相關知識，在分散式架構中，我們同樣會遇到資料共享操作問題，使用Redis來解決分散式架構中的資料一致性問題，下面一起來看一下，希望對大家有幫助。

推薦學習：Redis影片教學

在單體應用程式中，如果我們對共享資料不進行加鎖操作，會出現資料一致性問題，我們的解決方法通常是加鎖。

在分散式架構中，我們同樣會遇到資料共享操作問題，而本文章使用Redis來解決分散式架構中的資料一致性問題。

1. 單機資料一致性

單機資料一致性架構如下圖所示：多個可客戶存取同一個伺服器，連接同一個資料庫。

場景描述：客戶端模擬購買商品流程，在Redis中設定庫存總數剩100個，多個客戶端同時並發購買。

@RestController
public class IndexController1 {

    @Autowired
    StringRedisTemplate template;

    @RequestMapping("/buy1")
    public String index(){
        // Redis中存有goods:001号商品，数量为100
        String result = template.opsForValue().get("goods:001");
        // 获取到剩余商品数
        int total = result == null ? 0 : Integer.parseInt(result);
        if( total > 0 ){
            // 剩余商品数大于0 ，则进行扣减
            int realTotal = total -1;
            // 将商品数回写数据库
            template.opsForValue().set("goods:001",String.valueOf(realTotal));
            System.out.println("购买商品成功，库存还剩："+realTotal +"件， 服务端口为8001");
            return "购买商品成功，库存还剩："+realTotal +"件， 服务端口为8001";
        }else{
            System.out.println("购买商品失败，服务端口为8001");
        }
        return "购买商品失败，服务端口为8001";
    }
}

使用Jmeter模擬高並發場景，測試結果如下：

##測試結果出現多個用戶購買相同商品，發生了數據不一致問題！

解決方法：單體應用的情況下，對並發的操作進行加鎖操作，保證對資料的操作具有原子性

• synchronized
• #ReentrantLock

@RestController
public class IndexController2 {
// 使用ReentrantLock锁解决单体应用的并发问题
Lock lock = new ReentrantLock();

@Autowired
StringRedisTemplate template;

@RequestMapping("/buy2")
public String index() {

    lock.lock();
    try {
        String result = template.opsForValue().get("goods:001");
        int total = result == null ? 0 : Integer.parseInt(result);
        if (total > 0) {
            int realTotal = total - 1;
            template.opsForValue().set("goods:001", String.valueOf(realTotal));
            System.out.println("购买商品成功，库存还剩：" + realTotal + "件， 服务端口为8001");
            return "购买商品成功，库存还剩：" + realTotal + "件， 服务端口为8001";
        } else {
            System.out.println("购买商品失败，服务端口为8001");
        }
    } catch (Exception e) {
        lock.unlock();
    } finally {
        lock.unlock();
    }
    return "购买商品失败，服务端口为8001";
}
}

#2. 分散式資料一致性

上面解決了單體應用的資料一致性問題，但如果是分散式架構部署呢，架構如下：

提供兩個服務，連接埠分別為

8001、8002，連接同一個Redis服務，在服務前面有一台Nginx作為負載平衡

兩台服務程式碼相同，只是連接埠不同

將

8001、8002兩個服務啟動，每個服務還是用ReentrantLock加鎖，用Jmeter做並發測試，發現會出現資料一致性問題！

3. Redis實作分散式鎖定

3.1 方式一

取消單機鎖定，以下使用

redis#的set指令來實作分散式加鎖

SET KEY VALUE [EX seconds] [PX milliseconds] [NX|XX]

EX seconds 設定指定的到期時間(以秒為單位)
• PX milliseconds 設定指定的到期時間(以毫秒為單位)
• NX 僅在鍵不存在時設定鍵
• #XX 只有在鍵已存在時才設定

• @RestController
  public class IndexController4 {
  
      // Redis分布式锁的key
      public static final String REDIS_LOCK = "good_lock";
  
      @Autowired
      StringRedisTemplate template;
  
      @RequestMapping("/buy4")
      public String index(){
  
          // 每个人进来先要进行加锁，key值为"good_lock"，value随机生成
          String value = UUID.randomUUID().toString().replace("-","");
          try{
              // 加锁
              Boolean flag = template.opsForValue().setIfAbsent(REDIS_LOCK, value);
              // 加锁失败
              if(!flag){
                  return "抢锁失败！";
              }
              System.out.println( value+ " 抢锁成功");
              String result = template.opsForValue().get("goods:001");
              int total = result == null ? 0 : Integer.parseInt(result);
              if (total > 0) {
                  int realTotal = total - 1;
                  template.opsForValue().set("goods:001", String.valueOf(realTotal));
                  // 如果在抢到所之后，删除锁之前，发生了异常，锁就无法被释放，
                  // 释放锁操作不能在此操作，要在finally处理
  				// template.delete(REDIS_LOCK);
                  System.out.println("购买商品成功，库存还剩：" + realTotal + "件， 服务端口为8001");
                  return "购买商品成功，库存还剩：" + realTotal + "件， 服务端口为8001";
              } else {
                  System.out.println("购买商品失败，服务端口为8001");
              }
              return "购买商品失败，服务端口为8001";
          }finally {
              // 释放锁
              template.delete(REDIS_LOCK);
          }
      }
  }

上面的程式碼，可以解決分散式架構中資料一致性問題。但再仔細想想，還是會有問題，下面要改進。

3.2 方式二（改進方式一）

在上面的程式碼中，如果程式在運作期間，部署了微服務

jar套件的機器突然掛了，程式碼層面根本就沒有走到finally程式碼區塊，也就是說在宕機前，鎖並沒有被刪除掉，這樣的話，就沒辦法保證解鎖

所以，這裡需要對這個

key加一個過期時間，Redis中設定過期時間有兩種方法：

• template.expire(REDIS_LOCK,10, TimeUnit.SECONDS)
• template.opsForValue().setIfAbsent(REDIS_LOCK, value,10L,TimeUnit.SECONDS)

第一種方法需要單獨的一行程式碼，且並沒有與加鎖放在同一步操作，所以不具備原子性，也會出問題

第二種方法在加鎖的同時就進行了設定過期時間，所有沒有問題，這裡採用這種方式

調整下程式碼，在加鎖的同時，設定過期時間：

// 为key加一个过期时间，其余代码不变
Boolean flag = template.opsForValue().setIfAbsent(REDIS_LOCK,value,10L,TimeUnit.SECONDS);

這種方式解決了因服務突然宕機而無法釋放鎖的問題。但再仔細想想，還是會有問題，下面要改進。

3.3 方式三（改進方式二）

方式二設定了

key的過期時間，解決了key無法刪除的問題，但問題又來了

上面设置了key的过期时间为10秒，如果业务逻辑比较复杂，需要调用其他微服务，处理时间需要15秒（模拟场

景，别较真），而当10秒钟过去之后，这个key就过期了，其他请求就又可以设置这个key，此时如果耗时15秒

的请求处理完了，回来继续执行程序，就会把别人设置的key给删除了，这是个很严重的问题！

所以，谁上的锁，谁才能删除

@RestController
public class IndexController6 {

    public static final String REDIS_LOCK = "good_lock";

    @Autowired
    StringRedisTemplate template;

    @RequestMapping("/buy6")
    public String index(){

        // 每个人进来先要进行加锁，key值为"good_lock"
        String value = UUID.randomUUID().toString().replace("-","");
        try{
            // 为key加一个过期时间
            Boolean flag = template.opsForValue().setIfAbsent(REDIS_LOCK, value,10L,TimeUnit.SECONDS);

            // 加锁失败
            if(!flag){
                return "抢锁失败！";
            }
            System.out.println( value+ " 抢锁成功");
            String result = template.opsForValue().get("goods:001");
            int total = result == null ? 0 : Integer.parseInt(result);
            if (total > 0) {
                // 如果在此处需要调用其他微服务，处理时间较长。。。
                int realTotal = total - 1;
                template.opsForValue().set("goods:001", String.valueOf(realTotal));
                System.out.println("购买商品成功，库存还剩：" + realTotal + "件， 服务端口为8001");
                return "购买商品成功，库存还剩：" + realTotal + "件， 服务端口为8001";
            } else {
                System.out.println("购买商品失败，服务端口为8001");
            }
            return "购买商品失败，服务端口为8001";
        }finally {
            // 谁加的锁，谁才能删除！！！！
            if(template.opsForValue().get(REDIS_LOCK).equals(value)){
                template.delete(REDIS_LOCK);
            }
        }
    }
}

这种方式解决了因服务处理时间太长而释放了别人锁的问题。这样就没问题了吗？

3.4 方式四（改进方式三）

在上面方式三下，规定了谁上的锁，谁才能删除，但finally快的判断和del删除操作不是原子操作，并发的时候也会出问题，并发嘛，就是要保证数据的一致性，保证数据的一致性，最好要保证对数据的操作具有原子性。

在Redis的set命令介绍中，最后推荐Lua脚本进行锁的删除，地址

@RestController
public class IndexController7 {

    public static final String REDIS_LOCK = "good_lock";

    @Autowired
    StringRedisTemplate template;

    @RequestMapping("/buy7")
    public String index(){

        // 每个人进来先要进行加锁，key值为"good_lock"
        String value = UUID.randomUUID().toString().replace("-","");
        try{
            // 为key加一个过期时间
            Boolean flag = template.opsForValue().setIfAbsent(REDIS_LOCK, value,10L,TimeUnit.SECONDS);
            // 加锁失败
            if(!flag){
                return "抢锁失败！";
            }
            System.out.println( value+ " 抢锁成功");
            String result = template.opsForValue().get("goods:001");
            int total = result == null ? 0 : Integer.parseInt(result);
            if (total > 0) {
                // 如果在此处需要调用其他微服务，处理时间较长。。。
                int realTotal = total - 1;
                template.opsForValue().set("goods:001", String.valueOf(realTotal));
                System.out.println("购买商品成功，库存还剩：" + realTotal + "件， 服务端口为8001");
                return "购买商品成功，库存还剩：" + realTotal + "件， 服务端口为8001";
            } else {
                System.out.println("购买商品失败，服务端口为8001");
            }
            return "购买商品失败，服务端口为8001";
        }finally {
            // 谁加的锁，谁才能删除，使用Lua脚本，进行锁的删除

            Jedis jedis = null;
            try{
                jedis = RedisUtils.getJedis();

                String script = "if redis.call('get',KEYS[1]) == ARGV[1] " +
                        "then " +
                        "return redis.call('del',KEYS[1]) " +
                        "else " +
                        "   return 0 " +
                        "end";

                Object eval = jedis.eval(script, Collections.singletonList(REDIS_LOCK), Collections.singletonList(value));
                if("1".equals(eval.toString())){
                    System.out.println("-----del redis lock ok....");
                }else{
                    System.out.println("-----del redis lock error ....");
                }
            }catch (Exception e){

            }finally {
                if(null != jedis){
                    jedis.close();
                }
            }
        }
    }
}

3.5 方式五（改进方式四）

在方式四下，规定了谁上的锁，谁才能删除，并且解决了删除操作没有原子性问题。但还没有考虑缓存续命，以及Redis集群部署下，异步复制造成的锁丢失：主节点没来得及把刚刚set进来这条数据给从节点，就挂了。所以直接上RedLock的Redisson落地实现。

@RestController
public class IndexController8 {

    public static final String REDIS_LOCK = "good_lock";

    @Autowired
    StringRedisTemplate template;

    @Autowired
    Redisson redisson;

    @RequestMapping("/buy8")
    public String index(){

        RLock lock = redisson.getLock(REDIS_LOCK);
        lock.lock();

        // 每个人进来先要进行加锁，key值为"good_lock"
        String value = UUID.randomUUID().toString().replace("-","");
        try{
            String result = template.opsForValue().get("goods:001");
            int total = result == null ? 0 : Integer.parseInt(result);
            if (total > 0) {
                // 如果在此处需要调用其他微服务，处理时间较长。。。
                int realTotal = total - 1;
                template.opsForValue().set("goods:001", String.valueOf(realTotal));
                System.out.println("购买商品成功，库存还剩：" + realTotal + "件， 服务端口为8001");
                return "购买商品成功，库存还剩：" + realTotal + "件， 服务端口为8001";
            } else {
                System.out.println("购买商品失败，服务端口为8001");
            }
            return "购买商品失败，服务端口为8001";
        }finally {
            if(lock.isLocked() && lock.isHeldByCurrentThread()){
                lock.unlock();
            }
        }
    }
}

推荐学习：Redis视频教程

以上是Redis實現分散式鎖的五種方法總結的詳細內容。更多資訊請關注PHP中文網其他相關文章！