java中使用Redis实现分布式锁

前言

目前很多大型的互联网公司后端都采用了分布式架构来支撑前端应用，其中服务拆分就是分布式的一种体现，既然服务拆分了，那么多个服务协调工作就会出现一些资源竞争的情况。比如多个服务对同一个表中的数据进行处理。容易出现类似多线程的不同步问题。多线程不同步我们一般通过代码中的锁来强制同步执行。但是对于这种多个服务的，这种本地锁久显得无能为力了。

锁实现

锁的实现目的就是保证某个时间只允许一个持有锁的线程来操作一个对象。实现思路就是设置一个任何线程都可见可操作的对象（锁），通过控制对象的状态来让线程知道当前锁是否被人持有了。举个例子比如火车上的卫生间，就是通过列车上的灯的状态来标识厕所是否被占用，如果是红灯，表示厕所当前有人，其他人不可以开门。如果是绿灯，则表示当前厕所是空的，可以开门进去操作。那么对比着来看，厕所其实就是被锁定的资源，那个灯就是锁标识。当然也可以认为门上的锁是锁标识（其实本来就是个锁）。

基于redis的分布式锁

原理就是利用redis中string的key value，如果set key成功表示目标资源没人使用，如果没有set key失败或报错表示资源已经有人使用了。

加锁

如下：

很显然单纯的使用set key value这种方式是不可以行的，因为redis默认会进行覆盖。
mysql中建表的时候有一句 crate table if not exists test (xxxxxx); 同样的redis中也提供了类似的先判断在set的用法。

set key value nx

其中nx 表示的就是 not exists的意思

那么使用这种方式貌似就可以达到标识资源被占用的目的，如果一个线程设置了某个key，其他线程再set这个key就会失败。

但是如果A线程加锁之后去执行业务，但是业务失败了，导致A线程没有解锁，那么其他线程就只能看一直等待，就比如有人去了厕所，完事儿之后在里面玩手机玩上头了一直不出来，这个时候厕所外的标识灯一直是红色，那么别人就只能一直等待。所以要加一个机制来避免这种问题，比如设置一个超时时间。据说国外有一个发明就是为了防止有人长时间占用厕所，将厕所墙壁设计成透明的，墙壁是一个特殊的材料制成，可以让人从里面看不到外面，外面也看不到里面，但是人在里面如果超过十分钟后，这种材料就会褪色，最后变成全透明的，新闻在这里https://www.bilibili.com/read/cv6156620/ 。同样的道理，为了防止持有锁的线程在执行业务的时候报错导致不能及时的释放锁，所以redis本身可以针对特定key设置过期时间，如果到期了自动释放锁。
命令如下：

expire lock 5

设置锁的过期时间为5秒

上图模拟了线程设置锁标识之后5秒之后其他线程可以获取锁。

但是，这样还有一个问题，在高并发环境下，两个命令分别执行不能保证原子性，所以继续优化
如下：

set lock 1 ex 5 nx

这条命令其实是将上面的两条命令糅合在一起，这样就可以保证获取锁的原子性了。

下面考虑另外一个问题，如果A持有锁的线程在执行一个比较耗时的业务，达到了锁的自动释放时间，但是工作任没有完成，这样锁就会被提前释放，然后B线程获取到了锁开始工作，在B线程工作还没有结束时，A线程工作结束了，于是调用了释放锁的操作，这样就会引起连续的错误反应，最终导致锁没有起到任何作用。于是继续优化:
解决方案一
程序员根据业务情况评估被锁保护的公共资源的执行耗时，适当的增减redis中key的自动释放时间。
解决方法二
加锁的方式不变，锁的自动释放时间设置可以略长一些，程序加锁的时候给redis中的value设置一个随机数，释放锁的时候根据随机数来判断当前锁是不是自己的。可以采用lua的方式来保证判断的和释放的原子性问题。如下：

if redis.call("get",KEYS[1]) == ARGV[1] 
 then 
   return redis.call("del",KEYS[1])  
 else   
  return 0  
end

解锁

解锁其实就是将key清除

del key

java代码实现：

锁的主体逻辑

package team.lcf.lock.redis.impl;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.params.SetParams;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.UUID;

/**
 * @program: LuoThinking-cloud
 * @description: 基于redis实现的分布式锁
 * 原理： 锁的基本思想都是确保当前只有一个线程可以操作某个程序，为了
 * 加锁：set lock:test true nx 5 ex
 * 释放锁：del lock:test 为了保证原子性，一般采用lua脚本实现
 * @author: Cheng Zhi
 * @create: 2022-06-13 21:42
 **/
public class RedisLock {

    private static final Logger logger = LoggerFactory.getLogger(RedisLock.class);

    private Jedis redis;

    /**
     * 将key 的值设为value ，当且仅当key 不存在，等效于 SETNX。
     */
    public static final String NX = "NX";

    /**
     * seconds — 以秒为单位设置 key 的过期时间，等效于EXPIRE key seconds
     */
    public static final String EX = "EX";

    /**
     * 调用set后的返回值
     */
    public static final String OK = "OK";

    /**
     * 默认请求锁的超时时间(ms 毫秒)
     */
    private static final long TIME_OUT = 100;

    /**
     * 默认锁的有效时间(s)
     */
    public static final int EXPIRE = 30;

    /**
     * 解锁的lua脚本
     */
    public static final String UNLOCK_LUA;

    static {
        StringBuilder sb = new StringBuilder();
        sb.append("if redis.call("get",KEYS[1]) == ARGV[1] ");
        sb.append("then ");
        sb.append("    return redis.call("del",KEYS[1]) ");
        sb.append("else ");
        sb.append("    return 0 ");
        sb.append("end ");
        UNLOCK_LUA = sb.toString();
    }

    /**
     * 锁标志对应的key
     */
    private String lockKey;

    /**
     * 锁对应的值
     */
    private String lockValue;

    /**
     * 锁的有效时间(s)
     */
    private int expireTime = EXPIRE;

    /**
     * 请求锁的超时时间(ms)
     */
    private long timeout = TIME_OUT;

    /**
     * 锁标记
     */
    private boolean locked = false;

    final Random random = new Random();

    public int getExpireTime() {
        return expireTime;
    }

    public void setExpireTime(int expireTime) {
        this.expireTime = expireTime;
    }

    public long getTimeout() {
        return timeout;
    }

    public void setTimeout(long timeout) {
        this.timeout = timeout;
    }

    /**
     * 构造方法
     * @param jedis
     * @param lockKey
     */
    public RedisLock(Jedis jedis, String lockKey) {
        this.redis = jedis;
        this.lockKey = lockKey + "_lock";
    }

    /**
     * 自定义过期时间和超时时间
     * @param jedis
     * @param lockKey
     * @param expireTime
     * @param timeout
     */
    public RedisLock(Jedis jedis, String lockKey, int expireTime, long timeout) {
        this(jedis, lockKey);
        this.expireTime = expireTime;
        this.timeout = timeout;
    }
    /**
     * 默认加锁
     * @return
     */
    public boolean lock() {
        // 获取随机数作为key
        lockValue = UUID.randomUUID().toString();
        final String set = redis.set(lockKey, lockValue, new SetParams().nx().ex(expireTime));
        locked = OK.equalsIgnoreCase(set);
        return locked;
    }

    /**
     * 释放锁
     * @return
     */
    public boolean unLock() {

        if (locked) {
            List<String> keys = new ArrayList<>();
            keys.add(lockKey);
            List<String> values = new ArrayList<>();
            values.add(lockValue);
            Long result = (Long) redis.eval(UNLOCK_LUA, keys, values);

            locked = result == 0? true: false;
            if (locked) {
                logger.warn("分布式锁{}解锁失败", lockKey);
            }
        }

        return locked;
    }

    /**
     * 以阻塞的方式获取锁,一直尝试获取，直到取到为止
     * @return
     */
    public boolean tryLockBlock() {

        while(true) {
            if (lock()) {
                return locked;
            }
            // 每次请求等待一段时间
            sleep(10, 50000);
        }
    }

    /**
     * 在指定超时时间内获取
     * @return
     */
    public boolean tryLock() {
        // 请求锁超时时间，纳秒
        long _timeout = timeout * 1000000;
        // 系统当前时间，纳秒
        long nowTime = System.nanoTime();
        while ((System.nanoTime() - nowTime) < _timeout) {
            if (lock()) {
                locked = true;
                // 上锁成功结束请求
                return true;
            }
            // 每次请求等待一段时间
            sleep(10, 50000);
        }
        return locked;

    }

    /**
     * 线程等待时间
     *
     * @param millis 毫秒
     * @param nanos  纳秒
     */
    private void sleep(long millis, int nanos) {
        try {
            Thread.sleep(millis, random.nextInt(nanos));
        } catch (InterruptedException e) {
            logger.info("获取分布式锁休眠被中断：", e);
        }
    }
}

设计注解：

package team.lcf.lock;

import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

/**
 * 方法粒度锁
 */
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface Lock {

    /**
     * 使用锁的类型
     * 默认使用的是redis实现
   * @return
     */
    public String LockType() default "redis";

    /**
     * 锁超时时间 ms
     * @return
     */
    public int timeOut() default 1000;

    /**
     * key过期时间(s) 默认为10s
     * @return
     */
    public int expiteTime() default 10;

}

注解的处理：

package team.lcf.lock.redis.impl;

import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.Signature;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.reflect.MethodSignature;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import team.lcf.lock.Lock;

import java.lang.reflect.Method;

/**
 * @program: LuoThinking-cloud
 * @description: Redis分布式锁管理
 * @author: Cheng Zhi
 * @create: 2022-06-14 14:38
 **/
@Aspect // 声明为aop
@Component
public class RedisLockManager {

    private static Logger logger = LoggerFactory.getLogger(RedisLockManager.class);

    @Autowired
    private JedisPool jedisPool;

    @Around(value = "@annotation(team.lcf.lock.Lock)")
    public Object around(ProceedingJoinPoint pjp) throws Throwable {

        Object o = null;
        // 获取注解自定义信息
        MethodSignature signature = (MethodSignature) pjp.getSignature();
        Method method = signature.getMethod();
        Lock annotation = method.getAnnotation(Lock.class);
        String lockType = annotation.LockType();
        Integer timeOut = annotation.timeOut();
        Integer expireTime = annotation.expiteTime();
        // todo 后续引入其他锁实现这里可以采用策略模式
        if (lockType.equals("redis")) {

            Jedis jedis = jedisPool.getResource();
            // 获取所有的lockKey,lockKey采用类名 + 方法名称
            Class<?> aClass = pjp.getTarget().getClass();
            String methodName = pjp.getSignature().getName();
            // 分布式锁key名称
            String lockKey = "lock:" + aClass.getName() + ":" +  methodName;
            // 获取自定义参数，比如类型
            RedisLock redisLock = new RedisLock(jedis, lockKey, expireTime, timeOut);

            try {
                if (redisLock.tryLockBlock()) {
                    try {
                        o = pjp.proceed();
                    } finally {
                        // 释放锁
                        if (redisLock.unLock()) {
                            logger.warn("分布式锁释放失败，key = " + lockKey);
                        }
                    }
                } else {
                    logger.warn("分布式锁获取失败，当前不可用，请稍后再试。 key= " + lockKey );
                }
            } catch (Throwable throwable) {
                throwable.printStackTrace();
            } finally {
                jedis.close();
            }
        } else {
            // 不是redis的暂时不支持
            o = pjp.proceed();
        }

        return o;

    }

}

使用范例：

package team.lcf.service;

import org.springframework.stereotype.Service;
import team.lcf.lock.Lock;
import team.lcf.thread.ThreadUtils;

/**
 * @program: LuoThinking-cloud
 * @description: CzTestService
 * @author: Cheng Zhi
 * @create: 2022-06-14 18:02
 **/
@Service
public class CzTestService {

    @Lock(timeOut = 1000)
    public void print() {
        System.out.println("----开始执行----");
        System.out.println("休息中......");
        ThreadUtils.doSleep(1000L);
        System.out.println("----执行结束----");
    }
}


@GetMapping("/sayHello")
public String sayHello() {

    Object userInfo = ContextHolder.getRequestContext().get("userInfo");
    for (int i=0; i<5; i++) {
        new Thread(new Runnable() {
            @Override
            public void run() {
                service.print();
            }
        }).start();
    }
    return "hello " + userInfo.toString();
}

开启掘金成长之旅！这是我参与「掘金日新计划 · 2 月更文挑战」的第22天，点击查看活动详情