redis 分布式锁

1.在redis中的命令语句中，命令是忽略大小写的，而key是不忽略大小写的。

• 字符串
• Hash类型
• List
• Set
• SortedSet（zset）

string 类型 set key value

get key

hash类型 hset

set 类型 sadd

zset 是排序set集合

高并发情况下 setnx

1.场景: 在并发情况下 可能有线程1 把线程2的锁进行释放

解决： 每个线程 在lockkey 的value 设置自己的单独的value 和设置过期时间

2.在时间过期后 任务还未执行完 如何处理

luan 脚本 eval

加锁 解锁

重入锁的逻辑

redisson 是非公平锁

线程1抢到锁之后 会有监听器来监听执行 timetask 执行续命

线程2 会间隙性去判断是是否可以抢锁

双写 不一致

mvcc

分段锁

读写锁

eq：

    @RequestMapping("/testStock")
    public String testStock() {
        String lockKey = "lock:product_101";
        //获取锁对象
        RLock redissonLock = redisson.getLock(lockKey);
        //加分布式锁
        redissonLock.lock();  //  .setIfAbsent(lockKey, clientId, 30, TimeUnit.SECONDS);
        try {
            // 业务逻辑
        } finally {
            redissonLock.unlock();
        }
        return "end";
    }

// name 是 lockkey 
public RLock getLock(String name) {
    // 构造一个  RedissonLock 对象  internalLockLeaseTime 为锁租用时间 30s 
    return new RedissonLock(connectionManager.getCommandExecutor(), name);
}

public void lock() {
    try {
        lockInterruptibly();
    } catch (InterruptedException e) {
        // todo 为啥会有打断标志
        Thread.currentThread().interrupt();
    }
}

   // 第一步 : leaseTime  -1 unit null
   @Override
    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;
        }
		// 线程2走到这里来了  redisson_lock__channel  subscribe 订阅 
        // 订阅锁释放事件，并通过await方法阻塞等待锁释放，有效的解决了无效的锁申请浪费资源的问题
        RFuture future = subscribe(threadId);
        commandExecutor.syncSubscription(future);

        try {
            while (true) {
                ttl = tryAcquire(leaseTime, unit, threadId);
                // lock acquired
                if (ttl == null) {
                    break;
                }

                // waiting for message
                // 这个方法是 客户端2 等待ttl 秒 再去看锁是否被持有 todo  future的用法
                if (ttl >= 0) {
                    getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } else {
                    getEntry(threadId).getLatch().acquire();
                }
            }
        } finally {
            unsubscribe(future, threadId);
        }
//        get(lockAsync(leaseTime, unit));
    }

//  第一步 :  leaseTime  -1 unit null
private  RFuture tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {
    if (leaseTime != -1) {
         // 由于第一次进来 所以不走这个方法
        return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
    }
    // getLockWatchdogTimeout 默认时间为30s
    RFuture ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
     // 持有锁的 线程 加了一个Listener   
    // 我的理解是Future异步任务 异步任务完成的时候 会调用Listener回调方法 
    // future.isSuccess() 任务完成 future.getNow() 返回值(callback)  
    // scheduleExpirationRenewal(threadId); 这个是主要的看门狗方法
    ttlRemainingFuture.addListener(new FutureListener() {
        @Override
        public void operationComplete(Future future) throws Exception {
            if (!future.isSuccess()) {
                return;
            }

            Long ttlRemaining = future.getNow();
            // lock acquired
            if (ttlRemaining == null) {
                scheduleExpirationRenewal(threadId);
            }
        }
    });
    return ttlRemainingFuture;
}

// leaseTime 为 30s  unit 为 秒 单位 command为eval 
 RFuture tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand command) {
    internalLockLeaseTime = unit.toMillis(leaseTime);
	
    //   KEYS[1] 为 lockkeyName  ARGV[1] 为 id:线程id  ARGV[2]: 持有时间 默认30s
    //   exists  判断key是否存在 不存在 设置值 和 时间 
    //   hexists 判断key val 是否一致 一致的话   当前线程 值 加 1
    //   pttl key 以毫秒为单位返回 key 的剩余生存时间 当 key 不存在时，返回 -2 。 当 key 存在但没有设置剩余生存时间时，返回        -1 。
    //   否则，以毫秒为单位，返回 key 的剩余生存时间。     
    return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
              "if (redis.call('exists', KEYS[1]) == 0) then " +
                  "redis.call('hset', KEYS[1], ARGV[2], 1); " +
                  "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                  "return nil; " +
              "end; " +
              "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                  "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                  "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                  "return nil; 	" +
              "end; " +
              "return redis.call('pttl', KEYS[1]);",
                Collections.