基于redis的分布式锁实现

1.分布式锁介绍

　　在计算机系统中，锁做为一种控制并发的机制无处不在。

　　单机环境下，操做系统可以在进程或线程之间经过本地的锁来控制并发程序的行为。而在现在的大型复杂系统中，一般采用的是分布式架构提供服务。

　　分布式环境下，基于本地单机的锁没法控制分布式系统中分开部署客户端的并发行为，此时分布式锁就应运而生了。

一个可靠的分布式锁应该具有如下特性：

　　1.互斥性：做为锁，须要保证任什么时候刻只能有一个客户端(用户)持有锁

　　2.可重入： 同一个客户端在得到锁后，能够再次进行加锁

　　3.高可用：获取锁和释放锁的效率较高，不会出现单点故障

　　4.自动重试机制：当客户端加锁失败时，可以提供一种机制让客户端自动重试

2.分布式锁api接口

/**
 * 分布式锁 api接口
 */
public interface DistributeLock {

    /**
     * 尝试加锁
     * @param lockKey 锁的key
     * @return 加锁成功 返回uuid
     *         加锁失败 返回null
     * */
    String lock(String lockKey);

    /**
     * 尝试加锁 (requestID相等 可重入)
     * @param lockKey 锁的key
     * @param expireTime 过时时间 单位：秒
     * @return 加锁成功 返回uuid
     *         加锁失败 返回null
     * */
    String lock(String lockKey, int expireTime);

    /**
     * 尝试加锁 (requestID相等 可重入)
     * @param lockKey 锁的key
     * @param requestID 用户ID
     * @return 加锁成功 返回uuid
     *         加锁失败 返回null
     * */
    String lock(String lockKey, String requestID);

    /**
     * 尝试加锁 (requestID相等 可重入)
     * @param lockKey 锁的key
     * @param requestID 用户ID
     * @param expireTime 过时时间 单位：秒
     * @return 加锁成功 返回uuid
     *         加锁失败 返回null
     * */
    String lock(String lockKey, String requestID, int expireTime);

    /**
     * 尝试加锁，失败自动重试 会阻塞当前线程
     * @param lockKey 锁的key
     * @return 加锁成功 返回uuid
     *         加锁失败 返回null
     * */
    String lockAndRetry(String lockKey);

    /**
     * 尝试加锁，失败自动重试 会阻塞当前线程 (requestID相等 可重入)
     * @param lockKey 锁的key
     * @param requestID 用户ID
     * @return 加锁成功 返回uuid
     *         加锁失败 返回null
     * */
    String lockAndRetry(String lockKey, String requestID);

    /**
     * 尝试加锁 (requestID相等 可重入)
     * @param lockKey 锁的key
     * @param expireTime 过时时间 单位：秒
     * @return 加锁成功 返回uuid
     *         加锁失败 返回null
     * */
    String lockAndRetry(String lockKey, int expireTime);

    /**
     * 尝试加锁 (requestID相等 可重入)
     * @param lockKey 锁的key
     * @param expireTime 过时时间 单位：秒
     * @param retryCount 重试次数
     * @return 加锁成功 返回uuid
     *         加锁失败 返回null
     * */
    String lockAndRetry(String lockKey, int expireTime, int retryCount);

    /**
     * 尝试加锁 (requestID相等 可重入)
     * @param lockKey 锁的key
     * @param requestID 用户ID
     * @param expireTime 过时时间 单位：秒
     * @return 加锁成功 返回uuid
     *         加锁失败 返回null
     * */
    String lockAndRetry(String lockKey, String requestID, int expireTime);

    /**
     * 尝试加锁 (requestID相等 可重入)
     * @param lockKey 锁的key
     * @param expireTime 过时时间 单位：秒
     * @param requestID 用户ID
     * @param retryCount 重试次数
     * @return 加锁成功 返回uuid
     *         加锁失败 返回null
     * */
    String lockAndRetry(String lockKey, String requestID, int expireTime, int retryCount);

    /**
     * 释放锁
     * @param lockKey 锁的key
     * @param requestID 用户ID
     * @return true     释放本身所持有的锁 成功
     *         false    释放本身所持有的锁 失败
     * */
    boolean unLock(String lockKey, String requestID);
}

3.基于redis的分布式锁的简单实现

3.1 基础代码 

　　当前实现版本的分布式锁基于redis实现，使用的是jedis链接池来和redis进行交互，并将其封装为redisClient工具类(仅封装了demo所需的少数接口)

redisClient工具类：

public class RedisClient {

    private static final Logger LOGGER = LoggerFactory.getLogger(RedisClient.class);

    private JedisPool pool;

    private static RedisClient instance = new RedisClient();

    private RedisClient() {
        init();
    }

    public static RedisClient getInstance(){
        return instance;
    }

    public Object eval(String script, List<String> keys, List<String> args) {
        Jedis jedis = getJedis();
        Object result = jedis.eval(script, keys, args);
        jedis.close();
        return result;
    }

    public String get(final String key){
        Jedis jedis = getJedis();
        String result = jedis.get(key);
        jedis.close();
        return result;
    }

    public String set(final String key, final String value, final String nxxx, final String expx, final int time) {
        Jedis jedis = getJedis();
        String result = jedis.set(key, value, nxxx, expx, time);
        jedis.close();
        return result;
    }

    private void init(){
        Properties redisConfig = PropsUtil.loadProps("redis.properties");
        int maxTotal = PropsUtil.getInt(redisConfig,"maxTotal",10);
        String ip = PropsUtil.getString(redisConfig,"ip","127.0.0.1");
        int port = PropsUtil.getInt(redisConfig,"port",6379);

        JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
        jedisPoolConfig.setMaxTotal(maxTotal);
        pool = new JedisPool(jedisPoolConfig, ip,port);
        LOGGER.info("链接池初始化成功 ip={}, port={}, maxTotal={}",ip,port,maxTotal);
    }

    private Jedis getJedis(){
        return pool.getResource();
    }
}

View Code

所依赖的工具类：

package util;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;

/**
 * @Author xiongyx
 * @Create 2018/4/11.
 */
public final class PropsUtil {

    private static final Logger LOGGER = LoggerFactory.getLogger(PropsUtil.class);

    /**
     * 读取配置文件
     * */
    public static Properties loadProps(String fileName){
        Properties props = null;
        InputStream is = null;
        try{
            //:::绝对路径得到输入流
            is = Thread.currentThread().getContextClassLoader().getResourceAsStream(fileName);
            if(is == null){
                //:::没找到文件,抛出异常
                throw new FileNotFoundException(fileName + " is not found");
            }
            props = new Properties();
            props.load(is);
        }catch(IOException e){
            LOGGER.error("load propertis file fail",e);
        }finally {
            if(is != null){
                try{
                    //:::关闭io流
                    is.close();
                } catch (IOException e) {
                    LOGGER.error("close input Stream fail",e);
                }
            }
        }

        return props;
    }

    /**
     * 获取字符串属性(默认为空字符串)
     * */
    public static String getString(Properties properties,String key){
        //:::调用重载函数 默认值为:空字符串
        return getString(properties,key,"");
    }

    /**
     * 获取字符串属性
     * */
    public static String getString(Properties properties,String key,String defaultValue){
        //:::key对应的value数据是否存在
        if(properties.containsKey(key)){
            return properties.getProperty(key);
        }else{
            return defaultValue;
        }
    }

    /**
     * 获取int属性 默认值为0
     * */
    public static int getInt(Properties properties,String key){
        //:::调用重载函数，默认为:0
        return getInt(properties,key,0);
    }

    /**
     * 获取int属性
     * */
    public static int getInt(Properties properties,String key,int defaultValue){
        //:::key对应的value数据是否存在
        if(properties.containsKey(key)){
            return CastUtil.castToInt(properties.getProperty(key));
        }else{
            return defaultValue;
        }
    }

    /**
     * 获取boolean属性，默认值为false
     */
    public static boolean getBoolean(Properties properties,String key){
        return getBoolean(properties,key,false);
    }

    /**
     * 获取boolean属性
     */
    public static boolean getBoolean(Properties properties,String key,boolean defaultValue){
        //:::key对应的value数据是否存在
        if(properties.containsKey(key)){
            return CastUtil.castToBoolean(properties.getProperty(key));
        }else{
            return defaultValue;
        }
    }
}


public final class CastUtil {

    /**
     * 转为 string
     * */
    public static String castToString(Object obj){
        return castToString(obj,"");
    }

    /**
     * 转为 string 提供默认值
     * */
    public static String castToString(Object obj,String defaultValue){
        if(obj == null){
            return defaultValue;
        }else{
            return obj.toString();
        }
    }

    /**
     * 转为 int
     * */
    public static int castToInt(Object obj){
        return castToInt(obj,0);
    }

    /**
     * 转为 int 提供默认值
     * */
    public static int castToInt(Object obj,int defaultValue){
        if(obj == null){
            return defaultValue;
        }else{
            return Integer.parseInt(obj.toString());
        }
    }

    /**
     * 转为 double
     * */
    public static double castToDouble(Object obj){
        return castToDouble(obj,0);
    }

    /**
     * 转为 double 提供默认值
     * */
    public static double castToDouble(Object obj,double defaultValue){
        if(obj == null){
            return defaultValue;
        }else{
            return Double.parseDouble(obj.toString());
        }
    }

    /**
     * 转为 long
     * */
    public static long castToLong(Object obj){
        return castToLong(obj,0);
    }

    /**
     * 转为 long 提供默认值
     * */
    public static long castToLong(Object obj,long defaultValue){
        if(obj == null){
            return defaultValue;
        }else{
            return Long.parseLong(obj.toString());
        }
    }

    /**
     * 转为 boolean
     * */
    public static boolean castToBoolean(Object obj){
        return  castToBoolean(obj,false);
    }

    /**
     * 转为 boolean 提供默认值
     * */
    public static boolean castToBoolean(Object obj,boolean defaultValue){
        if(obj == null){
            return defaultValue;
        }else{
            return Boolean.parseBoolean(obj.toString());
        }
    }
}

View Code

初始化lua脚本 LuaScript.java：

　　在分布式锁初始化时，使用init方法读取lua脚本

public class LuaScript {
    /**
     * 加锁脚本 lock.lua
     * */
    public static String LOCK_SCRIPT = "";

    /**
     * 解锁脚本 unlock.lua
     * */
    public static String UN_LOCK_SCRIPT = "";

    public static void init(){
        try {
            initLockScript();
            initUnLockScript();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private static void initLockScript() throws IOException {
        String filePath = Objects.requireNonNull(LuaScript.class.getClassLoader().getResource("lock.lua")).getPath();
        LOCK_SCRIPT = readFile(filePath);
    }

    private static void initUnLockScript() throws IOException {
        String filePath = Objects.requireNonNull(LuaScript.class.getClassLoader().getResource("unlock.lua")).getPath();
        UN_LOCK_SCRIPT = readFile(filePath);
    }

    private static String readFile(String filePath) throws IOException {
        try (
            FileReader reader = new FileReader(filePath);
            BufferedReader br = new BufferedReader(reader)
        ) {
            String line;
            StringBuilder stringBuilder = new StringBuilder();
            while ((line = br.readLine()) != null) {
                stringBuilder.append(line).append(System.lineSeparator());
            }

            return stringBuilder.toString();
        }
    }
}

View Code

单例的RedisDistributeLock基础属性

public final class RedisDistributeLock implements DistributeLock {

    /**
     * 无限重试
     * */
    public static final int UN_LIMIT_RETRY = -1;

    private RedisDistributeLock() {
        LuaScript.init();
    }

    private static DistributeLock instance = new RedisDistributeLock();

    /**
     * 持有锁 成功标识
     * */
    private static final Long ADD_LOCK_SUCCESS = 1L;
    /**
     * 释放锁 失败标识
     * */
    private static final Integer RELEASE_LOCK_SUCCESS = 1;

    /**
     * 默认过时时间 单位：秒
     * */
    private static final int DEFAULT_EXPIRE_TIME_SECOND = 300;
    /**
     * 默认加锁重试时间 单位：毫秒
     * */
    private static final int DEFAULT_RETRY_FIXED_TIME = 3000;
    /**
     * 默认的加锁浮动时间区间 单位：毫秒
     * */
    private static final int DEFAULT_RETRY_TIME_RANGE = 1000;
    /**
     * 默认的加锁重试次数
     * */
    private static final int DEFAULT_RETRY_COUNT = 30;

    /**
     * lockCount Key前缀
     * */
    private static final String LOCK_COUNT_KEY_PREFIX = "lock_count:";

    public static DistributeLock getInstance(){
        return instance;
    }
}

3.2 加锁实现

　　使用redis实现分布式锁时，加锁操做必须是原子操做，不然多客户端并发操做时会致使各类各样的问题。详情请见：Redis分布式锁的正确实现方式。

　　因为咱们实现的是可重入锁，加锁过程当中须要判断客户端ID的正确与否。而redis原生的简单接口无法保证一系列逻辑的原子性执行，所以采用了lua脚原本实现加锁操做。lua脚本可让redis在执行时将一连串的操做以原子化的方式执行。

加锁lua脚本 lock.lua

-- 获取参数
local requestIDKey = KEYS[1]

local currentRequestID = ARGV[1]
local expireTimeTTL = ARGV[2]

-- setnx 尝试加锁
local lockSet = redis.call('hsetnx',KEYS[1],'lockKey',currentRequestID)

if lockSet == 1
then
    -- 加锁成功 设置过时时间和重入次数=1
    redis.call('expire',KEYS[1],expireTimeTTL)
    redis.call('hset',KEYS[1],'lockCount',1)
    return 1
else
    -- 判断是不是重入加锁
    local oldRequestID = redis.call('hget',KEYS[1],'lockKey')
    if currentRequestID == oldRequestID
    then
        -- 是重入加锁
        redis.call('hincrby',KEYS[1],'lockCount',1)
        -- 重置过时时间
        redis.call('expire',KEYS[1],expireTimeTTL)
        return 1
    else
        -- requestID不一致，加锁失败
        return 0
    end
end

加锁方法实现：

　　加锁时，经过判断eval的返回值来判断加锁是否成功。

   @Override
    public String lock(String lockKey) {
        String uuid = UUID.randomUUID().toString();

        return lock(lockKey,uuid);
    }

    @Override
    public String lock(String lockKey, int expireTime) {
        String uuid = UUID.randomUUID().toString();

        return lock(lockKey,uuid,expireTime);
    }

    @Override
    public String lock(String lockKey, String requestID) {
        return lock(lockKey,requestID,DEFAULT_EXPIRE_TIME_SECOND);
    }

    @Override
    public String lock(String lockKey, String requestID, int expireTime) {
        RedisClient redisClient = RedisClient.getInstance();

        List<String> keyList = Arrays.asList(
                lockKey
        );

        List<String> argsList = Arrays.asList(
                requestID,
                expireTime + ""
        );
        Long result = (Long)redisClient.eval(LuaScript.LOCK_SCRIPT, keyList, argsList);

        if(result.equals(ADD_LOCK_SUCCESS)){
            return requestID;
        }else{
            return null;
        }
    }

3.3 解锁实现

　　解锁操做一样须要一连串的操做，因为原子化操做的需求，所以一样使用lua脚本实现解锁功能。

解锁lua脚本 unlock.lua

-- 获取参数
local requestIDKey = KEYS[1]

local currentRequestID = ARGV[1]

-- 判断requestID一致性
if redis.call('hget',KEYS[1],'lockKey') == currentRequestID
then
    -- requestID相同，重入次数自减
    local currentCount = redis.call('hincrby',KEYS[1],'lockCount',-1)
    if currentCount == 0
    then
        -- 重入次数为0，删除锁
        redis.call('del',KEYS[1])
        return 1
    else
        return 0 end
else 
    return 0 end

 

解锁方法实现：

   @Override
    public boolean unLock(String lockKey, String requestID) {
        List<String> keyList = Arrays.asList(
                lockKey
        );

        List<String> argsList = Collections.singletonList(requestID);

        Object result = RedisClient.getInstance().eval(LuaScript.UN_LOCK_SCRIPT, keyList, argsList);

        // 释放锁成功
        return RELEASE_LOCK_SUCCESS.equals(result);
    }

3.4 自动重试机制实现

　　调用lockAndRetry方法进行加锁时，若是加锁失败，则当前客户端线程会短暂的休眠一段时间，并进行重试。在重试了必定的次数后，会终止重试加锁操做，从而加锁失败。

　　须要注意的是，加锁失败以后的线程休眠时长是"固定值 + 随机值"，引入随机值的主要目的是防止高并发时大量的客户端在几乎同一时间被唤醒并进行加锁重试，给redis服务器带来周期性的、没必要要的瞬时压力。

    @Override
    public String lockAndRetry(String lockKey) {
        String uuid = UUID.randomUUID().toString();

        return lockAndRetry(lockKey,uuid);
    }

    @Override
    public String lockAndRetry(String lockKey, String requestID) {
        return lockAndRetry(lockKey,requestID,DEFAULT_EXPIRE_TIME_SECOND);
    }

    @Override
    public String lockAndRetry(String lockKey, int expireTime) {
        String uuid = UUID.randomUUID().toString();

        return lockAndRetry(lockKey,uuid,expireTime);
    }

    @Override
    public String lockAndRetry(String lockKey, int expireTime, int retryCount) {
        String uuid = UUID.randomUUID().toString();

        return lockAndRetry(lockKey,uuid,expireTime,retryCount);
    }

    @Override
    public String lockAndRetry(String lockKey, String requestID, int expireTime) {
        return lockAndRetry(lockKey,requestID,expireTime,DEFAULT_RETRY_COUNT);
    }

    @Override
    public String lockAndRetry(String lockKey, String requestID, int expireTime, int retryCount) {
        if(retryCount <= 0){
            // retryCount小于等于0 无限循环，一直尝试加锁
            while(true){
                String result = lock(lockKey,requestID,expireTime);
                if(result != null){
                    return result;
                }

                // 休眠一会
                sleepSomeTime();
            }
        }else{
            // retryCount大于0 尝试指定次数后，退出
            for(int i=0; i<retryCount; i++){
                String result = lock(lockKey,requestID,expireTime);
                if(result != null){
                    return result;
                }

                // 休眠一会
                sleepSomeTime();
            }

            return null;
        }
    }

4.使用注解切面简化redis分布式锁的使用

　　经过在方法上引入RedisLock注解切面，让对应方法被redis分布式锁管理起来，能够简化redis分布式锁的使用。

切面注解 RedisLock 

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface RedisLock {
    /**
     * 无限重试
     * */
    int UN_LIMIT_RETRY = RedisDistributeLock.UN_LIMIT_RETRY;

    String lockKey();
    int expireTime();
    int retryCount();
}

RedisLock 切面实现

@Component
@Aspect
public class RedisLockAspect {

    private static final Logger LOGGER = LoggerFactory.getLogger(RedisLockAspect.class);

    private static final ThreadLocal<String> REQUEST_ID_MAP = new ThreadLocal<>();

    @Pointcut("@annotation(annotation.RedisLock)")
    public void annotationPointcut() {
    }

    @Around("annotationPointcut()")
    public Object around(ProceedingJoinPoint joinPoint) throws Throwable {
        MethodSignature methodSignature = (MethodSignature)joinPoint.getSignature();
        Method method = methodSignature.getMethod();
        RedisLock annotation = method.getAnnotation(RedisLock.class);

        boolean lockSuccess = lock(annotation);
        if(lockSuccess){
            Object result = joinPoint.proceed();
            unlock(annotation);
            return result;
        }
        return null;
    }

    /**
     * 加锁
     * */
    private boolean lock(RedisLock annotation){
        DistributeLock distributeLock = RedisDistributeLock.getInstance();

        int retryCount = annotation.retryCount();

        String requestID = REQUEST_ID_MAP.get();
        if(requestID != null){
            // 当前线程 已经存在requestID
            distributeLock.lockAndRetry(annotation.lockKey(),requestID,annotation.expireTime(),retryCount);
            LOGGER.info("重入加锁成功 requestID=" + requestID);

            return true;
        }else{
            // 当前线程 不存在requestID
            String newRequestID = distributeLock.lockAndRetry(annotation.lockKey(),annotation.expireTime(),retryCount);

            if(newRequestID != null){
                // 加锁成功，设置新的requestID
                REQUEST_ID_MAP.set(newRequestID);
                LOGGER.info("加锁成功 newRequestID=" + newRequestID);

                return true;
            }else{
                LOGGER.info("加锁失败，超太重试次数，直接返回 retryCount={}",retryCount);

                return false;
            }
        }
    }

    /**
     * 解锁
     * */
    private void unlock(RedisLock annotation){
        DistributeLock distributeLock = RedisDistributeLock.getInstance();
        String requestID = REQUEST_ID_MAP.get();
        if(requestID != null){
            // 解锁成功
            boolean unLockSuccess = distributeLock.unLock(annotation.lockKey(),requestID);
            if(unLockSuccess){
                // 移除 ThreadLocal中的数据
                REQUEST_ID_MAP.remove();
                LOGGER.info("解锁成功 requestID=" + requestID);
            }
        }
    }
}

使用例子

@Service("testService")
public class TestServiceImpl implements TestService {

    @Override
    @RedisLock(lockKey = "lockKey", expireTime = 100, retryCount = RedisLock.UN_LIMIT_RETRY)
    public String method1() {
        return "method1";
    }

    @Override
    @RedisLock(lockKey = "lockKey", expireTime = 100, retryCount = 3)
    public String method2() {
        return "method2";
    }
}

5.总结

5.1 当前版本缺陷

主从同步可能致使锁的互斥性失效

　　在redis主从结构下，出于性能的考虑，redis采用的是主从异步复制的策略，这会致使短期内主库和从库数据短暂的不一致。

　　试想，当某一客户端刚刚加锁完毕，redis主库尚未来得及和从库同步就挂了，以后从库中新选拔出的主库是没有对应锁记录的，这就可能致使多个客户端加锁成功，破坏了锁的互斥性。

休眠并反复尝试加锁效率较低

　　lockAndRetry方法在客户端线程加锁失败后，会休眠一段时间以后再进行重试。当锁的持有者持有锁的时间很长时，其它客户端会有大量无效的重试操做，形成系统资源的浪费。

　　进一步优化时，可使用发布订阅的方式。这时加锁失败的客户端会监听锁被释放的信号，在锁真正被释放时才会进行新的加锁操做，从而避免没必要要的轮询操做，以提升效率。

不是一个公平的锁

　　当前实现版本中，多个客户端同时对锁进行抢占时，是彻底随机的，既不遵循先来后到的顺序，客户端之间也没有加锁的优先级区别。

　　后续优化时能够提供一个建立公平锁的接口，能指定加锁的优先级，内部使用一个优先级队列维护加锁客户端的顺序。公平锁虽然效率稍低，但在一些场景能更好的控制并发行为。

5.2 经验总结

　　前段时间看了一篇关于redis分布式锁的技术文章，发现本身对于分布式锁的了解还颇有限。纸上得来终觉浅，为了更好的掌握相关知识，决定尝试着本身实现一个demo级别的redis分布式锁，经过此次实践，更进一步的学习了lua语言和redis相关内容。

　　这篇博客的完整代码在个人github上：https://github.com/1399852153/RedisDistributedLock，存在许多不足之处，请多多指教。