目前的单体项目结构基本没有了，大部分是分布式集群或者微服务。因为是多台服务器。资源共享的问题是不可避免的。既然是资源共享，就免不了并发的问题。对于这些问题，redis也给出了很好的解决方案，那就是分布式锁。本文主要讨论为什么需要分布式锁这个话题。前段时间群里有小哥问，既然分布式锁可以解决大部分的生产问题，那么java提供的那些锁有什么用呢？直接使用分布式锁是不行的。这个问题我想了很多。一开始，我在网上搜索，看看有没有类似的答案。想了想之后。为了解决这个问题，我们需要从本质上进行分析。好，我们上车出发吧。1.前言既然是分布式锁，那就意味着不是一台服务器，而是多台服务器。我们用一个案例来一步步解释。假设一个网站有一个闪购产品，而且还有100多个，那么陕西、江苏、西藏等地的人看到了这个活动，于是开始疯狂的闪购。假设这个秒杀产品的数量值存储在一个redis数据库中。但是不同地区的用户秒杀使用的服务器不同。这样就形成了集群接入方式。我们使用Springboot集成redis的方式。2、项目搭建准备（1）添加pom依赖org.springframework.bootspring-boot-starter-weborg.springframework.bootspring-boot-starter-testtestorg.springframework.bootspring-boot-starter-data-redisorg.apache.commonscommons-pool2(2）添加属性配置#Redis数据库索引（默认为0）spring.redis.database=0#Redis服务器地址spring.redis.host=localhost#Redis服务器连接端口spring.redis.port=6379#Redis服务器连接密码（默认为空）spring.redis.password=#连接池最大连接数（使用负值表示不限制）default8spring.redis.lettuce.pool.max-active=8#最大阻塞等待的时间连接池（使用负值表示无限制）Default-1spring.redis.lettuce.pool.max-wait=-1#连接池中最大空闲连接数默认为8spring.redis.lettuce.pool.max-idle=8#连接池中最小空闲连接默认为0(newStringRedisSerializer());redisTemplate.setValueSerializer(newGenericJackson2JsonRedisSerializer());redisTemplate.setConnectionFactory(connectionFactory);returnredisTemplate;}}(4)新建controller，创建Mycontroller类@RestControllerpublicclassMyController{@AutowiredprivateStringRedisTemplatestringRedisTemplate;@GetMapping("/test")publicStringdeduceGoods(){intgoods=Integer.parseInt(stringRedisTemplate.opsForValue().get("goods"));intrealGoods=goods-1;if(goods>0){stringRedisTemplate.opsForValue().set("goods",realGoods+"");return"你秒杀成功了，还有剩余："+realGoods+"pieces";}else{return"商品已售罄，欢迎参加下次活动";}}}一个很简单的集成教程端口是8080，我们复制这个项目，改端口为8090，使用Nginx用于负载均衡构建集群。现在我们已经整理好了环境。下面开始分析。3、为什么需要分布式锁Phase1：在原生的方式下，我们使用多线程访问8080端口，因为没有锁，这个时候肯定会出现并发问题。所以，我们可能会想，既然这货是共享资源，是多线程访问的，那么马上就能想到java中的各种锁，最著名的就是synchronized。所以我们不妨优化一下上面的代码。阶段二：使用同步锁此时我们修改代码：@RestControllerpublicclassMyController{@AutowiredprivateStringRedisTemplatestringRedisTemplate;@GetMapping("/test")publicStringdeduceGoods(){synchronized(this){intgoods=Integer.parseInt(stringRedisTemplate.opsForValue().get("goods"));intrealGoods=goods-1;if(goods>0){stringRedisTemplate.opsForValue().set("goods",realGoods+"");return"你秒杀成功了，还有还剩:"+realGoods+"piece";}else{return"商品已售罄，欢迎下次活动";}}}}看，现在我们用synchronized关键字加锁，这样当多个线程并发访问不会有数据不一致等问题。这种做法在单体结构下确实有用。目前结构单一的项目很少，一般都是集群式的。这时候synchronized就不行了。为什么同步不起作用？我们使用集群的方式访问限时抢购（nginx为我们做负载均衡）。您将看到数据不一致。也就是说synchronized关键字的作用范围其实是一个进程，这个进程下的所有线程都可以加锁。但是多进程就不行了。对于秒杀产品，这个值是固定的。但是每个区域可能有一个服务器。这样，不同地区的服务器不一样，地址不一样，流程也不一样。所以synchronized不能保证数据的一致性。第三阶段：分布式锁上的synchronized关键字不能保证多进程锁机制。为了解决这个问题，我们可以使用redis分布式锁。现在我们再修改一下代码：@RestControllerpublicclassMyController{@AutowiredprivateStringRedisTemplatestringRedisTemplate;@GetMapping("/test")publicStringdeduceGoods(){Booleanresult=stringRedisTemplate.opsForValue().setIfAbsent("lock","冯冬冬");if(!result){return"其他人正在秒杀，无法进入";}intgoods=Integer.parseInt(stringRedisTemplate.opsForValue().get("goods"));intrealGoods=goods-1;if(goods>0){stringRedisTemplate.opsForValue().set("goods",realGoods+"");System.out.println("你秒杀成功了，还剩："+realGoods+"pieces");}else{System.out.println("商品已售罄，欢迎参加下次活动");}stringRedisTemplate.delete("lock");return"success";}}就这么简单，我们只是加了一句，然后进行判断.其实setIfAbsent方法的作用就是redis中的setnx。这意味着如果当前键已经存在，则什么都不做并返回false。如果当前key不存在，那么我们就可以操作了。最后别忘了放开这把钥匙，这样其他人就可以实时进来杀人了。当然，这里只是一个基本的案例。其实分布式锁的实现还有很多步骤，很多坑就不给了。随便解决几个：第四阶段：分布式锁优化（一）第一个坑：秒杀产品异常，最终无法释放锁,"冯冬冬");if(!result){return"其他人正在instakilling,无法进入";}try{intgoods=Integer.parseInt(stringRedisTemplate.opsForValue().get("goods"));intrealGoods=goods-1;if(goods>0){stringRedisTemplate.opsForValue().set("goods",realGoods+"");System.out.println("你秒杀成功了，还有剩余："+realGoods+"pieces");}else{System.out.println("商品已售罄，欢迎参加下次活动");}}finally{stringRedisTemplate.delete("lock");}return"success";}此时我们添加一个try和finally语句就可以了。最终必须删除锁。(2)第二个坑：杀货时间过长，其他用户等不及).setIfAbsent("lock","冯冬冬");if(!result){return"他人instakilling,无法进入";}try{intgoods=Integer.parseInt(stringRedisTemplate.opsForValue().get("goods"));intrealGoods=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毫秒，说明他还可能秒杀不成功，其他用户就会进来，当这种情况多了，可能会有大量的用户闪不成功killing等其他用户会进来，有可能是其他用户提前删除了锁，而当前用户还没有成功kill锁。最终会导致数据不一致。看怎么解决：publicStringdeduceGoods()throwsException{Stringuser=UUID.randomUUID().toString();stringRedisTemplate.expire("lock",10,TimeUnit.MILLISECONDS);Booleanresult=stringRedisTemplate.opsForValue().setIfAbsent("lock",user);if(!result){return"别人秒杀不能进入";}try{intgoods=Integer.parseInt(stringRedisTemplate.opsForValue().get("goods"));intrealGoods=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";}也就是我们删除锁的时候判断是否是当前线程，如果是，则删除，如果不是，则不删除tdeleteit，这样其他线程进来就不会乱删锁，造成混乱。OK，至此基本介绍了分布式锁的原因。对于分布式锁redisson已经做得很好了，下一篇文章也会介绍分布式锁是如何实现的，以及围绕Redisson的原理。本文转载自微信公众号“愚公要移山”，可关注下方二维码。转载本文请联系愚公移山公众号。
