分布式锁抽象化的一种思路

 

    场景：

      最近遇到一个account dubbo服务的并发注册问题，日志中抛出了大量的主键冲突问题。注册用户过程中有多个SQL操作，且这些SQL可以实现单机本地事务。

 

      为何单机锁不能解决问题？正如下图请求是发送到不同的节点上的，单机的锁只能控制单节点的并发请求。

      
 

    

    解决思路：

      如果加锁操作，会极大的减少主键冲突的日志。开始写了一版最直接的思路，后来发现dubbo服务中很多地方都可能出现并发问题，肖老板指导说可以做的更抽象点，如果jar包的形式提供出来，那么别的工程也可以复用这套逻辑了。所以后来又做了一版抽象，模型主要分成幂等锁和非幂等锁。

 

    系统架构图：

      
      
      cache能实现分布式锁的原因是所有的putIfAbsent这种操作都放在了单机节点的队列上做串行操作，所以也就避免了分布式锁的问题。

 

    流程图：

      非幂等的情况很简单，没有抢到锁直接返回就是。幂等的情况比较复杂，从下面的流程图可以看出有轮询的过程。

      
      
 

 

    缺点：

      这个也是我看了别人的分布式锁框架对比发现的，跟别人一对比发现还是有挺多东西可以学习借鉴下的。

      1. 对代码侵入性大，最好能直接提供注解的方式，不影响业务逻辑代码。对非幂等的接口比较好处理，但是对于幂等的接口则难度较大，毕竟需要有地方承载轮询的代码啊。

      2. 实现单一，仅仅实现nkv，其实可以更抽象一层,然后以nkv，redis zk的方式实现。

      3. 我理解幂等其实一般都不用分布式锁了，即使并发请求也没啥关系啊，我们这是奇葩情况。

 

    核心代码：

    

public interface ConcurrencyService {


    /** 幂等的并发返回的枚举值定义 */

    /** 获取到并发锁 */
    final int GET_CONCURRENCY_LOCK = 1;

    /** 轮询事务对象成功 */
    final int GET_TRANSACTION_RET_OBJECT = 2;

    /** 轮询事务对象失败 */
    final int NOT_GET_TRANSACTION_RET_OBJECT = 3;

    /**
     * 获取分布式并发的锁(非幂等）
     * @param key
     *    请求并发锁的标识
     * @return
     */
    public boolean getConcurrencyLock(String key) ;


    /**
     * 获取分布式并发的锁（幂等）
     * @param key
     *      请求并发锁的标识
     * @param interval
     *      轮询的间隔时间(ms)，默认是20ms
     * @param concurrencyLoopCallBack
     *      用于轮询抢到分布式锁的事务是否成功的回调方法
     * @return
     */
    public int getConcurrencyLock(String key,Long interval, IConcurrencyLoopCallBack concurrencyLoopCallBack) ;
}

 

   

public interface IConcurrencyLoopCallBack {

    /**
     * 用于并发幂等的场景，方法主要用于第三步
     *   1. 查询对象不存在
     *   2. 获取分布式锁
     *   3. 获取锁失败则使用这个方法来进行轮询
     *   4. 返回true 则说明验证事务操作成功
     *   5. 返回false 则说明轮询失败针对具体业务做不同处理
     * @return
     */
    boolean getTransactionRetObject();

}

    

   

@Service("concurrencyService")
public class ConcurrencyServiceImpl extends CacheSupport<Object> implements ConcurrencyService {

    /** 缓存在多久之后失效（单位s) */
    public final Integer overDate = 5;

    /** 默认的轮询间隔时间 */
    public final Long defaultInterval = 20L;

    /** 轮询次数 */
    public final int tryTimes = 3;

    @Override
    public boolean getConcurrencyLock(String key) {
        NkvClient.NkvOption nkvOption = new NkvClient.NkvOption();
        nkvOption.setExpireTime(overDate);
        return setIfNotExist(key, true, nkvOption);
    }

    @Override
    public int getConcurrencyLock(String key, Long interval, IConcurrencyLoopCallBack callBack) {
        boolean transactionRet = false;
        try {
            // 获取并发控制权
            boolean getLock = getConcurrencyLock(key);
            LogConstant.accountLog.info("cacheKey:" + key + " ,getConcurrencyLock result :" + getLock);
            if(getLock) {
                return GET_CONCURRENCY_LOCK;
            }

            // 设置默认间隔时间
            if(null == interval) {
                interval = defaultInterval;
            }

            int tryingTime = 0;
            while(!transactionRet && tryingTime < tryTimes) {
                Thread.sleep(interval);
                tryingTime ++;
                transactionRet = callBack.getTransactionRetObject();
                LogConstant.accountLog.info("cacheKey:" + key + "the " + tryingTime + "th try , transaction is successful ? " + transactionRet);
            }
        } catch (Exception e) {
            LogConstant.accountLog.error("getConcurrencyLock Exception", e);
        }
        if(transactionRet) {
            return GET_TRANSACTION_RET_OBJECT;
        }
        return NOT_GET_TRANSACTION_RET_OBJECT;
    }
}