基于注解的方式实现分布式锁
关于分布式锁的实现由两种 1. 基于redis 2. 基于zookeeper
为了方便分布式锁的使用, 基于注解的方式抽取成公用组件
DisLock注解
/**
* 分布式锁的注解, 通过指定key作为分布式锁的key
*
* @author wang.js on 2019/1/29.
* @version 1.0
*/
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface DisLock {
/**
* 分布式锁的key
*
* @return
*/
String key();
/**
* 分布式锁用的业务场景id
*
* @return
*/
String biz();
/**
* 过期时间, 默认是5秒
* 单位是秒
*
* @return
*/
int expireTime() default 5;
}
处理DisLock的切面
/**
* 处理@DisLock注解的切面
*
* @author wang.js on 2019/1/29.
* @version 1.0
*/
@Aspect
@Order(value = 1)
@Component
public class DisLockAspect {
@Resource
private DisLockUtil disLockUtil;
private static final int MIN_EXPIRE_TIME = 3;
@Around(value = "@annotation(disLock)")
public Object execute(ProceedingJoinPoint proceedingJoinPoint, DisLock disLock) throws Throwable {
int expireTIme = disLock.expireTime() < MIN_EXPIRE_TIME ? MIN_EXPIRE_TIME : disLock.expireTime();
String disKey = CacheKeyParser.parse(proceedingJoinPoint, disLock.key(), disLock.biz());
boolean lock = disLockUtil.lock(disKey, expireTIme);
int count = 1;
while (!lock && count < MIN_EXPIRE_TIME) {
lock = disLockUtil.lock(disKey, expireTIme);
count++;
TimeUnit.SECONDS.sleep(1);
}
Object proceed;
if (lock) {
// 允许查询
try {
proceed = proceedingJoinPoint.proceed();
} finally {
// 删除分布式锁
disLockUtil.unlock(disKey, false);
}
} else {
throw new CustomException(ErrorCodeEnum.DUPLICATE_REQUEST.getMessage());
}
return proceed;
}
}
redis的配置
/**
* @author wang.js
* @date 2018/12/17
* @copyright yougou.com
*/
@Configuration
public class RedisConfig {
@Value("${spring.redis.host}")
private String host;
@Value("${spring.redis.port:6379}")
private Integer port;
@Bean
public JedisPool jedisPool() {
//1.设置连接池的配置对象
JedisPoolConfig config = new JedisPoolConfig();
//设置池中最大连接数
config.setMaxTotal(50);
//设置空闲时池中保有的最大连接数
config.setMaxIdle(10);
config.setMaxWaitMillis(3000L);
config.setTestOnBorrow(true);
//2.设置连接池对象
return new JedisPool(config,host,port);
}
}
redis分布式锁的实现
/**
* redis分布式锁的实现
*
* @author wang.js
* @date 2018/12/18
* @copyright yougou.com
*/
@Component
public class DisLockUtil {
@Resource
private JedisPool jedisPool;
private static final int DEFAULT_EXPIRE_TIME = 5;
private static final Long RELEASE_SUCCESS = 1L;
private static final String LOCK_SUCCESS = "OK";
private static final String SET_IF_NOT_EXIST = "NX";
private static final String SET_WITH_EXPIRE_TIME = "PX";
/**
* 尝试获取分布式锁
*
* @param jedis Redis客户端
* @param lockKey 锁
* @param requestId 请求标识
* @param expireTime 超期时间
* @return 是否获取成功
*/
public static boolean tryGetDistributedLock(Jedis jedis, String lockKey, String requestId, int expireTime) {
String result = jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
if (LOCK_SUCCESS.equals(result)) {
return true;
}
return false;
}
/**
* 释放分布式锁
*
* @param jedis Redis客户端
* @param lockKey 锁
* @param requestId 请求标识
* @return 是否释放成功
*/
public static boolean releaseDistributedLock(Jedis jedis, String lockKey, String requestId) {
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));
if (RELEASE_SUCCESS.equals(result)) {
return true;
}
return false;
}
/**
* 释放锁
*
* @param key
* @return
*/
public final boolean unlock(String key, boolean needCheck) {
boolean result = false;
Jedis jedis = jedisPool.getResource();
try {
if (needCheck) {
String expireTimeCache = jedis.get(key);
// 判断锁是否过期了
if (StringUtils.isBlank(expireTimeCache)) {
result = true;
}
if (System.currentTimeMillis() - Long.parseLong(expireTimeCache) > 0) {
// 直接删除
jedis.del(key);
result = true;
}
} else {
jedis.del(key);
}
} finally {
jedis.close();
}
return result;
}
/**
* 获取分布式锁
*
* @param key
* @param expireSecond
* @return
*/
public final boolean lock(String key, int expireSecond) {
if (StringUtils.isBlank(key)) {
throw new RuntimeException("传入的key为空");
}
expireSecond = expireSecond == 0 ? DEFAULT_EXPIRE_TIME : expireSecond;
// 过期的时候的时间戳
long expireTime = System.currentTimeMillis() + expireSecond * 1000 + 1;
boolean setResult = false;
Jedis jedis = jedisPool.getResource();
try {
if (jedis.setnx(key, String.valueOf(expireTime)) == 1) {
// 说明加锁成功
setResult = true;
}
if (jedis.ttl(key) < 0) {
jedis.expire(key, expireSecond);
}
if (setResult) {
return true;
}
String expireTimeCache = jedis.get(key);
System.out.println(expireTimeCache + "====" + jedis.ttl(key) + ", now:" + System.currentTimeMillis());
// 判断锁是否过期了
if (StringUtils.isNotBlank(expireTimeCache) && System.currentTimeMillis() - Long.parseLong(expireTimeCache) > 0) {
String oldExpireTime = jedis.getSet(key, String.valueOf(expireTime));
if (StringUtils.isNotBlank(oldExpireTime) && oldExpireTime.equals(String.valueOf(expireTime))) {
jedis.expire(key, expireSecond);
setResult = true;
}
}
} finally {
jedis.close();
}
return setResult;
}
}
实现分布式锁的关键是对key的设置, 需要获取实际的参数来设置分布式锁, 这里自定义了解析器
/**
* cache key 的解析器
*
* @author wang.js on 2019/2/27.
* @version 1.0
*/
public class CacheKeyParser {
/**
* 解析缓存的key
*
* @param proceedingJoinPoint 切面
* @param cacheKey 缓存的key
* @param biz 业务
* @return String
* @throws IllegalAccessException 异常
*/
public static String parse(ProceedingJoinPoint proceedingJoinPoint, String cacheKey, String biz) throws IllegalAccessException {
// 解析实际参数的key
String key = cacheKey.replace("#", "");
StringTokenizer stringTokenizer = new StringTokenizer(key, ".");
Map<String, Object> nameAndValue = getNameAndValue(proceedingJoinPoint);
Object actualKey = null;
while (stringTokenizer.hasMoreTokens()) {
if (actualKey == null) {
actualKey = nameAndValue.get(stringTokenizer.nextToken());
} else {
actualKey = getPropValue(actualKey, stringTokenizer.nextToken());
}
}
return biz + actualKey;
}
/**
* 获取参数Map集合
*
* @param joinPoint 切面
* @return Map<String, Object>
*/
private static Map<String, Object> getNameAndValue(ProceedingJoinPoint joinPoint) {
Object[] paramValues = joinPoint.getArgs();
String[] paramNames = ((CodeSignature) joinPoint.getSignature()).getParameterNames();
Map<String, Object> param = new HashMap<>(paramNames.length);
for (int i = 0; i < paramNames.length; i++) {
param.put(paramNames[i], paramValues[i]);
}
return param;
}
/**
* 获取指定参数名的参数值
*
* @param obj
* @param propName
* @return
* @throws IllegalAccessException
*/
public static Object getPropValue(Object obj, String propName) throws IllegalAccessException {
Field[] fields = obj.getClass().getDeclaredFields();
for (Field f : fields) {
if (f.getName().equals(propName)) {
//在反射时能访问私有变量
f.setAccessible(true);
return f.get(obj);
}
}
return null;
}
}
ErrorCodeEnum
public enum ErrorCodeEnum {
SUCCESS("查询成功", "200"),
SERVER_ERROR("服务器异常", "500"),
SECKILL_END("秒杀活动已结束", "250"),
GOODS_KILLED("秒杀成功", "502"),
ERROR_SIGN("签名不合法", "260"),
UPDATE_SUCCESS("更新成功", "0"),
SAVE_SUCCESS("保存成功", "0"),
UPDATE_FAIL("更新失败", "256"),
EMPTY_PARAM("参数为空", "257"),
SAVE_ERROR("保存失败", "262"),
SERVER_TIMEOUT("调用超时", "501"),
USER_NOT_FOUND("找不到用户", "502"),
COUPON_NOT_FOUND("找不到优惠券", "503"),
DUPLICATE("出现重复", "504"),
USER_STATUS_ABNORMAL("用户状态异常", "505"),
NO_TOKEN("无token，请重新登录", "506"),
ERROR_TOKEN("token不合法", "507"),
EMPTY_RESULT("暂无数据", "508"),
DUPLICATE_REQUEST("重复请求", "509"),
;
/**
* 定义的message
*/
private String message;
/**
* 定义的错误码
*/
private String errCode;
ErrorCodeEnum(String message, String errCode) {
this.message = message;
this.errCode = errCode;
}
public String getMessage() {
return message;
}
protected void setMessage(String message) {
this.message = message;
}
public String getErrCode() {
return errCode;
}
protected void setErrCode(String errCode) {
this.errCode = errCode;
}
}
自定义异常CustomException
/**
* @author Eric on 2018/12/24.
* @version 1.0
*/
@Data
@NoArgsConstructor
@AllArgsConstructor
@Accessors(chain = true)
@EqualsAndHashCode(callSuper = true)
public class CustomException extends RuntimeException {
private String message;
}
配置文件
spring:
redis:
host: mini7
port: 6379
测试
定义一个方法, 加上@RedisCache注解, cacheKey的值必须是#实际参数名.属性名的格式, 如果想要成其他的格式可以修改CacheKeyParser中的parse方法
@DisLock(key = "#id", biz = CommonBizConstant.SECOND_KILL)
@Override public String testRedisCache(String id) { LOGGER.info("调用方法获取值"); return "大傻逼"; } 在springboot启动类上加上@ComponentScan({"com.eric"})
/**
* @author Eric on 2019/1/26.
* @version 1.0
*/
@SpringBootApplication
@MapperScan("com.eric.base.data.dao")
@ComponentScan({"com.eric"})
@EnableFeignClients
@EnableDiscoveryClient
public class BaseDataApplication {
public static void main(String[] args) {
SpringApplication.run(BaseDataApplication.class, args);
}
}
写个测试类调用上面的方法
/**
* 基础数据
*
* @author wang.js on 2019/2/27.
* @version 1.0
*/
@SpringBootTest
@RunWith(SpringRunner.class)
public class BaseDataTest {
@Resource
private SysDictService sysDictService;
@Test
public void t1() {
for (int i = 0; i < 100; i++) {
sysDictService.testRedisCache("1");
}
}
}