redis 系列之——分布式锁

用户头像
诸葛小猿
关注
发布于: 2020 年 07 月 09 日
redis系列之——分布式锁

Redis系列目录



redis系列之——分布式锁



redis系列之——缓存穿透、缓存击穿、缓存雪崩



redis系列之——Redis为什么这么快?



redis系列之——数据持久化(RDB和AOF)



redis系列之——一致性hash算法



redis系列之——高可用(主从、哨兵、集群)



redis系列之——事物及乐观锁



redis系列之——数据类型geospatial:你隔壁有没有老王?



redis系列之——数据类型bitmaps:今天你签到了吗?



布隆过滤器是个啥!



端午节最后一天了,三天假期过得太快了,更坏的消息是,下周还要上六天班呢!





趁着假期的尾巴,和大家再叨逼叨redis是如何实现分布式锁的。这期也是为了填之前《redis系列之——缓存穿透、缓存击穿、缓存雪崩》留下的坑。



这一期不是专门聊分布式锁的,所以不会涉及到各种分布式锁实现及相关的比较,只是聊一下如何使用redis实现分布式锁。感觉上一个坑还没填完,这里又挖了一个大坑,各种分布式锁实现及相关的比较后面有时间我会跪着填上的,请大家给我多一点点时间,多一点点温柔。



本期是硬核输出,也是面试高频,更是实战必会。



什么是分布式锁



先说一个场景,消费者在购物网站上下单或收银员在POS机上下单,由于网络等问题,在连续点击了两下,后端网站如何处理,如何响应?对于这个问题,前端需要处理,后端也需要处理。这里主要说后端,后端不光要处理重复订单问题,还有处理幂等问题。幂等问题简单来说就是相同的请求,要有相同的响应结果,这里就不展开了。重复订单该如何处理?



对于一个小的访问量不大的网站,部署了一个tomcat,这个问题可以简单的通过JVM提供的同步锁synchronized实现。但是当网站访问量越来越大时,需要扩展机器,synchronized就不能起作用了。相同的下单参数连续两次请求后端服务器,可能会被分发到两个tomcat上,就会出现synchronized失效问题。





分布式锁要解决的就是多机器部署时,相同请求并发访问时资源竞争问题。请求到达每个tomcat时,首先要去redis中注册锁,注册成功返回true则说明获得了锁,可以继续处理相关的业务,处理完成后释放锁。同一时刻只能有一个tomcat能获得锁,其他没获得锁的tomcat则多次尝试继续获得锁,没有获得锁不能处理业务。获得锁的tomcat释放锁后,其他的tomcat才能有一个获得锁。





这里是使用redis做外部存储介质存储锁的,使用zookeeper也是类似的。万变不离其宗,原理都一样,只是技术选型有差别。



redis实现



废话就不说了,直接上代码。



1.pom.xml



<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.3.1.RELEASE</version>
</parent>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>



这里需要注意,我使用的是spring-boot。redis的相关jar使用的是spring-boot-starter-data-redis。



注意版本是2以上,版本1和版本2在redis锁的实现上有个关键的差异,后面会说到。



2.application.yml



server:
port: 8080
spring:
application:
name: java-summary
redis:
database: 10
password: 123456
timeout: 20000
host: 127.0.0.1
port: 6379
pool:
max-idle: 20
min-idle: 20
max-wait: 10000
max-active: 5000



3.RedisConfiguration.java



package com.wuxiaolong.redis;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
/**
* Description:
*
* @author 诸葛小猿
* @date 2020-06-27
*
* 去除redis序列化时,key-value的乱码问题
*/
@Configuration
public class RedisConfiguration {
@Bean
@Primary
public RedisTemplate setRedisTemplate(RedisTemplate redisTemplate) {
RedisSerializer stringSerializer = new StringRedisSerializer();
redisTemplate.setKeySerializer(stringSerializer);
redisTemplate.setValueSerializer(stringSerializer);
redisTemplate.setHashKeySerializer(stringSerializer);
redisTemplate.setHashValueSerializer(stringSerializer);
return redisTemplate;
}
}



这个配置类就是为了解决redis的key和value在序列化时的乱码问题,将序列化的方式设置为string格式



4.RedisLock.java



package com.wuxiaolong.redis;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
/**
* Description:
*
* @author 诸葛小猿
* @date 2020-06-27
*/
@Component
@Slf4j
public class RedisLock {
@Autowired
private RedisTemplate redisTemplate;
/**
* 添加元素
*
* @param key
* @param value
*/
public void set(String key, String value) {
if (key == null || value == null) {
return;
}
redisTemplate.opsForValue().set(key, value);
}
/**
* 如果已经存在返回false,否则返回true
*
* @param key
* @param value
* @return
*/
public Boolean setNx(String key, String value, Long expireTime, TimeUnit mimeUnit) {
if (key == null || value == null) {
return false;
}
//spiring boot 1.5 版本setIfAbsent只能设置key-value,需要单独对key设置过期时间,因为是两步操作,所以不是原子性
//Boolean tf = redisTemplate.opsForValue().setIfAbsent(key, value);
//redisTemplate.expire(key, expireTime, mimeUnit);
// 在spiring boot 2 可以直接使用 redisTemplate的setIfAbsent设置key-value和过期时间,是原子性
Boolean tf =redisTemplate.opsForValue().setIfAbsent(key,value,expireTime, mimeUnit);
return tf;
}
/**
* 获取数据
*
* @param key
* @return
*/
public Object get(String key) {
if (key == null) {
return null;
}
return redisTemplate.opsForValue().get(key);
}
/**
* 删除
*
* @param key
* @return
*/
public void remove(Object key) {
if (key == null) {
return ;
}
redisTemplate.delete(key);
}
/**
* 加锁
*
* @param key
* @param waitTime 等待时间,在这个时间内会多次尝试获取锁,超过这个时间还没获得锁,就返回false
* @param interval 间隔时间,每隔多长时间尝试一次获的锁
* @param expireTime key的过期时间
*/
public Boolean lock(String key, Long waitTime,Long interval, Long expireTime) {
String value = UUID.randomUUID().toString().replaceAll("-", "").toLowerCase();
Boolean flag = setNx(key, value, expireTime, TimeUnit.MILLISECONDS);
// 尝试获取锁 成功返回
if (flag) {
return flag;
} else {
// 获取失败
// 现在时间
long newTime = System.currentTimeMillis();
// 等待过期时间
long loseTime = newTime + waitTime;
// 不断尝试获取锁成功返回
while (System.currentTimeMillis() < loseTime) {
Boolean testFlag = setNx(key, value, expireTime, TimeUnit.MILLISECONDS);
if (testFlag) {
return testFlag;
}
try {
Thread.sleep(interval);
} catch (InterruptedException e) {
log.error("获取锁异常",e);
}
}
}
return false;
}
/**
* 释放锁
*
* @param key
* @return
*/
public void unLock(String key) {
remove(key);
}
public Boolean setIfAbsent(String key, String value){
Boolean tf = redisTemplate.opsForValue().setIfAbsent(key, value);
redisTemplate.expire(key, 60, TimeUnit.DAYS);
return tf;
}
}



这个就是redis锁的核心实现。上面提到的spring-boot版本问题就是这里的setNx方法中的redisTemplate的setIfAbsent的差异。



setIfAbsent的作用是,在存储key和value时,如果key不存在则保存后返回true,说明获得锁;如果key存在则保存后返回false,说明这个锁正在使用中,不能获取。



在获取锁时,有两步操作,首先是要存储这个key-value,然后需要对其设置一个过期时间,防止出现死锁。在spring-boot的版本为1时,只提供了setIfAbsent(key, value)的API,设置成功后需要调用redisTemplate.expire(key, expireTime, mimeUnit)对这个key设置过期时间,这是两步是非原子性的操作,如果第一步执行成功,第二步执行失败,就可能出现死锁,虽然概率很低。在spring-boot的版本为2时,提供了一个原子性的API setIfAbsent(key,value,expireTime, mimeUnit),在保存key-value的同时设置了过期时间,推荐使用。



4.RedisLockTest.java



package com.wuxiaolong.redis;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import java.util.Map;
/**
* Description:
*
* @author 诸葛小猿
* @date 2020-06-27
*/
@RestController
@Slf4j
public class RedisLockTest {
@Autowired
private RedisLock redisLock;
public static final String ORDER_LOCK_PREFIX = "order:lock:";
@RequestMapping(value = "save/order",method = RequestMethod.POST)
public String saveOrder(@RequestBody Map<String,String> order){
// 获取商户号、门店号、收银机号、收银单号,组成全局唯一的订单号
String orderSn = order.get("merchantSn")+order.get("storeSn")+order.get("workstationSn")+order.get("checkSn");
// 使用全局唯一的订单号做分布式锁的key
String lockKey = ORDER_LOCK_PREFIX + orderSn;
try{
// 加锁 每隔100毫秒获取一次锁;1分钟内拿不到锁就返回false;锁的过期时间5分钟(防止死锁)
Boolean tf = redisLock.lock(lockKey,1L * 60 * 1000,100L,5L * 60 * 1000);
if(tf){
// todo 处理拿到锁时的业务
return "success";
}else {
// todo 处理没有拿到锁时的业务
return "fail";
}
}catch (Exception e){
log.error("业务异常",e);
}finally {
// 一定要在finally中释放锁
redisLock.unLock(lockKey);
}
return "success";
}
}



这是测试类的实例。首先需要获得全局唯一的订单号,然后使用这个订单号为key。在try块的最前面先去redis中获得锁,根据是否获得锁做不同的操作。需要根据自己的业务定义waitTime,interval, expireTime这三个参数。一定要在最后的finally中释放锁



完成,收工!!





传播知识,共享价值】,感谢小伙伴们的关注和支持,我是【诸葛小猿】,一个彷徨中奋斗的互联网民工。





发布于: 2020 年 07 月 09 日 阅读数: 68
用户头像

诸葛小猿

关注

我是诸葛小猿,一个彷徨中奋斗的互联网民工 2020.07.08 加入

公众号:foolish_man_xl

评论

发布
暂无评论
redis系列之——分布式锁