并发场景下的幂等问题——分布式锁详解

private void renewExpiration() {
    ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
    if (ee == null) {
        return;
    }


    Timeout task = commandExecutor.getConnectionManager().newTimeout(timeout -> {
        ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());
        if (ent == null) {
            return;
        }
        Long threadId = ent.getFirstThreadId();
        if (threadId == null) {
            return;
        }


        RFuture<Boolean> future = renewExpirationAsync(threadId);
        future.onComplete((res, e) -> {
            if (e != null) {
                log.error("Can't update lock " + getRawName() + " expiration", e);
                EXPIRATION_RENEWAL_MAP.remove(getEntryName());
                return;
            }


            if (res) {
                // reschedule itself
                renewExpiration();
            } else {
                cancelExpirationRenewal(null);
            }
        });
    }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);


    ee.setTimeout(task);
}

protected RFuture<Boolean> renewExpirationAsync(long threadId) {
    return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                          "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                          "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                          "return 1; " +
                          "end; " +
                          "return 0;",
                          Collections.singletonList(getRawName()),
                          internalLockLeaseTime, getLockName(threadId));
}

public class MdbDistributeLock implements DistributeLock {


    /**
     * 锁的命名空间
     */
    private final int namespace;


    /**
     * 锁对应的缓存key
     */
    private final String lockName;


    /**
     * 锁的唯一标识，保证可重入，以应对put成功，但是返回超时的情况
     */
    private final String lockId;


    /**
     * 是否持有锁。true：是
     */
    private boolean locked;


    /**
     * 缓存实例
     */
    private final TairManager tairManager;


    public MdbDistributeLock(TairManager tairManager, int namespace, String lockCacheKey) {


        this.tairManager = tairManager;
        this.namespace = namespace;
        this.lockName = lockCacheKey;
        this.lockId = UUID.randomUUID().toString();
    }


    @Override
    public boolean tryLock() {


        try {
            //获取锁状态
            Result<DataEntry> getResult = null;
            ResultCode getResultCode = null;
            for (int cnt = 0; cnt < DEFAULT_RETRY_TIMES; cnt++) {
                getResult = tairManager.get(namespace, lockName);
                getResultCode = getResult == null ? null : getResult.getRc();
                if (noNeedRetry(getResultCode)) {
                    break;
                }
            }


            //重入，已持有锁，返回成功
            if (ResultCode.SUCCESS.equals(getResultCode)
                && getResult.getValue() != null && lockId.equals(getResult.getValue().getValue())) {
                locked = true;
                return true;
            }


            //不可获取锁，返回失败
            if (!ResultCode.DATANOTEXSITS.equals(getResultCode)) {
                log.error("tryLock fail code={} lock={} traceId={}", getResultCode, this, EagleEye.getTraceId());
                return false;
            }


            //尝试获取锁
            ResultCode putResultCode = null;
            for (int cnt = 0; cnt < DEFAULT_RETRY_TIMES; cnt++) {
                putResultCode = tairManager.put(namespace, lockName, lockId, MDB_CACHE_VERSION,
                    DEFAULT_EXPIRE_TIME_SEC);
                if (noNeedRetry(putResultCode)) {
                    break;
                }
            }
            if (!ResultCode.SUCCESS.equals(putResultCode)) {
                log.error("tryLock fail code={} lock={} traceId={}", getResultCode, this, EagleEye.getTraceId());
                return false;
            }
            locked = true;
            return true;


        } catch (Exception e) {
            log.error("DistributedLock.tryLock fail lock={}", this, e);
        }
        return false;
    }


    @Override
    public void unlock() {


        if (!locked) {
            return;
        }
        ResultCode resultCode = tairManager.invalid(namespace, lockName);
        if (!resultCode.isSuccess()) {
            log.error("DistributedLock.unlock fail lock={} resultCode={} traceId={}", this, resultCode,
                EagleEye.getTraceId());
        }
        locked = false;
    }


    /**
     * 判断是否需要重试
     *
     * @param resultCode 缓存的返回码
     * @return true：不用重试
     */
    private boolean noNeedRetry(ResultCode resultCode) {
        return resultCode != null && !ResultCode.CONNERROR.equals(resultCode) && !ResultCode.TIMEOUT.equals(
            resultCode) && !ResultCode.UNKNOW.equals(resultCode);
    }


}

public class MdbDistributeLockFactory implements DistributeLockFactory {


    /**
     * 缓存的命名空间
     */
    @Setter
    private int namespace;


    @Setter
    private MultiClusterTairManager mtairManager;


    @Override
    public DistributeLock getLock(String lockName) {
        return new MdbDistributeLock(mtairManager, namespace, lockName);
    }
}