缓存和分布式锁笔记_网站优化分享_上海毫米网络优化公司

您的位置：上海毫米网络优化公司 > 网站优化分享 >

概念

缓存的作用是减低对数据源的访问频率。从而提高我们系统的性能。
缓存和分布式锁笔记,第1张
缓存和分布式锁笔记,第2张
缓存的流程图
缓存和分布式锁笔记,第3张

缓存分类

本地缓存

把缓存数据存储在内存中(Map )，其实就是强引用，不会轻易被删除。
缓存和分布式锁笔记,第4张

缓存和分布式锁笔记,单体架构下的缓存处理,第5张

分布式缓存

数据冗余，效率不高
缓存和分布式锁笔记,分布式架构下本地缓存,第6张

缓存和分布式锁笔记,分布式缓存的结构图,第7张

整合Redis


    org.springframework.boot
    spring-boot-starter-redis
    1.3.2.RELEASE

spring:
    redis:
    host: 192.168.56.10
    port: 6379

@ResponseBody
@RequestMapping("/index/test")
public Object test(){
    BoundValueOperations boundValueOperations= stringRedisTemplate.boundValueOps("test");
    boundValueOperations.set("测试数据");
    return boundValueOperations.get();
}

redis缓存改造三级分类

public Map> getCatelog2JSON() throws IOException {
    String catalogJSON = stringRedisTemplate.opsForValue().get("Catelog2JSON");
    if (StringUtils.isEmpty(catalogJSON)) {
        Map> map = getCatelog2JSONDb();
        String json = objectMapper.writeValueAsString(map);
        stringRedisTemplate.opsForValue().set("Catelog2JSON", json);
        return map;
    }
    Map> map=objectMapper.readValue(catalogJSON, new TypeReference>>(){});
    return map;
}

缓存常见问题

缓存穿透

问题：
指查询一个一定不存在的数据，由于缓存是不命中，将去查询数据库，但是数据库也无此记录，我们没有将这次查询的null写入缓存，这将导致这个不存在的数据每次请求都要到存储层去查询，造成数据库压力增大。
缓存和分布式锁笔记,第8张
**解决：**直接把null结果缓存，并加入短暂的过期时间

缓存雪崩

**问题：**在我们设置缓存时key采用了相同的过期时间，导致缓存在某一时刻同时失效，请求全部转发到DB，DB瞬时压力过重雪崩。
缓存和分布式锁笔记,第9张
**解决：**原有的失效时间基础上增加一个随机值，比如1-5分钟随机，这样每一个缓存的过期时间的重复率就会降低，就很难引发集体失效的事件。

缓存击穿

**问题：**对于一些设置了过期时间的key，如果这些key可能会在某些时间点被超高并发地访问，是一种非常“热点”的数据。如果这个key在大量请求同时进来前正好失效，那么所有对这个key的数据查询都落到db。
缓存和分布式锁笔记,第9张
**解决：**加锁，大量并发只让一个去查，其他人等待，查到以后释放锁，其他人获取到锁，先查缓存，就会有数据，不用去db。

synchronized (this) {
  。。。。。。
}

缓存和分布式锁笔记,imagepng,第11张

public Map> getCatelog2JSON() throws IOException {
String catalogJSON = stringRedisTemplate.opsForValue().get("Catelog2JSON");
if (StringUtils.isEmpty(catalogJSON)) {
    synchronized (this) {
        System.out.println("查数据库啊-----");
        Map> map = getCatelog2JSONDb();
        if (null == map) {
            //加个null缓存，防止穿透
            stringRedisTemplate.opsForValue().set("Catelog2JSON", "null", 5, TimeUnit.SECONDS);
        } else {
            String json = objectMapper.writeValueAsString(map);
            //加个过期时间防止雪崩
            stringRedisTemplate.opsForValue().set("Catelog2JSON", json, 5, TimeUnit.SECONDS);
            return map;
        }
    }
}
Map> map = objectMapper.readValue(catalogJSON, new TypeReference>>() {
});
return map;
}

本地锁的局限

缓存和分布式锁笔记,本地锁在分布式环境下，是没有办法锁住其他节点的操作的,第12张
缓存和分布式锁笔记,第13张
**问题：**针对本地锁的问题，我们需要通过分布式锁来解决，那么是不是意味着本身锁在分布式场景下就不需要了呢？
**答：**因为如果分布式环境下的每个节点不控制请求的数量，那么分布式锁的压力会非常大，这时我们需要本地锁来控制每个节点的同步，来降低分布式锁的压力，所以实际开发中我们都是本地锁和分布式锁结合使用的。

分布式锁

分布式锁原理

缓存和分布式锁笔记,分布式锁或者本地锁的本质其实是一样的，都是将并行的操作转换为了串行的操作,第14张

分布式锁常用解决方案

https://blog.csdn.net/qq_33594592/article/details/133932876

数据库

通过在数据库中创建一个唯一索引的表，然后通过插入一条数据来获取锁，如果插入成功则获取锁成功，否则获取锁失败。释放锁的操作就是删除这条数据。这种方式的优点是实现简单，缺点是性能较低，因为涉及到数据库的操作。
可以利用MySQL隔离性：唯一索引

use test;
CREATE TABLE `DistributedLock` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '主键',
  `name` varchar(64) NOT NULL DEFAULT '' COMMENT '锁名',
  `desc` varchar(1024) NOT NULL DEFAULT '备注信息',
  `update_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '保存数据时间，自动生成',
  PRIMARY KEY (`id`),
  UNIQUE KEY `uidx_name` (`name`) 
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='锁定中的方法';
//数据库中的每一条记录就是一把锁，利用的mysql唯一索引的排他性
//lock和unlock只是函数名
lock(name,desc){
    insert into DistributedLock(`name`,`desc`) values (#{name},#{desc});
}
unlock(name){
    delete from DistributedLock where name = #{name}
}

乐观锁：乐观的任务数据不会出现数据安全问题，如果出现了就重试一次

select ...,version;
update table set version+1 where version = xxx

Redis

setNX: setNX(key,value) :如果key不存在那么就添加key的值，否则添加失败，Redisson。
缓存和分布式锁笔记,第15张

Zookeeper

缓存和分布式锁笔记,第16张
其基本思想是创建一个临时有序节点，临时节点保证在客户端断开连接或者超时zookeeper会自动删除该节点，从而避免死锁的发生，有序节点可以实现一种公平锁，加锁顺序按照客户端进程请求顺序进行，加锁的为产生节点序号最小的那个客户端。
加锁：客户端在zookeeper创建临时节点，并判断自己是不是节点序号最小的，如果是则加锁成功，如果不是则watch监听它前一个节点删除事件，当收到前一个节点删除通知时重复前面逻辑，直到加锁完成
解锁：客户端完成业务逻辑，在zookeeper删除自己的临时节点，同时通知它后一个节点获取锁
缓存和分布式锁笔记,两个客户端A和B加锁解锁的过程,第17张

Redis实现分布式锁

public Map> getCatelog2JSONDbWithRedisLock() {
    String keys = "catalogJSON";
    // 加锁
    Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", "1111");
    if(lock){
        // 加锁成功
        Map> data = getDataForDB(keys);
        // 从数据库中获取数据成功后，我们应该要释放锁
        stringRedisTemplate.delete("lock");
        return data;
    }else{
        // 加锁失败
        // 休眠+重试
        // Thread.sleep(1000);
        return getCatelog2JSONDbWithRedisLock();
    }
}

///如果getDataForDB出现异常，则会出现死锁，
///加上过期时间可以解决
// 给对应的key设置过期时间
stringRedisTemplate.expire("lock",20,TimeUnit.SECONDS);

///如果在expire方法执行之前就中断也会出现死锁
///创建锁时同时加上过期时间
// 加锁 在执行插入操作的同时设置了过期时间
Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", "1111",30,TimeUnit.SECONDS);

///如果获取锁的业务执行时间比较长，超过了我们设置的过期时间，那么就有可能业务还没执行完，锁就释放了，然后另一个请求进来了，并创建了key，这时原来的业务处理完成后，再去删除key的时候，那么就有可能删除别人的key,如何解决
///查询的锁的信息通过UUID来区分
public Map> getCatelog2JSONDbWithRedisLock() {
    String keys = "catalogJSON";
    // 加锁 在执行插入操作的同时设置了过期时间
    String uuid = UUID.randomUUID().toString();
    Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", uuid,30,TimeUnit.SECONDS);
    if(lock){
        // 给对应的key设置过期时间
        stringRedisTemplate.expire("lock",20,TimeUnit.SECONDS);
        // 加锁成功
        Map> data = getDataForDB(keys);
        // 获取当前key对应的值
        String val = stringRedisTemplate.opsForValue().get("lock");
        if(uuid.equals(val)){
            // 说明这把锁是自己的
            // 从数据库中获取数据成功后，我们应该要释放锁
            stringRedisTemplate.delete("lock");
        }
        return data;
    }else{
        // 加锁失败
        // 休眠+重试
        // Thread.sleep(1000);
        return getCatelog2JSONDbWithRedisLock();
    }
}
///key的值和删除key其实不是一个原子性操作，这就会出现我查询出来key之后，时间过期了，然后key被删除了，然后其他的请求创建了一个新的key，然后原来的执行删除了这个key，又出现了删除别人key的情况
///脚本执行（lua）
String srcipts = "if redis.call('get',KEYS[1]) == ARGV[1]  then return redis.call('del',KEYS[1]) else  return 0 end ";
// 通过Redis的lua脚本实现 查询和删除操作的原子性
stringRedisTemplate.execute(new DefaultRedisScript(srcipts,Integer.class)
,Arrays.asList("lock"),uuid);

Redisson分布式锁

https://github.com/redisson/redisson/wiki/8.-%E5%88%86%E5%B8%83%E5%BC%8F%E9%94%81%E5%92%8C%E5%90%8C%E6%AD%A5%E5%99%A8#81-%E5%8F%AF%E9%87%8D%E5%85%A5%E9%94%81reentrant-lock

原理

加锁原理

"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]);"

一个客户端要加锁，它首先会根据hash节点选择一台机器，这里注意仅仅只是选择一台机器，紧接着就会发送一段封装好的 lua 脚本到 redis 上，比如加锁的那个锁key就是"mylock"，并且设置的时间是30秒，30秒后mylock锁就会被自动释放。

锁互斥机制

第一个线程加锁完成后，此时，如果客户端 2 来尝试加锁会如何呢？首先，第一个 if 判断会执行 exists myLock，发现 myLock 这个锁 key 已经存在了。接着第二个 if 判断，判断一下，myLock 锁 key 的 hash 数据结构中，是否包含客户端 2 的 ID，这里明显不包含客户端 2 的ID，因为那里包含的是客户端 1 的 ID。所以，客户端 2 会执行：

return redis.call('pttl', KEYS[1]);
//返回的是一个数字，这个数字就是 myLock 这个锁 key 的剩余生存时间

锁阻塞机制

当锁正在被其他线程占用的时候，等待获取锁的线程并不是通过一个 while(true) 死循环去获取锁，而是利用了 Redis 的发布订阅机制，通过 await 方法阻塞等待锁的进程，有效的解决不断的申请锁浪费资源的问题。

锁续期机制

【1】客户端 1 加锁的默认生存(等待)时间为30秒，如果超过了30秒，客户端 1 还想一直持有这把锁，怎么办？
【2】Redisson 提供了一个续期机制，只要客户端 1 一旦加锁成功，就会启动一个 Watch Dog (看门狗)。
【3】Watch Dog 它是一个后台定时任务线程，会每隔10秒（internalLockLeaseTime / 3）检查一下(遍历 EXPIRATION_RENEWAL_MAP 里面线程 id 然后根据线程 id 去 Redis 中查，如果存在就会延长 key 的时间)，如果客户端 1 还持有锁key，那么就会不断的延长锁key的生存时间，直至业务执行结束或者该线程主动释放锁。

源码

private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
    long threadId = Thread.currentThread().getId();
    //尝试获取锁
    Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
    // lock acquired
    //获得锁直接返回
    if (ttl == null) {
        return;
    }
    //订阅与锁关联的Redis频道
    RFuture future = subscribe(threadId);
    if (interruptibly) {
        commandExecutor.syncSubscriptionInterrupted(future);
    } else {
        commandExecutor.syncSubscription(future);
    }
    try {
        while (true) {
            ttl = tryAcquire(-1, leaseTime, unit, threadId);
            // lock acquired
            if (ttl == null) {
                break;
            }
            // waiting for message
            //如果ttl >= 0，表示等待获取锁的时间（在给定的TTL时间内）。在这个时间段内，线程将尝试获取锁。如果在此期间发生中断，并且interruptibly为真，则抛出中断异常；否则，再次尝试获取锁。
            if (ttl >= 0) {
                try {
                    future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } catch (InterruptedException e) {
                    if (interruptibly) {
                        throw e;
                    }
                    future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                }
            } else {
                if (interruptibly) {
                    future.getNow().getLatch().acquire();
                } else {
                    future.getNow().getLatch().acquireUninterruptibly();
                }
            }
        }
    } finally {
        //清理订阅的资源
        unsubscribe(future, threadId);
    }
    //        get(lockAsync(leaseTime, unit));
}

private Long tryAcquire(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
    return get(tryAcquireAsync(waitTime, leaseTime, unit, threadId));
}

private  RFuture tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
    RFuture ttlRemainingFuture;
    if (leaseTime != -1) {
        ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
    } else {
        //过期时间为看门狗的默认时间，30秒
        ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime,
                TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
    }
    //开启一个定时任务进行不断刷新过期时间
    ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
        if (e != null) {
            return;
        }
        // lock acquired
        if (ttlRemaining == null) {
            if (leaseTime != -1) {
                internalLockLeaseTime = unit.toMillis(leaseTime);
            } else {
                scheduleExpirationRenewal(threadId);
            }
        }
    });
    return ttlRemainingFuture;
}

this.internalLockLeaseTime = commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout();
private long lockWatchdogTimeout = 30 * 1000;

 RFuture tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand command) {
    internalLockLeaseTime = unit.toMillis(leaseTime);
    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; " +
              // hexists查看哈希表的指定字段是否存在，存在锁并且是当前线程持有锁
              "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                      // hincrby自增一
                  "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                      // 锁的值大于1，说明是可重入锁，重置过期时间
                  "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                  "return nil; " +
              "end; " +
              // 锁已存在，且不是本线程，则返回过期时间ttl
              "return redis.call('pttl', KEYS[1]);",
                Collections.