加锁了还有并发问题？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的分布式锁实现。

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

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

从数据库读取用户金额; 程序修改金额; 再将最新金额存储到数据库; 下面从最初不加锁，不同步处理，逐步推演出最终的分布式锁。 基础集成及类构建

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

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

  org.springframework.boot 
 spring-boot-starter-data-redis  
  org.springframework.boot 
 spring-boot-starter-web  

账户对应实体类UserAccount：

public class UserAccount {  
 //用户ID  private String userId; 
 //账户内金额  private int amount; 
  //添加账户金额 
 public void addAmount(int amount) {   this.amount = this.amount + amount; 
 }  // 省略构造方法和getter/setter  
} 

创建一个线程实现类AccountOperationThread：

public class AccountOperationThread implements Runnable {  
 private final static Logger logger = LoggerFactory.getLogger(AccountOperationThread.class);  
 private static final Long RELEASE_SUCCESS = 1L;  
 private String userId;  
 private RedisTemplate redisTemplate;  
 public AccountOperationThread(String userId, RedisTemplate redisTemplate) {   this.userId = userId; 
  this.redisTemplate = redisTemplate;  } 
  @Override 
 public void run() {   noLock(); 
 }  
   private void noLock() { 
  try {    Random random = new Random(); 
   // 模拟线程进行业务处理    TimeUnit.MILLISECONDS.sleep(random.nextInt(100) + 1); 
  } catch (InterruptedException e) {    e.printStackTrace(); 
  }   //模拟数据库中获取用户账号 
  UserAccount userAccount = (UserAccount) redisTemplate.opsForValue().get(userId);   // 金额+1 
  userAccount.addAmount(1);   logger.info(Thread.currentThread().getName() + " : user id : " + userId + " amount : " + userAccount.getAmount()); 
  //模拟存回数据库   redisTemplate.opsForValue().set(userId, userAccount); 
 } } 

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

@Configuration public class RedisConfig { 
  @Bean 
 public RedisTemplate redisTemplate(RedisConnectionFactory redisConnectionFactory) {   RedisTemplate redisTemplate = new RedisTemplate<>(); 
  redisTemplate.setConnectionFactory(redisConnectionFactory);   Jackson2JsonRedisSerializer