导航：首页 > 数据库 >

详解redis如何实现分布式锁

发表于：2024-11-25 作者：千家信息网编辑

千家信息网最后更新 2024年11月25日，小编这次要给大家分享的是详解redis如何实现分布式锁，文章内容丰富，感兴趣的小伙伴可以来了解一下，希望大家阅读完这篇文章之后能够有所收获。前言系统的不断扩大，分布式锁是最基本的保障。与单机的多线程不

千家信息网最后更新 2024年11月25日详解redis如何实现分布式锁

小编这次要给大家分享的是详解redis如何实现分布式锁，文章内容丰富，感兴趣的小伙伴可以来了解一下，希望大家阅读完这篇文章之后能够有所收获。

前言

系统的不断扩大，分布式锁是最基本的保障。与单机的多线程不一样的是，分布式跨多个机器。线程的共享变量无法跨机器。

为了保证一个在高并发存场景下只能被同一个线程操作，java并发处理提供ReentrantLock或Synchronized进行互斥控制。但是这仅仅对单机环境有效。我们实现分布式锁大概通过三种方式。

redis实现分布式锁
数据库实现分布式锁
zk实现分布式锁

今天我们介绍通过redis实现分布式锁。实际上这三种和java对比看属于一类。都是属于程序外部锁。

原理剖析

上述三种分布式锁都是通过各自为依据对各个请求进行上锁，解锁从而控制放行还是拒绝。redis锁是基于其提供的setnx命令。
setnx当且仅当key不存在。若给定key已经存在，则setnx不做任何动作。setnx是一个原子性操作。
和数据库分布式相比，因为redis内存轻量。所以redis分布式锁性能更好

实现

原理很简单。结合springboot项目我们实现一套通过注解形式对接口进行库存上锁案例进行理解

编写注解

我们编写注解。方便我们在接口上添加注解提供拦截信息

/** * @author 张新华 * @version V1.0 * @Package com.ay.framework.order.redis.product * @date 2020年03月26日, 0026 10:29 * @Copyright © 2020 安元科技有限公司 */@Target(ElementType.METHOD)@Retention(RetentionPolicy.RUNTIME)@Documented@Inheritedpublic @interface StockLock {  /**   * @author zxhtom   * @Description 锁key的前缀   * @Date 15:25 2020年03月25日, 0025   * @Param []   * @return java.lang.String   */  String prefix() default "";  /**   * @author zxhtom   * @Description key的分隔符   * @Date 15:27 2020年03月25日, 0025   * @Param []   * @return java.lang.String   */  String delimiter() default ":";}

/** * @author 张新华 * @version V1.0 * @Package com.ay.framework.order.redis.product * @date 2020年03月26日, 0026 11:09 * @Copyright © 2020 安元科技有限公司 */@Target({ElementType.PARAMETER , ElementType.METHOD , ElementType.FIELD})@Retention(RetentionPolicy.RUNTIME)@Documented@Inheritedpublic @interface StockParam {  /**  * @author zxhtom  * @Description 组成key  * @Date 11:11 2020年03月26日, 0026  * @Param []  * @return java.lang.String[]  */  String[] names() default {""};}

拦截器拦截

redis分布式锁实现的关键就是拦截器的编写。上面的注解只是为了实现拦截的一个辅助。

@Around("execution(public * *(..)) && @annotation(com.ay.framework.order.redis.product.StockLock)")

通过springboot的Around进行针对StockLock注解的拦截。通过拦截我们可以获取到拦截的方法、参数、及需要的锁的参数。

我们获取到需要锁的名称这里叫做【a】之后通过redis的原子性操作对该key进行递减操作。

为了方便我们在削减库存的时候可以对库存进行更新操作。我们在递减库存前还需要借助于另一把锁。这一把锁我们叫做【a_key】

换句话说我们接口想访问就必须获取【a】锁，拿到【a】锁需要减少库存。减少库存之前需要获取【a_key】锁。

拿到锁之后处理完逻辑之后我们需要释放对应锁。

RedisAtomicLong entityIdCounter = new RedisAtomicLong(lockKey, redisTemplate.getConnectionFactory());  if (redisTemplate.hasKey(CoreConstants.UPDATEPRODUCTREDISLOCKKEY + lockKey)) {    //表示lockKey的库存信息有变动。此时无法进行交易    throw new BusinessException("库存变动。暂无法交易");  }  Long increment = entityIdCounter.decrementAndGet();  if (increment >= 0) {    try {      Object proceed = pjp.proceed();    } catch (Throwable throwable) {      //所占资源需要释放回资源池      while (!redisLock.tryGetLock(CoreConstants.UPDATEPRODUCTREDISLOCKKEY + lockKey, "")) {      }      //表示lockKey的库存信息有变动。此时无法进行交易      long l = entityIdCounter.incrementAndGet();      if (l < 1) {        redisTemplate.opsForValue().set(lockKey,1);      }      redisLock.unLock(CoreConstants.UPDATEPRODUCTREDISLOCKKEY + lockKey);      throwable.printStackTrace();    }  } else {    redisTemplate.opsForValue().set(lockKey,0);    throw new BusinessException("库存不足！无法操作");  }

因为我们上锁就需要释放锁。但是程序在中途处理业务是发生异常导致没有走到释放锁的步骤。这个时候就导致我们的分布式锁一直被锁。俗称【死锁】。为了避免这种场景的发生。我们常常在上锁的时候给一个有效期。有效期已过自动释放锁。这个特性恰好和redis的过期策略不摩尔和。

上述提及工具

RedisLock

public Boolean tryGetLock(String key , String value) {  return tryGetLock(key, value, -1, TimeUnit.DAYS);}public Boolean tryGetLock(String key , String value, Integer expire) {  return tryGetLock(key, value, expire, TimeUnit.SECONDS);}public Boolean tryGetLock(String key , String value, Integer expire , TimeUnit timeUnit) {  ValueOperations operations = redisTemplate.opsForValue();  if (operations.setIfAbsent(key, value)) {    //说明 redis没有该key , 换言之 加锁成功 设置过期时间防止死锁    if (expire > 0) {      redisTemplate.expire(key, expire, timeUnit);    }    return true;  }  return false;}public Boolean unLock(String key) {  return redisTemplate.delete(key);}

StockKeyGenerator

@Component()@Primarypublic class StockKeyGenerator implements CacheKeyGenerator {  @Override  public String getLockKey(ProceedingJoinPoint pjp) {    //获取方法签名    MethodSignature signature = (MethodSignature) pjp.getSignature();    Method method = signature.getMethod();    //获取方法cacheLock注解    StockLock stockLock = method.getAnnotation(StockLock.class);    //获取方法参数    Object[] args = pjp.getArgs();    Parameter[] parameters = method.getParameters();    StringBuilder builder = new StringBuilder();    for (int i = 0; i < parameters.length; i++) {      StockParam stockParam = parameters[i].getAnnotation(StockParam.class);      Object arg = args[i];      if (arg instanceof Map) {        Map temArgMap = (Map) arg;        String[] names = stockParam.names();        for (String name : names) {          if (builder.length() > 0) {            builder.append(stockLock.delimiter());          }          builder.append(temArgMap.get(name));        }      }    }    return builder.toString();  }}

问题分析

上面分析了一个死锁的场景，理论上出了死锁我们redis分布锁很好的解决了分布式问题。但是还是会出现问题。下面列举写小编遇到的问题。

业务处理时间>上锁过期时间

a线程获取到锁，开始进行业务处理需要8S，

在8S内，锁的有效期是5S，在锁过期后也就是第6S ， b线程进入开始获取锁这个时候b是可以获取到新锁的。这个时候就是有问题的。

假设b线程业务处理只需要3S ，但是因为a线程释放了锁，所以在第8S的时候虽然b线程没有释放锁，b的锁也没有过期但是这时候也没有了锁。从而导致C线程也可以进入

看完这篇关于详解redis如何实现分布式锁的文章，如果觉得文章内容写得不错的话，可以把它分享出去给更多人看到。

很赞哦！