redission

deipss · Oct 7, 2023 · e27564e · e27564e
1 parent 524161e
commit e27564e
Show file tree

Hide file tree

Showing 7 changed files with 360 additions and 2 deletions.
diff --git a/docs/Database/Redis.md b/docs/Database/Redis.md
@@ -104,6 +104,7 @@ redis会单独创建（fork）一个子进程来进行持久化，会将数据
 - hsetnx
 
 # 缓存问题
+
 - https://cloud.tencent.com/developer/article/1525256
 
 ## 穿透
@@ -113,14 +114,15 @@ redis会单独创建（fork）一个子进程来进行持久化，会将数据
 ### 应对方法
 
 - 入口加判断，如非法的参数拦截
-- 将这类请求作为key，缓存起来
+- 将这类请求作为key，缓存起来,可以设置TTL时间
 - 使用布隆过滤器，判断一个key是否已经查过了，如果已经查过了，就不去数据库查询。
 
 ## 击穿
 
 某个key,在特定时刻被大规模访问，过期后，后续的流量全部落到了DB上。
 
 - 对于热点数据，慎重考虑过期时间，确保热点期间key不会过期，甚至有些可以设置永不过期。
+- 基于逻辑过期应对击穿问题，写入redis时，不指定过期时间，在数据中添加过期时间，以数据中过期时间为准
 - 使用互斥锁（比如Java的多线程锁机制），第一个线程访问key的时候就锁住，等查询数据库返回后，把值插入到缓存后再释放锁，这样后面的请求就可以直接取缓存里面的数据了。
 
 ### 应对方法
@@ -145,4 +147,60 @@ Web 服务器、第三方 API 或任何其他可以返回数据的东西。
 2. 数据库由于巨大的 CPU 峰值发生崩溃，并导致超时错误。
 3. 收到超时错误后，所有的线程都会发起重试，从而导致另一次踩踏。
 4. 这个循环不断持续。
-    - https://www.infoq.cn/article/Bb2YC0yHVSz4qVwdgZmO
+    - https://www.infoq.cn/article/Bb2YC0yHVSz4qVwdgZmO
+
+# 应用场景
+
+- session 短信登陆
+- 商品信息缓存
+- 优惠秒杀：计时器、LUA脚本、分布式锁、redis消息队列
+- 附近的商户：GeoHash
+- UV统计：HyperLogLog
+- 用户签到、圈选：BitMAP
+- 好友关注：Set 交集、差集
+- 达人探店：SortSet的点赞排行榜
+
+# 问题讨论
+
+## 删除缓存还是更新缓存
+
+- ~~更新缓存~~：每次更新数据库时都更新缓存，无效写过多
+- 删除缓存：让缓存失效，查询时再重新加载新的信息到缓存中
+
+## 缓存与DB一致性
+
+- 单体系统：放在同一事务中
+- 分布式系统：使用TCC分布式事务
+
+## 先操作缓存，还是先操作DB
+
+- 先删除缓存，再删除DB，并发时（一个线程查询、一个线程删除）会存在不一致的情况
+- 先删除DB，再删除缓存，并发时（一个线程查询、一个线程删除）会存在不一致的情况，但比前者的概率低，因为DB的时间比缓存的操作时间长
+
+## 乐观锁应对库存超卖问题
+
+- 用户A读取出是100件，更新库存时添加查询条件 **where quantity=100**
+
+## 一人一单（单机）
+
+- 因为要insert,所以要使用悲观锁来应对这个问题
+- 这个悲观锁可以使用UserID，如
+
+```shell
+synchronized(userId.toString().intern)
+```
+
+- 悲观锁的释放应该在事务的提交后，不然事务未提交锁已释放，第2个线程读取的未提交事务前的数据，即第1个线程读取的数据
+- 注意事务失效的情况
+
+![transaction.png](img%2Ftransaction.png)
+
+## 一人单（集群）
+
+NX是互斥，EX是超时
+
+- SET lock UUID EX 10 NX
+- 锁超时自动释放后，他人获取了锁，自己释放锁时，判断UUID是不是自己的，不是自己的就不释放，避免删除他人持有的锁
+- 自己释放锁时，判断UUID是自己的，释放锁前，阻塞了（JVM FGC），此时锁超时自动释放，他人获取了锁，还是会导致误删
+
+> 判断锁和释放锁需要是原子性的，可以通过LUA脚本来处理
diff --git a/docs/Database/Redission.md b/docs/Database/Redission.md
@@ -0,0 +1,300 @@
+---
+layout: default
+title: Redission
+parent: Database
+---
+
+# 分布式锁需要优化的问题
+- 不可重入：同一个线程，依次执行两个方法都要获取同一把分布式锁，需要可重入锁
+- 不可重试：获取锁只尝试一次就返回false，缺少重试机制
+- 超时释放：业务执行时间长，导致锁释放，存在安全隐患
+- 主从一致性：主从集群，数据同步存在时延，主机宕机，从机还没有完成同步锁数据
+
+
+![lock.png](img%2Flock.png)
+
+# 可重入锁
+- 判断持有锁是不是自己
+- 是自己将上锁次数+1
+- 释放锁一次，上锁次数-1
+- 释放锁两次，上锁次数-1，为0，释放锁
+
+利用HASH结构记录线程ID和重入次数 
+
+```shell
+org.redisson.RedissonLock#tryLockInnerAsync 
+
+<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
+        internalLockLeaseTime = unit.toMillis(leaseTime);
+
+        return evalWriteAsync(getName(), LongCodec.INSTANCE, command,
+                "if (redis.call('exists', KEYS[1]) == 0) then " +
+                        "redis.call('hincrby', 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.singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
+    }
+
+  org.redisson.RedissonLock#unlockInnerAsync 
+
+     protected RFuture<Boolean> unlockInnerAsync(long threadId) {
+        return evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
+                "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
+                        "return nil;" +
+                        "end; " +
+                        "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
+                        "if (counter > 0) then " +
+                        "redis.call('pexpire', KEYS[1], ARGV[2]); " +
+                        "return 0; " +
+                        "else " +
+                        "redis.call('del', KEYS[1]); " +
+                        "redis.call('publish', KEYS[2], ARGV[1]); " +
+                        "return 1; " +
+                        "end; " +
+                        "return nil;",
+                Arrays.asList(getName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));
+    }
+```
+
+# 锁重试
+
+```shell
+org.redisson.RedissonLock#tryLock(long, long, java.util.concurrent.TimeUnit)
+不是单一的死循环，而是通过消息订阅和信号量的方式来节省资源，收到消息或信号量，才去重试获取锁 
+@Override
+    public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
+        long time = unit.toMillis(waitTime);
+        long current = System.currentTimeMillis();
+        long threadId = Thread.currentThread().getId();
+        Long ttl = tryAcquire(leaseTime, unit, threadId);
+        // lock acquired
+        if (ttl == null) {
+            return true;
+        }
+
+        time -= System.currentTimeMillis() - current;
+        if (time <= 0) {
+            acquireFailed(threadId);
+            return false;
+        }
+
+        current = System.currentTimeMillis();
+        RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId); // 订阅一个通知，这个通知就是锁释放时的通知 "redis.call('publish', KEYS[2], ARGV[1]); "
+        if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {
+            if (!subscribeFuture.cancel(false)) {
+                subscribeFuture.onComplete((res, e) -> {
+                    if (e == null) {
+                        unsubscribe(subscribeFuture, threadId);
+                    }
+                });
+            }
+            acquireFailed(threadId);
+            return false;
+        }
+
+        try {
+            time -= System.currentTimeMillis() - current;
+            if (time <= 0) {
+                acquireFailed(threadId);
+                return false;
+            }
+        
+            while (true) {
+                long currentTime = System.currentTimeMillis();
+                ttl = tryAcquire(leaseTime, unit, threadId);
+                // lock acquired
+                if (ttl == null) {
+                    return true;
+                }
+
+                time -= System.currentTimeMillis() - currentTime;
+                if (time <= 0) {
+                    acquireFailed(threadId);
+                    return false;
+                }
+
+                // waiting for message
+                currentTime = System.currentTimeMillis();
+                if (ttl >= 0 && ttl < time) {
+                    subscribeFuture.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
+                } else {
+                    subscribeFuture.getNow().getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
+                }
+
+                time -= System.currentTimeMillis() - currentTime;
+                if (time <= 0) {
+                    acquireFailed(threadId);
+                    return false;
+                }
+            }
+        } finally {
+            unsubscribe(subscribeFuture, threadId);
+        }
+//        return get(tryLockAsync(waitTime, leaseTime, unit));
+    }
+```
+
+# 锁超时
+- 获取锁之后，利用watchDog定时任务去续约
+```shell
+org.redisson.RedissonLock#tryAcquireAsync
+org.redisson.RedissonLock#scheduleExpirationRenewal
+org.redisson.RedissonLock#renewExpiration
+org.redisson.RedissonLock#scheduleExpirationRenewal
+private void renewExpiration() {
+        ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
+        if (ee == null) {
+            return;
+        }
+
+        Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
+            @Override
+            public void run(Timeout timeout) throws Exception {
+                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 " + getName() + " expiration", e);
+                        return;
+                    }
+                    
+                    if (res) {
+                        // reschedule itself
+                        renewExpiration();// 递归调用自身，会不断的续约，什么时候取消续约。在锁释放时             cancelExpirationRenewal(threadId);
+                    }
+                });
+            }
+        }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
+        
+        ee.setTimeout(task);
+    }
+    
+private void scheduleExpirationRenewal(long threadId) {
+        ExpirationEntry entry = new ExpirationEntry();
+        ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);// 续约时，锁的可重入
+        if (oldEntry != null) {
+            oldEntry.addThreadId(threadId);
+        } else {
+            entry.addThreadId(threadId);
+            renewExpiration();
+        }
+    }
+```
+
+# 主从一致性
+如下图所示：主节点宕机后，数据同步没有完成，导致新的主节点没有同步原主节点的数据，来上锁时，别的JVM上锁成功了。
+![distribute2.png](img%2Fdistribute2.png)
+所以不使用主从的形式来管理集群，使用同等节点的方式 **redissonClient.getMultiLock()**
+![distribute.png](img%2Fdistribute.png)
+- 所有节点要全部上锁
+- 一个失败，其他要全部释放
+- 全部上锁后，要续约一次，使首次上锁与末次上锁的释放时间保持一致
+- 
+
+```shell
+@Override
+org.redisson.RedissonMultiLock#tryLock(long, long, java.util.concurrent.TimeUnit)
+    public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
+//        try {
+//            return tryLockAsync(waitTime, leaseTime, unit).get();
+//        } catch (ExecutionException e) {
+//            throw new IllegalStateException(e);
+//        }
+        long newLeaseTime = -1;
+        if (leaseTime != -1) {
+            if (waitTime == -1) {
+                newLeaseTime = unit.toMillis(leaseTime);
+            } else {
+                newLeaseTime = unit.toMillis(waitTime)*2;
+            }
+        }
+        
+        long time = System.currentTimeMillis();
+        long remainTime = -1;
+        if (waitTime != -1) {
+            remainTime = unit.toMillis(waitTime);
+        }
+        long lockWaitTime = calcLockWaitTime(remainTime);
+        
+        int failedLocksLimit = failedLocksLimit();
+        List<RLock> acquiredLocks = new ArrayList<>(locks.size());
+        for (ListIterator<RLock> iterator = locks.listIterator(); iterator.hasNext();) {
+            RLock lock = iterator.next();
+            boolean lockAcquired;
+            try {
+                if (waitTime == -1 && leaseTime == -1) {
+                    lockAcquired = lock.tryLock();
+                } else {
+                    long awaitTime = Math.min(lockWaitTime, remainTime);
+                    lockAcquired = lock.tryLock(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS);
+                }
+            } catch (RedisResponseTimeoutException e) {
+                unlockInner(Arrays.asList(lock));
+                lockAcquired = false;
+            } catch (Exception e) {
+                lockAcquired = false;
+            }
+
+            if (lockAcquired) {
+                acquiredLocks.add(lock);
+            } else {
+                if (locks.size() - acquiredLocks.size() == failedLocksLimit()) {
+                    break;
+                }
+
+                if (failedLocksLimit == 0) {
+                    unlockInner(acquiredLocks);
+                    if (waitTime == -1) {
+                        return false;
+                    }
+                    failedLocksLimit = failedLocksLimit();
+                    acquiredLocks.clear();
+                    // reset iterator
+                    while (iterator.hasPrevious()) {
+                        iterator.previous();
+                    }
+                } else {
+                    failedLocksLimit--;
+                }
+            }
+
+            if (remainTime != -1) {
+                remainTime -= System.currentTimeMillis() - time;
+                time = System.currentTimeMillis();
+                if (remainTime <= 0) {
+                    unlockInner(acquiredLocks);
+                    return false;
+                }
+            }
+        }
+
+        if (leaseTime != -1) {
+            List<RFuture<Boolean>> futures = new ArrayList<>(acquiredLocks.size());
+            for (RLock rLock : acquiredLocks) {
+                RFuture<Boolean> future = ((RedissonLock) rLock).expireAsync(unit.toMillis(leaseTime), TimeUnit.MILLISECONDS);
+                futures.add(future);
+            }
+            
+            for (RFuture<Boolean> rFuture : futures) {
+                rFuture.syncUninterruptibly();
+            }
+        }
+        
+        return true;
+    }
+```
diff --git a/docs/Database/img/distribute.png b/docs/Database/img/distribute.png
diff --git a/docs/Database/img/distribute2.png b/docs/Database/img/distribute2.png
diff --git a/docs/Database/img/lock.png b/docs/Database/img/lock.png
diff --git a/docs/Database/img/redis.png b/docs/Database/img/redis.png
diff --git a/docs/Database/img/transaction.png b/docs/Database/img/transaction.png