基于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();
    }
}

所依赖的工具类：

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());
        }
    }
}

初始化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();
        }
    }
}

单例的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 lockcountkey = keys[2]

local currentrequestid = argv[1]
local expiretimettl = argv[2]

-- setnx 尝试加锁
local lockset = redis.call('setnx',requestidkey,currentrequestid)

if lockset == 1
then
    -- 加锁成功 设置过期时间和重入次数
    redis.call('expire',requestidkey,expiretimettl)
    redis.call('set',lockcountkey,1)
    redis.call('expire',lockcountkey,expiretimettl)
    return 1
else
    -- 判断是否是重入加锁
    local oldrequestid = redis.call('get',requestidkey)
    if currentrequestid == oldrequestid
    then
        -- 是重入加锁
        redis.call('incr',lockcountkey)
        -- 重置过期时间
        redis.call('expire',requestidkey,expiretimettl)
        redis.call('expire',lockcountkey,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,
                lock_count_key_prefix + 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 lockcountkey = keys[2]

local currentrequestid = argv[1]

-- 判断requestid一致性
if redis.call('get', requestidkey) == currentrequestid
then
    -- requestid相同，重入次数自减
    local currentcount = redis.call('decr',lockcountkey)
    if currentcount == 0
    then
        -- 重入次数为0，删除锁
        redis.call('del',requestidkey)
        redis.call('del',lockcountkey)
        return 1
    else
        return 0 end
else 
    return 0 end

解锁方法实现：

   @override
    public boolean unlock(string lockkey, string requestid) {
        list<string> keylist = arrays.aslist(
                lockkey,
                lock_count_key_prefix + 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，存在许多不足之处，请多多指教。