首页 / 缓存 / 分布式缓存的穿透、雪崩、击穿以及分布式锁

分布式缓存的穿透、雪崩、击穿以及分布式锁

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了分布式缓存的穿透、雪崩、击穿以及分布式锁，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含12973字，纯文字阅读大概需要19分钟。

内容图文

1.缓存穿透

什么是缓存穿透,咱们来咬文嚼字理解一下，穿透穿透，穿过了还透过去了。缓存穿透主要指的是服务器在某一时间内接收到了大量数据库根本不存在的记录的查询请求(可能是恶意攻击)，这些请求没有命中缓存，直接大量压向数据库最终导致服务不可用。
我们传统的查询缓存，大都是先看缓存中有没有，有的话直接从缓存取。如果没有则去数据库查询，如果查询到了则保存在缓存里，返回结果。如果没有的话则直接返回null。
这就导致了对于根本不存在的请求，我们每次都会查数据库，这其实就是上述问题出现的原因。对此有两种解决办法。
解决办法

  1. 针对于查询了根本不存在的数据出现的问题这个角度考虑，所以我们可以采用布隆过滤器，对于无效请求进行过滤。
  2. 针对于查过数据库没有的数据下次任然去数据库查询这个角度考虑，我们查询数据库之后，如果发现没有数据，不要直接给客户端返回null，我们可以在缓存中存放一个自定义的Null值，也可能是某种标志位，代表这个数据在数据库没有。但是这个缓存不能一直存在，应该设置例如30s这样的过期时间，等缓存过期后再去数据库看看有没有。

2.缓存雪崩

什么是缓存雪崩，咱们继续来咬文嚼字哈哈，雪崩雪崩、不就是很多缓存突然掉下去了吗？这就涉及到一个问题，因为我们要尽量保证缓存和数据库数据的一致性，缓存不可能永久存在，缓存应该是有有效时间的，过了有效时间就应该再去数据库查询。
缓存雪崩就是在某一时间段，缓存中的key大面积同时过期，导致大量请求直接压入到数据库，导致数据库被压垮。
我们来思考一下这个问题出现的原因，缓存大面积同时过期，那我们让他别同时过期不就行了
解决办法

  1. 在每个缓存过期时间的原有值上加上一个1-5min的随机值，这样就能有效避免缓存大面积同时过期。
  2. 补充一个楼主的个人观点：其实随着缓存的刷新，请求去查数据库是必然的，所以大面积过期的情况是一定会存在的，那么如何尽量维护我们服务的稳定可用呢？我们数据库是咋垮的？压力过大呗，我们在设计的时候，应该将热点数据尽量分散到各个数据库，平摊压力。这个不单单是解决这个问题的，而是我们本就应该如此设计。

3.缓存击穿

缓存击穿很容易和缓存穿透混淆，缓存击穿就像是单点打击，针对于某个热点key，他在某一时刻过期了，那么大量的请求就直接压入到数据库了。而缓存穿透是针对于根本不存在的数据的查询请求直接压入到数据库。
解决办法

  1. 简单暴力，“热点数据缓存过期了”,那某些不经常改的数据我们让他不过期不就完了哈哈哈，需要跟新的时候再手动更新一下。
  2. 主备缓存，主缓存过期了，先去备用缓存中查询，然后主缓存趁机查询数据库。
  3. 主流方案：既然是大量的请求要查数据库，我们排队加锁，一个一个查，但是第一个查到之后直接放入缓存中，后面拿到锁之后先看看缓存中此时还有没有，所以实际上只有一个请求会真正去查询数据库。

其实上述三个问题，都可以使用分布式锁解决，但是锁毕竟是一种效率较低的解决方法，我们对于不同问题应该去具体分析而不是套用万能公式，具体问题具体分析才是解决问题的真正方法，抓住引起问题的原因，从源头上解决才是根本

4.分布式锁

针对于缓存击穿的问题，我们需要使用分布式锁来解决，使得真正意义上只有一个请求会去查数据库。
在说分布式锁之前我们应该考虑一下，能不能使用本地锁？
这个问题在这个情况下，实际上是可以的，尽管我们的本地锁只能锁住本进程，但是针对于我们这个问题，假设我们有100台服务器，每个都锁自己，最多也有100个请求去查数据库，所以理论上是可以的。但是随着服务规模的扩大，一个服务有100个请求，那成百上千个呢？尽管可以，但是本地锁并不是明智的解决方案。

4.1分布式锁的实现方式

楼主比较熟悉的分布式锁的实现有两种，一个是zookeeper实现，一个是redis实现，后面会详细介绍redis实现分布式锁，咱们前面先简单提一下zookeeper怎么实现。
zookeeper实现，设置zookeeper的某个路径为锁路径，需要获取锁时，在锁路径下创建一个临时有序节点，然后判断当前节点是不是所有存在的节点的第一个节点，如果是的话则获取到锁，如果不是，则watcher前一个锁的删除事件，当前线程通watcher调用wait方法进行等待，当前一个锁的删除节点事件触发后，回调函数中调用notify方法，当前线程恢复执行，重新调用获取锁。
下面是实例代码

public class MyLock {
    // zk的连接串
    String IP = "192.168.60.130:2181";
    // 计数器对象
    CountDownLatch countDownLatch = new CountDownLatch(1);
    //ZooKeeper配置信息
    ZooKeeper zooKeeper;
    private static final String LOCK_ROOT_PATH = "/Locks";
    private static final String LOCK_NODE_NAME = "Lock_";
    private String lockPath;

    // 打开zookeeper连接
    public MyLock() {
        try {
            zooKeeper = new ZooKeeper(IP, 5000, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    if (event.getType() == Event.EventType.None) {
                        if (event.getState() ==
                                Event.KeeperState.SyncConnected) {
                            System.out.println("连接成功!");
                            countDownLatch.countDown();
                        }
                    }
                }
            });
            countDownLatch.await();
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }

    //获取锁
    public void acquireLock() throws Exception {
//创建锁节点
    createLock();

    //尝试获取锁
    attemptLock();

}

    //创建锁节点
    private void createLock() throws Exception {
//判断Locks是否存在，不存在创建
        Stat stat = zooKeeper.exists(LOCK_ROOT_PATH, false);
        if (stat == null) {
            zooKeeper.create(LOCK_ROOT_PATH, new byte[0],
                    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
// 创建临时有序节点
        lockPath = zooKeeper.create(LOCK_ROOT_PATH + "/" +
                        LOCK_NODE_NAME, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println("节点创建成功:" + lockPath);
    }

    //监视器对象，监视上一个节点是否被删除
    Watcher watcher = new Watcher() {
        @Override
        public void process(WatchedEvent event) {
            if (event.getType() == Event.EventType.NodeDeleted) {
                synchronized (this) {
                    notifyAll();
                }
            }
        }
    };

    //尝试获取锁
    private void attemptLock() throws Exception {
// 获取Locks节点下的所有子节点
        List<String> list = zooKeeper.getChildren(LOCK_ROOT_PATH, false);
// 对子节点进行排序
        Collections.sort(list);
// /Locks/Lock_000000001
    int index = list.indexOf(lockPath.substring(LOCK_ROOT_PATH.length() + 1));
        if (index == 0) {
            System.out.println("获取锁成功!");
            return;
        } else {
// 上一个节点的路径
            String path = list.get(index ‐ 1);
            Stat stat = zooKeeper.exists(LOCK_ROOT_PATH + "/" + path,
                    watcher);
            if (stat == null) {
                attemptLock();
            } else {
                synchronized (watcher) {
                    watcher.wait();
                }
                attemptLock();
            }
        }
    }

    //释放锁
    public void releaseLock() throws Exception {
//删除临时有序节点
        zooKeeper.delete(this.lockPath,‐1);
        zooKeeper.close();
        System.out.println("锁已经释放:" + this.lockPath);
    }

    public static void main(String[] args) {
        try {
            MyLock myLock = new MyLock();
            myLock.createLock();
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }
}
public class TicketSeller {
    private void sell() {
        System.out.println("售票开始");
// 线程随机休眠数毫秒，模拟现实中的费时操作
        int sleepMillis = 5000;
        try {
//代表复杂逻辑执行了一段时间
            Thread.sleep(sleepMillis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("售票结束");
    }

    public void sellTicketWithLock() throws Exception {
        MyLock lock = new MyLock();
// 获取锁
        lock.acquireLock();
        sell();
//释放锁
        lock.releaseLock();
    }

    public static void main(String[] args) throws Exception {
        TicketSeller ticketSeller = new TicketSeller();
        for (int i = 0; i < 10; i++) {
            ticketSeller.sellTicketWithLock();
        }
    }
}

Redis实现分布式锁

本地锁的之所以在分布式下失效就是因为锁有很多把，所以要实现分布式锁，使用一把锁就行了。redis实现分布式锁，类似于平时我们去公共卫生间上厕所，别人占着了我们就得等着，只有别人上完了我们才能继续上。
使用redis实现分布式锁有几个关键点

  1. 保证加锁和设置过期时间两个操作的原子性：set nx ex同时执行
  2. 保证判断锁内容和删除所两个操作的原子性：使用lua脚本
  3. 如何实现缓存过期时间的续期：使用一个新的线程在原线程正常工作的情况下不断续期

前面几个点我们先摆出来，我们来慢慢理一下如何使用redis实现分布式锁
redis设置分布式锁的思想是在redis中保存一个key，如果这个key已经存在了，那么说明有人抢到了这把锁，当抢到锁的人使用完了锁之后需要将锁删除。
针对于缓存击穿问题我们简单写一下下面的代码步骤

	Data = getDataFromCache();
	if(Data == null){
		while(!getLock());//如果取锁失败，进行自旋
		data = getDataFromCache();//再查询一遍缓存是否有数据
		if(data == null){
			data = getDataFromDB();
			putDataToCache();
			releaseLock();
			retrun data;
		}
	} 
	return Data;

下面是楼主最近在学习的谷粒商城项目的代码实例。代码的目的是获取一个catalogJson数据。

    @Override
    public Map<String, List<Catalog2Vo>> getCatalogJsonMap() {
        ValueOperations<String, String> ops = redisTemplate.opsForValue();
        //先从缓存中获取数据
        String catalogJson = ops.get("catalogJson");
        if (StringUtils.isEmpty(catalogJson)){
            //如果缓存中没有数据，准备去数据库查询了、先获取锁
            try {
                while (!ops.setIfAbsent("lock", UUID.randomUUID().toString()));
                //拿到锁之后，先查一遍缓存，可能在等待锁的过程中，其他人已经查好放缓存里了
                catalogJson = ops.get("catalogJson");
                if (StringUtils.isEmpty(catalogJson)){
                    //缓存还是没有。去查数据库
                    Map<String, List<Catalog2Vo>> catalogJsonMap = getCatalogJsonMapFromDB();
                    ops.set("catalogJson",JSON.toJSONString(catalogJsonMap));
                    return catalogJsonMap;
                }
            } finally {
                redisTemplate.delete("lock");
            }
        }
        Map<String, List<Catalog2Vo>> jsonMap = JSON.parseObject(catalogJson, new TypeReference<Map<String, List<Catalog2Vo>>>() {
        });
        return jsonMap;
    }

但是上面的代码有一个很大的问题，假设我们一个服务设置锁了，但是他挂了，没有删除锁，那岂不是死锁了吗？所以我们的锁不能一直存在需要有过期时间。

  @Override
    public Map<String, List<Catalog2Vo>> getCatalogJsonMap() {
        ValueOperations<String, String> ops = redisTemplate.opsForValue();
        //先从缓存中获取数据
        String catalogJson = ops.get("catalogJson");
        if (StringUtils.isEmpty(catalogJson)){
            //如果缓存中没有数据，准备去数据库查询了、先获取锁
            try {

                while (!ops.setIfAbsent("lock", UUID.randomUUID().toString()));
                //设置30s的过期时间
                redisTemplate.expire("lock",30,TimeUnit.SECONDS);
                //拿到锁之后，先查一遍缓存，可能在等待锁的过程中，其他人已经查好放缓存里了
                catalogJson = ops.get("catalogJson");
                if (StringUtils.isEmpty(catalogJson)){
                    //缓存还是没有。去查数据库
                    Map<String, List<Catalog2Vo>> catalogJsonMap = getCatalogJsonMapFromDB();
                    ops.set("catalogJson",JSON.toJSONString(catalogJsonMap));
                    return catalogJsonMap;
                }
            } finally {
                redisTemplate.delete("lock");
            }

        }
        Map<String, List<Catalog2Vo>> jsonMap = JSON.parseObject(catalogJson, new TypeReference<Map<String, List<Catalog2Vo>>>() {
        });
        return jsonMap;
    }

好了设置了30s过期时间，这样我们貌似没什么问题了，但是这还存在一个问题。我们是抢到锁之后，在设置的过期时间，我刚抢到锁就挂了，那还不是死锁了，所以这就是我们开头的三个问题的第一个：保证加锁与设置过期时间的原子性

while (!ops.setIfAbsent("lock", UUID.randomUUID().toString(),30,TimeUnit.SECONDS));

好了，我们过期时间也设置了，这会总没有什么问题了吧？
不，你想想我们设置的过期时间是30s，假设我们业务总共执行了40s，因为所有的锁都叫lock，那么我们执行到40s的时候，删除的肯定不是自己的锁，自己啪一下把别人的锁给删了，所以我们还需要判断我们删除的时候锁是不是自己的那把。如何确定呢，就是自己set的value是专属于自己的，这里使用时间戳+uuid的方式。

    @Override
    public Map<String, List<Catalog2Vo>> getCatalogJsonMap() {
        ValueOperations<String, String> ops = redisTemplate.opsForValue();
        //先从缓存中获取数据
        String catalogJson = ops.get("catalogJson");
        if (StringUtils.isEmpty(catalogJson)){
            //如果缓存中没有数据，准备去数据库查询了、先获取锁
            String val = null;
            try {
                //设置30s的过期时间,加锁和过期时间一起设置
                val = System.currentTimeMillis() + UUID.randomUUID().toString();
                while (!ops.setIfAbsent("lock", val,30,TimeUnit.SECONDS));

                //拿到锁之后，先查一遍缓存，可能在等待锁的过程中，其他人已经查好放缓存里了
                catalogJson = ops.get("catalogJson");
                if (StringUtils.isEmpty(catalogJson)){
                    //缓存还是没有。去查数据库
                    Map<String, List<Catalog2Vo>> catalogJsonMap = getCatalogJsonMapFromDB();
                    ops.set("catalogJson",JSON.toJSONString(catalogJsonMap));
                    return catalogJsonMap;
                }
            } finally {
                String cacheVal = ops.get("lock");
                if (val.equals(cacheVal)){
                    redisTemplate.delete("lock");
                }
            }
        }
        Map<String, List<Catalog2Vo>> jsonMap = JSON.parseObject(catalogJson, new TypeReference<Map<String, List<Catalog2Vo>>>() {
        });
        return jsonMap;
    }

好，这样总没问题了吧？不，如果你获取val数据的时候，你的缓存还剩下最后0.1s，然后redis给你返回了你得数据，0.1s之后别的服务重新抢占了锁。你收到数据后，发现，噢是自己的锁，啪又一删，你有把人家的锁删了，所以这就引出了我们开头的第二个问题
：保证判断锁内容和删除所两个操作的原子性
这里使用lua脚本来进行锁的删除，lua脚本保证了我们获取值和删除是原子性的操作。

   @Override
    public Map<String, List<Catalog2Vo>> getCatalogJsonMap() {
        ValueOperations<String, String> ops = redisTemplate.opsForValue();
        //先从缓存中获取数据
        String catalogJson = ops.get("catalogJson");
        if (StringUtils.isEmpty(catalogJson)){
            //如果缓存中没有数据，准备去数据库查询了、先获取锁
            String val = null;
            try {
                //设置30s的过期时间,加锁和过期时间一起设置
                val = System.currentTimeMillis() + UUID.randomUUID().toString();
                while (!ops.setIfAbsent("lock", val,30,TimeUnit.SECONDS));

                //拿到锁之后，先查一遍缓存，可能在等待锁的过程中，其他人已经查好放缓存里了
                catalogJson = ops.get("catalogJson");
                if (StringUtils.isEmpty(catalogJson)){
                    //缓存还是没有。去查数据库
                    Map<String, List<Catalog2Vo>> catalogJsonMap = getCatalogJsonMapFromDB();
                    ops.set("catalogJson",JSON.toJSONString(catalogJsonMap));
                    return catalogJsonMap;
                }
            } finally {
                //删除锁
                RedisScript script = RedisScript.of("if redis.call(\"get\",KEYS[1]) == ARGV[1] then\n" +
                        "    return redis.call(\"del\",KEYS[1])\n" +
                        "else\n" +
                        "    return 0\n" +
                        "end", Long.class);
                redisTemplate.execute(script, Arrays.asList("lock"), val);

            }
        }
        Map<String, List<Catalog2Vo>> jsonMap = JSON.parseObject(catalogJson, new TypeReference<Map<String, List<Catalog2Vo>>>() {
        });
        return jsonMap;
    }

然后最后就剩下我们第三个问题啦，我们设置的过期时间肯定是固定的，但是我们的业务执行时间确实不确定的，那怎么办呢？如何给我们的缓存过期时间进行续期呢？
这里可以开一个线程去给我们自动续期，当我们业务线程还在工作时，另一个线程自动给我们的缓存时间进行续期。这样就完成了。
上面就是博主分享的全部内容了，楼主毕竟才疏学浅，如有错误望不吝赐教。
redis分布式锁参考官网： redis分布式锁.

内容总结

以上是互联网集市为您收集整理的分布式缓存的穿透、雪崩、击穿以及分布式锁全部内容，希望文章能够帮你解决分布式缓存的穿透、雪崩、击穿以及分布式锁所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

扫描二维码推送至手机访问。

本文链接：https://qyyshop.com/info/1017363.html

来源：【匿名】