set balance=1100,version=version+1 where id=#{id} and version=#{version};

int countLimit = select count from limit where id = ${id};
if(countlimit>0){
    set balance=1100,version=version+1 where id=#{id} and version=#{version};
}
update count;

private boolean acquireLock(Jedis j,String lock) throws Exception{
    int timeOut = timeoutSeconds*1000;
    boolean acquired = false;
    long start = System.currentTimeMillis();
    int times = 0;
    do {
        String value = String.valueOf(System.currentTimeMillis() + timeOut + 1);
        // 第一个得到这个锁
        if (j.setnx(lock, value) == 1) {
            logger.info("第一次获取全局锁:{} 成功", lock);
            acquired = true;
            break;
        }
        // j.expire(lock, timeoutSeconds); 网络抖动，可能失败
        String currentValue = j.get(lock);
        // 小于时，可能是上次没有清除，自上次超时后没有别的线程操作过
        if (currentValue != null && Long.valueOf(currentValue) < System.currentTimeMillis()) {
            // 这是同步操作，只会一个成功
            String oldValue = j.getSet(lock, value);
            // 别的线程没有赋上值，当前成功得到锁
            if (oldValue != null && oldValue.equals(currentValue)) {
                acquired = true;
                logger.info("获取全局锁:{} 成功，尝试了{}次,经过了{}ms",lock,times,System.currentTimeMillis()-start);
                break;
            }
        }
        times++;
        Thread.sleep(100);
    } while (start + timeOut > System.currentTimeMillis());
    if(!acquired){
        logger.info("获取全局锁:{} 失败，尝试了{}次",lock,times);
    }            
    return acquired;
}

private void releaseLock(Jedis j,String lock){
    String currentValue = j.get(lock);
    if(currentValue != null){
        if(System.currentTimeMillis() < Long.valueOf(currentValue) ){
            j.del(lock);
            logger.info("释放锁{}",lock);
        }
    }
}

Long result = jedis.setnx(lockKey, value); 
 if (result == 1) { 
     // 若在这里程序突然崩溃，则无法设置过期时间，将发生死锁 
     jedis.expire(lockKey, expireTime); 
 }

public static void releaseLock(Jedis jedis, String lockKey, String requestId) { 
    // 判断加锁与解锁是不是同一个客户端 
    if (requestId.equals(jedis.get(lockKey))) { 
        // 若在此时，这把锁突然不是这个客户端的，则会误解锁 
        jedis.del(lockKey);
    } 
}

Lock lock = DistributedReentrantLock.newLock("testlock11");//定义testlock11为key的锁，默认可重入锁
if(lock.tryLock()){
    try{
     xxxxxx
    }finally{
      lock.unlock(); //释放testlock11为key的锁，释放需要放在finally里，防止出异常导致锁没有及时释放
    }
  }

/**
 * 尝试获取分布式锁
 * @param jedis Redis客户端
 * @param lockKey 锁
 * @param requestId 请求标识
 * @param expireTime 超期时间
 * @return 是否获取成功
 */
public static boolean tryGetDistributedLock(Jedis jedis, String lockKey, String requestId, int expireTime) {
    String result = jedis.set(lockKey, requestId, "NX", "PX", expireTime);
    if (LOCK_SUCCESS.equals(result)) {
        return true;
    }
    return false;
}

/**
 * 释放分布式锁
 * @param jedis Redis客户端
 * @param lockKey 锁
 * @param requestId 请求标识
 * @return 是否释放成功
 */
public static boolean releaseDistributedLock(Jedis jedis, String lockKey, String requestId) {
    String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
    Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));
    if (RELEASE_SUCCESS.equals(result)) {
        return true;
    }
    return false;
}

public interface RLock extends Lock, RExpirable, RLockAsync

public void lock() ;
public void lock(long leaseTime, TimeUnit unit) ;

public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException {
    long threadId = Thread.currentThread().getId();
    Long ttl = tryAcquire(leaseTime, unit, threadId);
    // lock acquired
    if (ttl == null) {
        return;
    }
    RFuture<RedissonLockEntry> future = subscribe(threadId);
    commandExecutor.syncSubscription(future);
    try {
        while (true) {
            ttl = tryAcquire(leaseTime, unit, threadId);
            // lock acquired
            if (ttl == null) {
                break;
            }
            // waiting for message
            if (ttl >= 0) {
                getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
            } else {
                getEntry(threadId).getLatch().acquire();
            }
        }
    } finally {
        unsubscribe(future, threadId);
    }
//        get(lockAsync(leaseTime, unit));
}

1. private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {
2.     if (leaseTime != -1) {
3.         return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
4.     }
5.     RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
6.     ttlRemainingFuture.addListener(new FutureListener<Long>() {
7.         @Override
8.         public void operationComplete(Future<Long> future) throws Exception {
9.             if (!future.isSuccess()) {
10.                 return;
11.             }
12. 
13.             Long ttlRemaining = future.getNow();
14.             // lock acquired
15.             if (ttlRemaining == null) {
16.                 scheduleExpirationRenewal(threadId);
17.             }
18.         }
19.     });
20.     return ttlRemainingFuture;
21. }
• 1.根据锁的持续时间不同，处理也不同
• 2.没有设置持续时间，那就是阻塞型，一直等待
• 2.1.为了防止业务方法执行时间超过锁timeout,则定时续约scheduleExpirationRenewal()
• 3.设置了持续时间，则不需要进行续约
23. private void scheduleExpirationRenewal(final long threadId) {
24.     if (expirationRenewalMap.containsKey(getEntryName())) {
25.         return;
26.     }
27.     Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
28.         @Override
29.         public void run(Timeout timeout) throws Exception {
30. 
31.             RFuture<Boolean> future = renewExpirationAsync(threadId);
32. 
33.             future.addListener(new FutureListener<Boolean>() {
34.                 @Override
35.                 public void operationComplete(Future<Boolean> future) throws Exception {
36.                     expirationRenewalMap.remove(getEntryName());
37.                     if (!future.isSuccess()) {
38.                         log.error("Can't update lock " + getName() + " expiration", future.cause());
39.                         return;
40.                     }
41. 
42.                     if (future.getNow()) {
43.                         // reschedule itself
44.                         scheduleExpirationRenewal(threadId);
45.                     }
46.                 }
47.             });
48.         }
49. 
50.     }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
51. 
52.     if (expirationRenewalMap.putIfAbsent(getEntryName(), new ExpirationEntry(threadId, task)) != null) {
53.         task.cancel();
54.     }
55. }
56. 
57. protected RFuture<Boolean> renewExpirationAsync(long threadId) {
58.     return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
59.             "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
60.                 "redis.call('pexpire', KEYS[1], ARGV[1]); " +
61.                 "return 1; " +
62.             "end; " +
63.             "return 0;",
64.         Collections.<Object>singletonList(getName()), 
65.         internalLockLeaseTime, getLockName(threadId));
66. }
67. 以internalLockLeaseTime/3间隔时间，定时续约
68. 如果当前client自身有并发时，通过putIfAbsent保证只有一个task
69. 续约：当lock存在时，使用pexpire设置过期时间
70.  RFuture tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand command) {
71.     internalLockLeaseTime = unit.toMillis(leaseTime);
72. 
73.     return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
74.               "if (redis.call('exists', KEYS[1]) == 0) then " +
75.                   "redis.call('hset', KEYS[1], ARGV[2], 1); " +
76.                   "redis.call('pexpire', KEYS[1], ARGV[1]); " +
77.                   "return nil; " +
78.               "end; " +
79.               "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
80.                   "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
81.                   "redis.call('pexpire', KEYS[1], ARGV[1]); " +
82.                   "return nil; " +
83.               "end; " +
84.               "return redis.call('pttl', KEYS[1]);",
85.                 Collections.singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
86. }
87. 
88. protected String getLockName(long threadId) { 
89.     return id + ":" + threadId;
90. }

unLock
1. @Override
2. public RFuture<Void> unlockAsync(final long threadId) {
3.     final RPromise<Void> result = new RedissonPromise<Void>();
4.     RFuture<Boolean> future = unlockInnerAsync(threadId);
5. 
6.     future.addListener(new FutureListener<Boolean>() {
7.         @Override
8.         public void operationComplete(Future<Boolean> future) throws Exception {
9.             if (!future.isSuccess()) {
10.                 cancelExpirationRenewal(threadId);
11.                 result.tryFailure(future.cause());
12.                 return;
13.             }
14. 
15.             Boolean opStatus = future.getNow();
16.             if (opStatus == null) {
17.                 IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
18.                         + id + " thread-id: " + threadId);
19.                 result.tryFailure(cause);
20.                 return;
21.             }
22.             if (opStatus) {
23.                 cancelExpirationRenewal(null);
24.             }
25.             result.trySuccess(null);
26.         }
27.     });
28. 
29.     return result;
30. }
31. 
32. void cancelExpirationRenewal(Long threadId) {
33.     ExpirationEntry task = expirationRenewalMap.get(getEntryName());
34.     if (task != null && (threadId == null || task.getThreadId() == threadId)) {
35.         expirationRenewalMap.remove(getEntryName());
36.         task.getTimeout().cancel();
37.     }
38. }
39. 从方法名看，虽然对外好像是直接解锁，但内部是异步执行的
40. unlockInnerAsync()进行解锁
41. 从expirationRenewalMap移除，并把task.cancel()
42. protected RFuture<Boolean> unlockInnerAsync(long threadId) {
43.     return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
44.             "if (redis.call('exists', KEYS[1]) == 0) then " +
45.                 "redis.call('publish', KEYS[2], ARGV[1]); " +
46.                 "return 1; " +
47.             "end;" +
48.             "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
49.                 "return nil;" +
50.             "end; " +
51.             "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
52.             "if (counter > 0) then " +
53.                 "redis.call('pexpire', KEYS[1], ARGV[2]); " +
54.                 "return 0; " +
55.             "else " +
56.                 "redis.call('del', KEYS[1]); " +
57.                 "redis.call('publish', KEYS[2], ARGV[1]); " +
58.                 "return 1; "+
59.             "end; " +
60.             "return nil;",
61.             Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId));
62. 
63. }
64. lockname不存在，说明已经解锁，publish channelname unlockmessage；return 1
65. lockname存在，但对于uuid+id不存在，说明不是加锁的client,return nil
66. lockname存在，并且是当前加锁client
67. 对lockname uuid+id进行-1，如果counter>0则走5，如果=0 则走6
68. counter>0 说明锁重入了，计数器-1，并expire
69. counter=0 说明最终解锁，直接del key，并publish channelname unlockmessage;return 1

redission缺陷
使用cluster时
一个场景：A在向主机1请求到锁成功后，主机1宕机了。现在从机1a变成了主机。但是数据没有同步，从机1a是没有A的锁的。那么B又可以获得一个锁。这样就会造成数据错误。
redlock主要思想就是做数据冗余。建立5台独立的集群，当我们发送一个数据的时候，要保证3台（n/2+1）以上的机器接受成功才算成功，否则重试或报错
redlock实现会更复杂，但从他的算法上看，有zk选举的味道。对于更高可用分布锁，可以借助zk本身特性去实现