内容摘要：目前的项目单体结构的基本上已经没有了，大多是分布式集群或者是微服务这些。既然是多台服务器。就免不了资源的共享问题。既然是资源共享就免不了并发的问题。针对这些问题，redis也给出了一个很好的解决方案，

目前的供各项目单体结构的基本上已经没有了，大多是种锁分布式集群或者是微服务这些。既然是还需多台服务器。就免不了资源的分布共享问题。既然是式锁资源共享就免不了并发的问题。针对这些问题，供各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来介绍一下分布式如何实现，以及其中的原理。

本文转载自微信公众号「愚公要移山」，可以通过以下二维码关注。转载本文请联系愚公要移山公众号。