加锁了还有并发问题？Redis分布式锁，真的用对了？

加锁了还有并发问题？Redis分布式锁，真的用对了？

作者：丑胖侠二师兄 2021-10-25 09:50:57

新闻

前端

分布式

Redis 新接手的项目，偶尔会出现账不平的问题。之前的技术老大临走时给的解释是：排查了，没找到原因，之后太忙就没再解决，可能是框架的原因……

成都创新互联公司坚持“要么做到，要么别承诺”的工作理念，服务领域包括：做网站、网站制作、企业官网、英文网站、手机端网站、网站推广等服务，满足客户于互联网时代的遂平网站设计、移动媒体设计的需求，帮助企业找到有效的互联网解决方案。努力成为您成熟可靠的网络建设合作伙伴！

新接手的项目，偶尔会出现账不平的问题。之前的技术老大临走时给的解释是：排查了，没找到原因，之后太忙就没再解决，可能是框架的原因……

既然项目交付到手中，这样的问题是必须要解决的。梳理了所有账务处理逻辑，最终找到了原因：数据库并发操作热点账户导致。就这这个问题，来聊一聊分布式系统下基于Redis的分布式锁。顺便也分解一下问题形成原因及解决方案。

原因分析

系统并发量并不高，存在热点账户，但也不至于那么严重。问题的根源在于系统架构设计，人为的制造了并发。场景是这样的：商户批量导入一批数据，系统会进行前置处理，并对账户余额进行增减。

此时，另外一个定时任务，也会对账户进行扫描更新。而且对同一账户的操作分布到各个系统当中，热点账户也就出现了。

针对此问题的解决方案，从架构层面可以考虑将账务系统进行抽离，集中在一个系统中进行处理，所有的数据库事务及执行顺序由账务系统来统筹处理。从技术方面来讲，则可以通过锁机制来对热点账户进行加锁。

本篇文章就针对热点账户基于分布式锁的实现方式进行详细的讲解。

锁的分析

在Java的多线程环境下，通常有几类锁可以使用：

  • JVM内存模型级别的锁，常用的有：synchronized、Lock等；
  • 数据库锁，比如乐观锁，悲观锁等；
  • 分布式锁；

JVM内存级别的锁，可以保证单体服务下线程的安全性，比如多个线程访问/修改一个全局变量。但当系统进行集群部署时，JVM级别的本地锁就无能为力了。

悲观锁与乐观锁

像上述案例中，热点账户就属于分布式系统中的共享资源，我们通常会采用数据库锁或分布式锁来进行解决。

数据库锁，又分为乐观锁和悲观锁。

悲观锁是基于数据库（Mysql的InnoDB）提供的排他锁来实现的。在进行事务操作时，通过select ... for update语句，MySQL会对查询结果集中每行数据都添加排他锁，其他线程对该记录的更新与删除操作都会阻塞。从而达到共享资源的顺序执行（修改）；

乐观锁是相对悲观锁而言的，乐观锁假设数据一般情况不会造成冲突，所以在数据进行提交更新的时候，才会正式对数据的冲突与否进行检测。如果冲突则返回给用户异常信息，让用户决定如何去做。乐观锁适用于读多写少的场景，这样可以提高程序的吞吐量。在乐观锁实现时通常会基于记录状态或添加version版本来进行实现。

悲观锁失效场景

项目中使用了悲观锁，但悲观锁却失效了。这也是使用悲观锁时，常见的误区，下面来分析一下。

正常使用悲观锁的流程：

  • 通过select ... for update锁定记录；
  • 计算新余额，修改金额并存储；
  • 执行完成释放锁；

经常犯错的处理流程：

  • 查询账户余额，计算新余额；
  • 通过select ... for update锁定记录；
  • 修改金额并存储；
  • 执行完成释放锁；

错误的流程中，比如A和B服务查询到的余额都是100，A扣减50，B扣减40，然后A锁定记录，更新数据库为50；A释放锁之后，B锁定记录，更新数据库为60。显然，后者把前者的更新给覆盖掉了。解决的方案就是扩大锁的范围，将锁提前到计算新余额之前。

通常悲观锁对数据库的压力是非常大的，在实践中通常会根据场景使用乐观锁或分布式锁等方式来实现。

下面进入正题，讲讲基于Redis的分布式锁实现。

Redis分布式锁实战演习

这里以Spring Boot、Redis、Lua脚本为例来演示分布式锁的实现。为了简化处理，示例中Redis既承担了分布式锁的功能，也承担了数据库的功能。

场景构建

集群环境下，对同一个账户的金额进行操作，基本步骤：

  • 从数据库读取用户金额；
  • 程序修改金额；
  • 再将最新金额存储到数据库；

下面从最初不加锁，不同步处理，逐步推演出最终的分布式锁。

基础集成及类构建

准备一个不加锁处理的基础业务环境。

首先在Spring Boot项目中引入相关依赖：

  
 
 
 
    
   
  
  
  
  1.  
    2. 
   
  
  
  
  2.  org.springframework.boot 
    3. 
   
  
  
  
  3.  spring-boot-starter-data-redis 
    4. 
   
  
  
  
  4.  
    5. 
   
  
  
  
  5.  
    6. 
   
  
  
  
  6.  org.springframework.boot 
    7. 
   
  
  
  
  7.  spring-boot-starter-web 
    8. 
   
  
  
  
  8.  
    9.

账户对应实体类UserAccount：

  
 
 
 
    
   
  
  
  
  1. public class UserAccount { 
    2. 
   
  
  
  
  2.  
    3. 
   
  
  
  
  3.  //用户ID 
    4. 
   
  
  
  
  4.  private String userId; 
    5. 
   
  
  
  
  5.  //账户内金额 
    6. 
   
  
  
  
  6.  private int amount; 
    7. 
   
  
  
  
  7.  
    8. 
   
  
  
  
  8.  //添加账户金额 
    9. 
   
  
  
  
  9.  public void addAmount(int amount) { 
    10. 
   
  
  
  
  10.   this.amount = this.amount + amount; 
    11. 
   
  
  
  
  11.  } 
    12. 
   
  
  
  
  12.  // 省略构造方法和getter/setter  
    13. 
   
  
  
  
  13. } 
    14.

创建一个线程实现类AccountOperationThread：

  
 
 
 
    
   
  
  
  
  1. public class AccountOperationThread implements Runnable { 
    2. 
   
  
  
  
  2.  
    3. 
   
  
  
  
  3.  private final static Logger logger = LoggerFactory.getLogger(AccountOperationThread.class); 
    4. 
   
  
  
  
  4.  
    5. 
   
  
  
  
  5.  private static final Long RELEASE_SUCCESS = 1L; 
    6. 
   
  
  
  
  6.  
    7. 
   
  
  
  
  7.  private String userId; 
    8. 
   
  
  
  
  8.  
    9. 
   
  
  
  
  9.  private RedisTemplate redisTemplate; 
    10. 
   
  
  
  
  10.  
    11. 
   
  
  
  
  11.  public AccountOperationThread(String userId, RedisTemplate redisTemplate) { 
    12. 
   
  
  
  
  12.   this.userId = userId; 
    13. 
   
  
  
  
  13.   this.redisTemplate = redisTemplate; 
    14. 
   
  
  
  
  14.  } 
    15. 
   
  
  
  
  15.  
    16. 
   
  
  
  
  16.  @Override 
    17. 
   
  
  
  
  17.  public void run() { 
    18. 
   
  
  
  
  18.   noLock(); 
    19. 
   
  
  
  
  19.  } 
    20. 
   
  
  
  
  20.  
    21. 
   
  
  
  
  21.  /** 
    22. 
   
  
  
  
  22.   * 不加锁 
    23. 
   
  
  
  
  23.   */ 
    24. 
   
  
  
  
  24.  private void noLock() { 
    25. 
   
  
  
  
  25.   try { 
    26. 
   
  
  
  
  26.    Random random = new Random(); 
    27. 
   
  
  
  
  27.    // 模拟线程进行业务处理 
    28. 
   
  
  
  
  28.    TimeUnit.MILLISECONDS.sleep(random.nextInt(100) + 1); 
    29. 
   
  
  
  
  29.   } catch (InterruptedException e) { 
    30. 
   
  
  
  
  30.    e.printStackTrace(); 
    31. 
   
  
  
  
  31.   } 
    32. 
   
  
  
  
  32.   //模拟数据库中获取用户账号 
    33. 
   
  
  
  
  33.   UserAccount userAccount = (UserAccount) redisTemplate.opsForValue().get(userId); 
    34. 
   
  
  
  
  34.   // 金额+1 
    35. 
   
  
  
  
  35.   userAccount.addAmount(1); 
    36. 
   
  
  
  
  36.   logger.info(Thread.currentThread().getName() + " : user id : " + userId + " amount : " + userAccount.getAmount()); 
    37. 
   
  
  
  
  37.   //模拟存回数据库 
    38. 
   
  
  
  
  38.   redisTemplate.opsForValue().set(userId, userAccount); 
    39. 
   
  
  
  
  39.  } 
    40. 
   
  
  
  
  40. } 
    41.

其中RedisTemplate的实例化交给了Spring Boot：

  
 
 
 
    
   
  
  
  
  1. @Configuration 
    2. 
   
  
  
  
  2. public class RedisConfig { 
    3. 
   
  
  
  
  3.  
    4. 
   
  
  
  
  4.  @Bean 
    5. 
   
  
  
  
  5.  public RedisTemplate redisTemplate(RedisConnectionFactory redisConnectionFactory) { 
    6. 
   
  
  
  
  6.   RedisTemplate redisTemplate = new RedisTemplate<>(); 
    7. 
   
  
  
  
  7.   redisTemplate.setConnectionFactory(redisConnectionFactory); 
    8. 
   
  
  
  
  8.   Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = 
   
  
  
  
  9.     new Jackson2JsonRedisSerializer<>(Object.class); 
    10. 
   
  
  
  
  10.   ObjectMapper objectMapper = new ObjectMapper(); 
    11. 
   
  
  
  
  11.   objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY); 
    12. 
   
  
  
  
  12.   objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL); 
    13. 
   
  
  
  
  13.   jackson2JsonRedisSerializer.setObjectMapper(objectMapper); 
    14. 
   
  
  
  
  14.   // 设置value的序列化规则和 key的序列化规则 
    15. 
   
  
  
  
  15.   redisTemplate.setValueSerializer(jackson2JsonRedisSerializer); 
    16. 
   
  
  
  
  16.   redisTemplate.setKeySerializer(new StringRedisSerializer()); 
    17. 
   
  
  
  
  17.   redisTemplate.afterPropertiesSet(); 
    18. 
   
  
  
  
  18.   return redisTemplate; 
    19. 
   
  
  
  
  19.  } 
    20. 
   
  
  
  
  20. } 
    21. 
  
 
 
 
    最后，再准备一个TestController来进行触发多线程的运行：
      
 
 
 
      
   
  
  
  
    1. @RestController 
      2. 
   
  
  
  
    2. public class TestController { 
      3. 
   
  
  
  
    3.  
      4. 
   
  
  
  
    4.  private final static Logger logger = LoggerFactory.getLogger(TestController.class); 
      5. 
   
  
  
  
    5.  
      6. 
   
  
  
  
    6.  private static ExecutorService executorService = Executors.newFixedThreadPool(10); 
      7. 
   
  
  
  
    7.  
      8. 
   
  
  
  
    8.  @Autowired 
      9. 
   
  
  
  
    9.  private RedisTemplate redisTemplate; 
      10. 
   
  
  
  
    10.  
      11. 
   
  
  
  
    11.  @GetMapping("/test") 
      12. 
   
  
  
  
    12.  public String test() throws InterruptedException { 
      13. 
   
  
  
  
    13.   // 初始化用户user_001到Redis，账户金额为0 
      14. 
   
  
  
  
    14.   redisTemplate.opsForValue().set("user_001", new UserAccount("user_001", 0)); 
      15. 
   
  
  
  
    15.   // 开启10个线程进行同步测试，每个线程为账户增加1元 
      16. 
   
  
  
  
    16.   for (int i = 0; i < 10; i++) { 
      17. 
   
  
  
  
    17.    logger.info("创建线程i=" + i); 
      18. 
   
  
  
  
    18.    executorService.execute(new AccountOperationThread("user_001", redisTemplate)); 
      19. 
   
  
  
  
    19.   } 
      20. 
   
  
  
  
    20.  
      21. 
   
  
  
  
    21.   // 主线程休眠1秒等待线程跑完 
      22. 
   
  
  
  
    22.   TimeUnit.MILLISECONDS.sleep(1000); 
      23. 
   
  
  
  
    23.   // 查询Redis中的user_001账户 
      24. 
   
  
  
  
    24.   UserAccount userAccount = (UserAccount) redisTemplate.opsForValue().get("user_001"); 
      25. 
   
  
  
  
    25.   logger.info("user id : " + userAccount.getUserId() + " amount : " + userAccount.getAmount()); 
      26. 
   
  
  
  
    26.   return "success"; 
      27. 
   
  
  
  
    27.  } 
      28. 
   
  
  
  
    28. } 
      29. 
  
 
 
 
    执行上述程序，正常来说10个线程，每个线程加1，结果应该是10。但多执行几次，会发现，结果变化很大，基本上都要比10小。
      
 
 
 
      
   
  
  
  
    1. [pool-1-thread-5] c.s.redis.thread.AccountOperationThread  : pool-1-thread-5 : user id : user_001 amount : 1 
      2. 
   
  
  
  
    2. [pool-1-thread-4] c.s.redis.thread.AccountOperationThread  : pool-1-thread-4 : user id : user_001 amount : 1 
      3. 
   
  
  
  
    3. [pool-1-thread-3] c.s.redis.thread.AccountOperationThread  : pool-1-thread-3 : user id : user_001 amount : 1 
      4. 
   
  
  
  
    4. [pool-1-thread-1] c.s.redis.thread.AccountOperationThread  : pool-1-thread-1 : user id : user_001 amount : 1 
      5. 
   
  
  
  
    5. [pool-1-thread-1] c.s.redis.thread.AccountOperationThread  : pool-1-thread-1 : user id : user_001 amount : 2 
      6. 
   
  
  
  
    6. [pool-1-thread-2] c.s.redis.thread.AccountOperationThread  : pool-1-thread-2 : user id : user_001 amount : 2 
      7. 
   
  
  
  
    7. [pool-1-thread-5] c.s.redis.thread.AccountOperationThread  : pool-1-thread-5 : user id : user_001 amount : 2 
      8. 
   
  
  
  
    8. [pool-1-thread-4] c.s.redis.thread.AccountOperationThread  : pool-1-thread-4 : user id : user_001 amount : 3 
      9. 
   
  
  
  
    9. [pool-1-thread-1] c.s.redis.thread.AccountOperationThread  : pool-1-thread-1 : user id : user_001 amount : 4 
      10. 
   
  
  
  
    10. [pool-1-thread-3] c.s.redis.thread.AccountOperationThread  : pool-1-thread-3 : user id : user_001 amount : 5 
      11. 
   
  
  
  
    11. [nio-8080-exec-1] c.s.redis.controller.TestController      : user id : user_001 amount : 5 
      12. 
  
 
 
 
    以上述日志为例，前四个线程都将值改为1，也就是后面三个线程都将前面的修改进行了覆盖，导致最终结果不是10，只有5。这显然是有问题的。
    Redis同步锁实现
    

    针对上面的情况，在同一个JVM当中，我们可以通过线程加锁来完成。但在分布式环境下，JVM级别的锁是没办法实现的，这里可以采用Redis同步锁实现。
    基本思路：第一个线程进入时，在Redis中进记录，当后续线程过来请求时，判断Redis是否存在该记录，如果存在则说明处于锁定状态，进行等待或返回。如果不存在，则进行后续业务处理。
      
 
 
 
      
   
  
  
  
    1.  /** 
      2. 
   
  
  
  
    2.  * 1.抢占资源时判断是否被锁。 
      3. 
   
  
  
  
    3.  * 2.如未锁则抢占成功且加锁，否则等待锁释放。 
      4. 
   
  
  
  
    4.  * 3.业务完成后释放锁,让给其它线程。 
      5. 
   
  
  
  
    5.  * 
       
      6. 
   
  
  
  
    6.  * 该方案并未解决同步问题，原因：线程获得锁和加锁的过程，并非原子性操作，可能会导致线程A获得锁，还未加锁时，线程B也获得了锁。 
      7. 
   
  
  
  
    7.  */ 
      8. 
   
  
  
  
    8. private void redisLock() { 
      9. 
   
  
  
  
    9.  Random random = new Random(); 
      10. 
   
  
  
  
    10.  try { 
      11. 
   
  
  
  
    11.   TimeUnit.MILLISECONDS.sleep(random.nextInt(1000) + 1); 
      12. 
   
  
  
  
    12.  } catch (InterruptedException e) { 
      13. 
   
  
  
  
    13.   e.printStackTrace(); 
      14. 
   
  
  
  
    14.  } 
      15. 
   
  
  
  
    15.  while (true) { 
      16. 
   
  
  
  
    16.   Object lock = redisTemplate.opsForValue().get(userId + ":syn"); 
      17. 
   
  
  
  
    17.   if (lock == null) { 
      18. 
   
  
  
  
    18.    // 获得锁 -> 加锁 -> 跳出循环 
      19. 
   
  
  
  
    19.    logger.info(Thread.currentThread().getName() + ":获得锁"); 
      20. 
   
  
  
  
    20.    redisTemplate.opsForValue().set(userId + ":syn", "lock"); 
      21. 
   
  
  
  
    21.    break; 
      22. 
   
  
  
  
    22.   } 
      23. 
   
  
  
  
    23.   try { 
      24. 
   
  
  
  
    24.    // 等待500毫秒重试获得锁 
      25. 
   
  
  
  
    25.    TimeUnit.MILLISECONDS.sleep(500); 
      26. 
   
  
  
  
    26.   } catch (InterruptedException e) { 
      27. 
   
  
  
  
    27.    e.printStackTrace(); 
      28. 
   
  
  
  
    28.   } 
      29. 
   
  
  
  
    29.  } 
      30. 
   
  
  
  
    30.  try { 
      31. 
   
  
  
  
    31.   //模拟数据库中获取用户账号 
      32. 
   
  
  
  
    32.   UserAccount userAccount = (UserAccount) redisTemplate.opsForValue().get(userId); 
      33. 
   
  
  
  
    33.   if (userAccount != null) { 
      34. 
   
  
  
  
    34.    //设置金额 
      35. 
   
  
  
  
    35.    userAccount.addAmount(1); 
      36. 
   
  
  
  
    36.    logger.info(Thread.currentThread().getName() + " : user id : " + userId + " amount : " + userAccount.getAmount()); 
      37. 
   
  
  
  
    37.    //模拟存回数据库 
      38. 
   
  
  
  
    38.    redisTemplate.opsForValue().set(userId, userAccount); 
      39. 
   
  
  
  
    39.   } 
      40. 
   
  
  
  
    40.  } finally { 
      41. 
   
  
  
  
    41.   //释放锁 
      42. 
   
  
  
  
    42.   redisTemplate.delete(userId + ":syn"); 
      43. 
   
  
  
  
    43.   logger.info(Thread.currentThread().getName() + ":释放锁"); 
      44. 
   
  
  
  
    44.  } 
      45. 
   
  
  
  
    45. } 
      46. 
  
 
 
 
    在while代码块中，先判断对应用户ID是否在Redis中存在，如果不存在，则进行set加锁，如果存在，则跳出循环继续等待。
    上述代码，看起来实现了加锁的功能，但当执行程序时，会发现与未加锁一样，依旧存在并发问题。原因是：获取锁和加锁的操作并不是原子的。比如两个线程发现lock都是null，都进行了加锁，此时并发问题依旧存在。
    Redis原子性同步锁
    

    针对上述问题，可将获取锁和加锁的过程原子化处理。基于spring-boot-data-redis提供的原子化API可以实现：
      
 
 
 
      
   
  
  
  
    1. // 该方法使用了redis的指令：SETNX key value 
      2. 
   
  
  
  
    2. // 1.key不存在，设置成功返回value，setIfAbsent返回true； 
      3. 
   
  
  
  
    3. // 2.key存在，则设置失败返回null，setIfAbsent返回false； 
      4. 
   
  
  
  
    4. // 3.原子性操作； 
      5. 
   
  
  
  
    5. Boolean setIfAbsent(K var1, V var2); 
      6. 
  
 
 
 
    上述方法的原子化操作是对Redis的setnx命令的封装，在Redis中setnx的使用如下实例：
      
 
 
 
      
   
  
  
  
    1. redis> SETNX mykey "Hello" 
      2. 
   
  
  
  
    2. (integer) 1 
      3. 
   
  
  
  
    3. redis> SETNX mykey "World" 
      4. 
   
  
  
  
    4. (integer) 0 
      5. 
   
  
  
  
    5. redis> GET mykey 
      6. 
   
  
  
  
    6. "Hello" 
      7. 
  
 
 
 
    第一次，设置mykey时，并不存在，则返回1，表示设置成功；第二次设置mykey时，已经存在，则返回0，表示设置失败。再次查询mykey对应的值，会发现依旧是第一次设置的值。也就是说redis的setnx保证了唯一的key只能被一个服务设置成功。
    理解了上述API及底层原理，来看看线程中的实现方法代码如下：
      
 
 
 
      
   
  
  
  
    1. /** 
      2. 
   
  
  
  
    2.   * 1.原子操作加锁 
      3. 
   
  
  
  
    3.   * 2.竞争线程循环重试获得锁 
      4. 
   
  
  
  
    4.   * 3.业务完成释放锁 
      5. 
   
  
  
  
    5.   */ 
      6. 
   
  
  
  
    6.  private void atomicityRedisLock() { 
      7. 
   
  
  
  
    7.   //Spring data redis 支持的原子性操作 
      8. 
   
  
  
  
    8.   while (!redisTemplate.opsForValue().setIfAbsent(userId + ":syn", "lock")) { 
      9. 
   
  
  
  
    9.    try { 
      10. 
   
  
  
  
    10.     // 等待100毫秒重试获得锁 
      11. 
   
  
  
  
    11.     TimeUnit.MILLISECONDS.sleep(100); 
      12. 
   
  
  
  
    12.    } catch (InterruptedException e) { 
      13. 
   
  
  
  
    13.     e.printStackTrace(); 
      14. 
   
  
  
  
    14.    } 
      15. 
   
  
  
  
    15.   } 
      16. 
   
  
  
  
    16.   logger.info(Thread.currentThread().getName() + ":获得锁"); 
      17. 
   
  
  
  
    17.   try { 
      18. 
   
  
  
  
    18.    //模拟数据库中获取用户账号 
      19. 
   
  
  
  
    19.    UserAccount userAccount = (UserAccount) redisTemplate.opsForValue().get(userId); 
      20. 
   
  
  
  
    20.    if (userAccount != null) { 
      21. 
   
  
  
  
    21.     //设置金额 
      22. 
   
  
  
  
    22.     userAccount.addAmount(1); 
      23. 
   
  
  
  
    23.     logger.info(Thread.currentThread().getName() + " : user id : " + userId + " amount : " + userAccount.getAmount()); 
      24. 
   
  
  
  
    24.     //模拟存回数据库 
      25. 
   
  
  
  
    25.     redisTemplate.opsForValue().set(userId, userAccount); 
      26. 
   
  
  
  
    26.    } 
      27. 
   
  
  
  
    27.   } finally { 
      28. 
   
  
  
  
    28.    //释放锁 
      29. 
   
  
  
  
    29.    redisTemplate.delete(userId + ":syn"); 
      30. 
   
  
  
  
    30.    logger.info(Thread.currentThread().getName() + ":释放锁"); 
      31. 
   
  
  
  
    31.   } 
      32. 
   
  
  
  
    32.  } 
      33. 
  
 
 
 
    再次执行代码，会发现结果正确了，也就是说可以成功的对分布式线程进行了加锁。
    Redis分布式锁的死锁
    

    虽然上述代码执行结果没问题，但如果应用异常宕机，没来得及执行finally中释放锁的方法，那么其他线程则永远无法获得这个锁。
    此时可采用setIfAbsent的重载方法：
      
 
 
 
      
   
  
  
  
    1. Boolean setIfAbsent(K var1, V var2, long var3, TimeUnit var5); 
      2. 
  
 
 
 
    基于该方法，可以设置锁的过期时间。这样即便获得锁的线程宕机，在Redis中数据过期之后，其他线程可正常获得该锁。
    示例代码如下：
      
 
 
 
      
   
  
  
  
    1. private void atomicityAndExRedisLock() { 
      2. 
   
  
  
  
    2.   try { 
      3. 
   
  
  
  
    3.    //Spring data redis 支持的原子性操作,并设置5秒过期时间 
      4. 
   
  
  
  
    4.    while (!redisTemplate.opsForValue().setIfAbsent(userId + ":syn", 
      5. 
   
  
  
  
    5.      System.currentTimeMillis() + 5000, 5000, TimeUnit.MILLISECONDS)) { 
      6. 
   
  
  
  
    6.     // 等待100毫秒重试获得锁 
      7. 
   
  
  
  
    7.     logger.info(Thread.currentThread().getName() + ":尝试循环获取锁"); 
      8. 
   
  
  
  
    8.     TimeUnit.MILLISECONDS.sleep(1000); 
      9. 
   
  
  
  
    9.    } 
      10. 
   
  
  
  
    10.    logger.info(Thread.currentThread().getName() + ":获得锁--------"); 
      11. 
   
  
  
  
    11.    // 应用在这里宕机，进程退出，无法执行 finally; 
      12. 
   
  
  
  
    12.    Thread.currentThread().interrupt(); 
      13. 
   
  
  
  
    13.    // 业务逻辑... 
      14. 
   
  
  
  
    14.   } catch (InterruptedException e) { 
      15. 
   
  
  
  
    15.    e.printStackTrace(); 
      16. 
   
  
  
  
    16.   } finally { 
      17. 
   
  
  
  
    17.    //释放锁 
      18. 
   
  
  
  
    18.    if (!Thread.currentThread().isInterrupted()) { 
      19. 
   
  
  
  
    19.     redisTemplate.delete(userId + ":syn"); 
      20. 
   
  
  
  
    20.     logger.info(Thread.currentThread().getName() + ":释放锁"); 
      21. 
   
  
  
  
    21.    } 
      22. 
   
  
  
  
    22.   } 
      23. 
   
  
  
  
    23.  } 
      24. 
  
 
 
 
    

    业务超时及守护线程
    

    上面添加了Redis所的超时时间，看似解决了问题，但又引入了新的问题。
    比如，正常情况下线程A在5秒内可正常处理完业务，但偶发会出现超过5秒的情况。如果将超时时间设置为5秒，线程A获得了锁，但业务逻辑处理需要6秒。此时，线程A还在正常业务逻辑，线程B已经获得了锁。当线程A处理完时，有可能将线程B的锁给释放掉。
    在上述场景中有两个问题点：
      

    •  第一，线程A和线程B可能会同时在执行，存在并发问题。 
      • 

    •  第二，线程A可能会把线程B的锁给释放掉，导致一系列的恶性循环。 
      • 

    

    当然，可以通过在Redis中设置value值来判断锁是属于线程A还是线程B。但仔细分析会发现，这个问题的本质是因为线程A执行业务逻辑耗时超出了锁超时的时间。
    那么就有两个解决方案了：
      

    •  第一，将超时时间设置的足够长，确保业务代码能够在锁释放之前执行完成； 
      • 

    •  第二，为锁添加守护线程，为将要过期释放但未释放的锁增加时间； 
      • 

    

    第一种方式需要全行大多数情况下业务逻辑的耗时，进行超时时间的设定。
    第二种方式，可通过如下守护线程的方式来动态增加锁超时时间。
      
 
 
 
      
   
  
  
  
    1. public class DaemonThread implements Runnable { 
      2. 
   
  
  
  
    2.  private final static Logger logger = LoggerFactory.getLogger(DaemonThread.class); 
      3. 
   
  
  
  
    3.  
      4. 
   
  
  
  
    4.  // 是否需要守护 主线程关闭则结束守护线程 
      5. 
   
  
  
  
    5.  private volatile boolean daemon = true; 
      6. 
   
  
  
  
    6.  // 守护锁 
      7. 
   
  
  
  
    7.  private String lockKey; 
      8. 
   
  
  
  
    8.  
      9. 
   
  
  
  
    9.  private RedisTemplate redisTemplate; 
      10. 
   
  
  
  
    10.  
      11. 
   
  
  
  
    11.  public DaemonThread(String lockKey, RedisTemplate redisTemplate) { 
      12. 
   
  
  
  
    12.   this.lockKey = lockKey; 
      13. 
   
  
  
  
    13.   this.redisTemplate = redisTemplate; 
      14. 
   
  
  
  
    14.  } 
      15. 
   
  
  
  
    15.  
      16. 
   
  
  
  
    16.  @Override 
      17. 
   
  
  
  
    17.  public void run() { 
      18. 
   
  
  
  
    18.   try { 
      19. 
   
  
  
  
    19.    while (daemon) { 
      20. 
   
  
  
  
    20.     long time = redisTemplate.getExpire(lockKey, TimeUnit.MILLISECONDS); 
      21. 
   
  
  
  
    21.     // 剩余有效期小于1秒则续命 
      22. 
   
  
  
  
    22.     if (time < 1000) { 
      23. 
   
  
  
  
    23.      logger.info("守护进程: " + Thread.currentThread().getName() + " 延长锁时间 5000 毫秒"); 
      24. 
   
  
  
  
    24.      redisTemplate.expire(lockKey, 5000, TimeUnit.MILLISECONDS); 
      25. 
   
  
  
  
    25.     } 
      26. 
   
  
  
  
    26.     TimeUnit.MILLISECONDS.sleep(300); 
      27. 
   
  
  
  
    27.    } 
      28. 
   
  
  
  
    28.    logger.info(" 守护进程: " + Thread.currentThread().getName() + "关闭 "); 
      29. 
   
  
  
  
    29.   } catch (InterruptedException e) { 
      30. 
   
  
  
  
    30.    e.printStackTrace(); 
      31. 
   
  
  
  
    31.   } 
      32. 
   
  
  
  
    32.  } 
      33. 
   
  
  
  
    33.  
      34. 
   
  
  
  
    34.  // 主线程主动调用结束 
      35. 
   
  
  
  
    35.  public void stop() { 
      36. 
   
  
  
  
    36.   daemon = false; 
      37. 
   
  
  
  
    37.  } 
      38. 
   
  
  
  
    38. } 
      39. 
  
 
 
 
    上述线程每隔300毫秒获取一下Redis中锁的超时时间，如果小于1秒，则延长5秒。当主线程调用关闭时，守护线程也随之关闭。
    主线程中相关代码实现：
      
 
 
 
      
   
  
  
  
    1. private void deamonRedisLock() { 
      2. 
   
  
  
  
    2.   //守护线程 
      3. 
   
  
  
  
    3.   DaemonThread daemonThread = null; 
      4. 
   
  
  
  
    4.   //Spring data redis 支持的原子性操作,并设置5秒过期时间 
      5. 
   
  
  
  
    5.   String uuid = UUID.randomUUID().toString(); 
      6. 
   
  
  
  
    6.   String value = Thread.currentThread().getId() + ":" + uuid; 
      7. 
   
  
  
  
    7.   try { 
      8. 
   
  
  
  
    8.    while (!redisTemplate.opsForValue().setIfAbsent(userId + ":syn", value, 5000, TimeUnit.MILLISECONDS)) { 
      9. 
   
  
  
  
    9.     // 等待100毫秒重试获得锁 
      10. 
   
  
  
  
    10.     logger.info(Thread.currentThread().getName() + ":尝试循环获取锁"); 
      11. 
   
  
  
  
    11.     TimeUnit.MILLISECONDS.sleep(1000); 
      12. 
   
  
  
  
    12.    } 
      13. 
   
  
  
  
    13.    logger.info(Thread.currentThread().getName() + ":获得锁----"); 
      14. 
   
  
  
  
    14.    // 开启守护线程 
      15. 
   
  
  
  
    15.    daemonThread = new DaemonThread(userId + ":syn", redisTemplate); 
      16. 
   
  
  
  
    16.    Thread thread = new Thread(daemonThread); 
      17. 
   
  
  
  
    17.    thread.start(); 
      18. 
   
  
  
  
    18.    // 业务逻辑执行10秒... 
      19. 
   
  
  
  
    19.    TimeUnit.MILLISECONDS.sleep(10000); 
      20. 
   
  
  
  
    20.   } catch (InterruptedException e) { 
      21. 
   
  
  
  
    21.    e.printStackTrace(); 
      22. 
   
  
  
  
    22.   } finally { 
      23. 
   
  
  
  
    23.    //释放锁 这里也需要原子操作,今后通过 Redis + Lua 讲 
      24. 
   
  
  
  
    24.    String result = (String) redisTemplate.opsForValue().get(userId + ":syn"); 
      25. 
   
  
  
  
    25.    if (value.equals(result)) { 
      26. 
   
  
  
  
    26.     redisTemplate.delete(userId + ":syn"); 
      27. 
   
  
  
  
    27.     logger.info(Thread.currentThread().getName() + ":释放锁-----"); 
      28. 
   
  
  
  
    28.    } 
      29. 
   
  
  
  
    29.    //关闭守护线程 
      30. 
   
  
  
  
    30.    if (daemonThread != null) { 
      31. 
   
  
  
  
    31.     daemonThread.stop(); 
      32. 
   
  
  
  
    32.    } 
      33. 
   
  
  
  
    33.   } 
      34. 
   
  
  
  
    34.  } 
      35. 
  
 
 
 
    其中在获得锁之后，开启守护线程，在finally中将守护线程关闭。
    基于Lua脚本的实现
    

    在上述逻辑中，我们是基于spring-boot-data-redis提供的原子化操作来保证锁判断和执行的原子化的。在非Spring Boot项目中，则可以基于Lua脚本来实现。
    首先定义加锁和解锁的Lua脚本及对应的DefaultRedisScript对象，在RedisConfig配置类中添加如下实例化代码：
      
 
 
 
      
   
  
  
  
    1. @Configuration 
      2. 
   
  
  
  
    2. public class RedisConfig { 
      3. 
   
  
  
  
    3.  
      4. 
   
  
  
  
    4.  //lock script 
      5. 
   
  
  
  
    5.  private static final String LOCK_SCRIPT = " if redis.call('setnx',KEYS[1],ARGV[1]) == 1 " + 
      6. 
   
  
  
  
    6.    " then redis.call('expire',KEYS[1],ARGV[2]) " + 
      7. 
   
  
  
  
    7.    " return 1 " + 
      8. 
   
  
  
  
    8.    " else return 0 end "; 
      9. 
   
  
  
  
    9.  private static final String UNLOCK_SCRIPT = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call" + 
      10. 
   
  
  
  
    10.    "('del', KEYS[1]) else return 0 end"; 
      11. 
   
  
  
  
    11.  
      12. 
   
  
  
  
    12.  // ... 省略部分代码 
      13. 
   
  
  
  
    13.   
      14. 
   
  
  
  
    14.  @Bean 
      15. 
   
  
  
  
    15.  public DefaultRedisScript lockRedisScript() { 
      16. 
   
  
  
  
    16.   DefaultRedisScript defaultRedisScript = new DefaultRedisScript<>(); 
      17. 
   
  
  
  
    17.   defaultRedisScript.setResultType(Boolean.class); 
      18. 
   
  
  
  
    18.   defaultRedisScript.setScriptText(LOCK_SCRIPT); 
      19. 
   
  
  
  
    19.   return defaultRedisScript; 
      20. 
   
  
  
  
    20.  } 
      21. 
   
  
  
  
    21.  
      22. 
   
  
  
  
    22.  @Bean 
      23. 
   
  
  
  
    23.  public DefaultRedisScript unlockRedisScript() { 
      24. 
   
  
  
  
    24.   DefaultRedisScript defaultRedisScript = new DefaultRedisScript<>(); 
      25. 
   
  
  
  
    25.   defaultRedisScript.setResultType(Long.class); 
      26. 
   
  
  
  
    26.   defaultRedisScript.setScriptText(UNLOCK_SCRIPT); 
      27. 
   
  
  
  
    27.   return defaultRedisScript; 
      28. 
   
  
  
  
    28.  } 
      29. 
   
  
  
  
    29. } 
      30. 
  
 
 
 
    再通过在AccountOperationThread类中新建构造方法，将上述两个对象传入类中（省略此部分演示）。然后，就可以基于RedisTemplate来调用了，改造之后的代码实现如下：
      
 
 
 
      
   
  
  
  
    1. private void deamonRedisLockWithLua() { 
      2. 
   
  
  
  
    2.  //守护线程 
      3. 
   
  
  
  
    3.  DaemonThread daemonThread = null; 
      4. 
   
  
  
  
    4.  //Spring data redis 支持的原子性操作,并设置5秒过期时间 
      5. 
   
  
  
  
    5.  String uuid = UUID.randomUUID().toString(); 
      6. 
   
  
  
  
    6.  String value = Thread.currentThread().getId() + ":" + uuid; 
      7. 
   
  
  
  
    7.  try { 
      8. 
   
  
  
  
    8.   while (!redisTemplate.execute(lockRedisScript, Collections.singletonList(userId + ":syn"), value, 5)) { 
      9. 
   
  
  
  
    9.    // 等待1000毫秒重试获得锁 
      10. 
   
  
  
  
    10.    logger.info(Thread.currentThread().getName() + ":尝试循环获取锁"); 
      11. 
   
  
  
  
    11.    TimeUnit.MILLISECONDS.sleep(1000); 
      12. 
   
  
  
  
    12.   } 
      13. 
   
  
  
  
    13.   logger.info(Thread.currentThread().getName() + ":获得锁----"); 
      14. 
   
  
  
  
    14.   // 开启守护线程 
      15. 
   
  
  
  
    15.   daemonThread = new DaemonThread(userId + ":syn", redisTemplate); 
      16. 
   
  
  
  
    16.   Thread thread = new Thread(daemonThread); 
      17. 
   
  
  
  
    17.   thread.start(); 
      18. 
   
  
  
  
    18.   // 业务逻辑执行10秒... 
      19. 
   
  
  
  
    19.   TimeUnit.MILLISECONDS.sleep(10000); 
      20. 
   
  
  
  
    20.  } catch (InterruptedException e) { 
      21. 
   
  
  
  
    21.   logger.error("异常", e); 
      22. 
   
  
  
  
    22.  } finally { 
      23. 
   
  
  
  
    23.   //使用Lua脚本：先判断是否是自己设置的锁，再执行删除 
      24. 
   
  
  
  
    24.   // key存在,当前值=期望值时,删除key;key存在,当前值!=期望值时,返回0; 
      25. 
   
  
  
  
    25.   Long result = redisTemplate.execute(unlockRedisScript, Collections.singletonList(userId + ":syn"), value); 
      26. 
   
  
  
  
    26.   logger.info("redis解锁:{}", RELEASE_SUCCESS.equals(result)); 
      27. 
   
  
  
  
    27.   if (RELEASE_SUCCESS.equals(result)) { 
      28. 
   
  
  
  
    28.    if (daemonThread != null) { 
      29. 
   
  
  
  
    29.     //关闭守护线程 
      30. 
   
  
  
  
    30.     daemonThread.stop(); 
      31. 
   
  
  
  
    31.     logger.info(Thread.currentThread().getName() + ":释放锁---"); 
      32. 
   
  
  
  
    32.    } 
      33. 
   
  
  
  
    33.   } 
      34. 
   
  
  
  
    34.  } 
      35. 
   
  
  
  
    35. } 
      36. 
  
 
 
 
    其中while循环中加锁和finally中的释放锁都是基于Lua脚本来实现了。
    Redis锁的其他因素
    

    除了上述实例，在使用Redis分布式锁时，还可以考虑以下情况及方案。
    Redis锁的不可重入
    

    当线程在持有锁的情况下再次请求加锁，如果一个锁支持一个线程多次加锁，那么这个锁就是可重入的。如果一个不可重入锁被再次加锁，由于该锁已经被持有，再次加锁会失败。Redis可通过对锁进行重入计数，加锁时加 1，解锁时减 1，当计数归 0时释放锁。
    可重入锁虽然高效但会增加代码的复杂性，这里就不举例说明了。
    等待锁释放
    

    有的业务场景，发现被锁则直接返回。但有的场景下，客户端需要等待锁释放然后去抢锁。上述示例就属于后者。针对等待锁释放也有两种方案：
      

    •  客户端轮训：当未获得锁时，等待一段时间再重新获取，直到成功。上述示例就是基于这种方式实现的。这种方式的缺点也很明显，比较耗费服务器资源，当并发量大时会影响服务器的效率。 
      • 

    •  使用Redis的订阅发布功能：当获取锁失败时，订阅锁释放消息，获取锁成功后释放时，发送释放消息。 
      • 

    

    集群中的主备切换和脑裂
    

    在Redis包含主从同步的集群部署方式中，如果主节点挂掉，从节点提升为主节点。如果客户端A在主节点加锁成功，指令还未同步到从节点，此时主节点挂掉，从节点升为主节点，新的主节点中没有锁的数据。这种情况下，客户端B就可能加锁成功，从而出现并发的场景。
    当集群发生脑裂时，Redis master节点跟slave 节点和 sentinel 集群处于不同的网络分区。sentinel集群无法感知到master的存在，会将 slave 节点提升为 master 节点，此时就会存在两个不同的 master 节点。从而也会导致并发问题的出现。Redis Cluster集群部署方式同理。
    小结
    

    通过生产环境中的一个问题，排查原因，寻找解决方案，到最终对基于Redis分布式的深入研究，这便是学习的过程。
    同时，每当面试或被问题如何解决分布式共享资源时，我们会脱口而出”基于Redis实现分布式锁“，但通过本文的学习会发现，Redis分布式锁并不是万能的，而且在使用的过程中还需要注意超时、死锁、误解锁、集群选主/脑裂等问题。
    Redis以高性能著称，但在实现分布式锁的过程中还是存在一些问题。因此，基于Redis的分布式锁可以极大的缓解并发问题，但要完全防止并发，还是得从数据库层面入手。
                
    
                网页标题：加锁了还有并发问题？Redis分布式锁，真的用对了？
                
    
                标题URL：http://www.shufengxianlan.com/qtweb/news47/98197.html
            
    
            
    
                网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等
                                
    
            
    
                广告
                
            
    
            
    
                声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源：
                创新互联
            
    
        
        
    
            
    猜你还喜欢下面的内容
    
            
        
    
    
    
    
        
    
            
    python知识
    
            
        
    
        
    
            
    同城分类信息
    
            
        
    
    
    


    
    
    
        成都网站建设　　 　雅安电信机房　　 　盐亭做网站　　 　成都名片印刷　　 　成都网站设计　　 　软文推广　　 　平城网站运维　　 　手机网站建设　　 　成都圣月服饰　　 　广安搬家公司　　 　名片印刷厂　　 　虚拟主机购买　　 　研学实践　　 　蜜朵婚庆　　 　LED显示屏　　 　软文发布平台　　 　正泰动物制药　　 　遂宁做网站　　 　营销型网站建设　　 　成都西信服务器托管