一、前言
既然是分布式锁,这就说明服务器不是一台,可能是很多台。我们使用一个案例,来一步一步说明。假设某网站有一个秒杀商品,一看还有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来介绍一下分布式如何实现,以及其中的原理。