在这篇文章中，我首要介绍的是分布式缓存和本地缓存的运用技巧，包含缓存品种介绍，各种的运用场景，以及怎么运用，最后再给出实战事例。

众所周知，缓存最首要的目的便是加快拜访，缓解数据库压力。最常用的缓存便是分布式缓存，比方redis，在面临大部分并发场景或者一些中小型公司流量没有那么高的状况，运用redis根本都能解决了。可是在流量较高的状况下或许得运用到本地缓存了，比方guava的LoadingCache和快手开源的ReloadableCache。

三种缓存的运用场景

这部分会介绍redis，本地缓存比方guava的LoadingCache和快手开源的ReloadableCache的运用场景和限制，经过这一部分的介绍就能知道在怎样的事务场景下应该运用哪种缓存，以及为什么。

Redis的运用场景和限制性

假如宽泛的说redis何时运用，那么天然便是用户拜访量过高的当地运用，然后加快拜访，而且缓解数据库压力。假如细分的话，还得分为单节点问题和非单节点问题。

假如一个页面用户拜访量比较高，可是拜访的不是同一个资源。比方用户详情页，拜访量比较高，可是每个用户的数据都是不一样的，这种状况显然只能用分布式缓存了，假如运用redis，key为用户唯一键，value则是用户信息。

redis导致的缓存击穿。

可是需求留意一点，必定要设置过期时刻，而且不能设置到同一时刻点过期。举个比方，比方用户又个活动页，活动页能看到用户活动期间获奖数据，大意的人或许会设置用户数据的过期时刻点为活动结束，这样会导致同一时刻过期，必然导致了击穿问题。

单（热）点问题

单节点问题说的是redis的单个节点的并发问题，由于关于相同的key会落到redis集群的同一个节点上，那么假如对这个key的拜访量过高，那么这个redis节点就存在并发危险,这个key就称为热key。

假如一切用户拜访的都是同一个资源，比方小爱同学app主页对一切用户展现的内容都一样（初期），服务端给h5返回的是同一个大json，显然得运用到缓存。首要咱们考虑下用redis是否可行，由于redis存在单点问题，假如流量过大的话，那么一切用户的恳求到达redis的同一个节点，需求评价该节点能否抗住这么大流量。咱们的规则是，假如单节点qps达到了千等级就要解决单点问题了（即便redis声称能抗住十万等级的qps），最常见的做法便是运用本地缓存。显然小爱app主页流量不过百，运用redis是没问题的。

LoadingCache的运用场景和限制性

关于这上面说的热key问题，咱们最直接的做法便是运用本地缓存，比方你最熟悉的guava的LoadingCache，可是运用本地缓存要求能够承受必定的脏数据，由于假如你更新了主页，本地缓存是不会更新的，它只会根据必定的过期战略来从头加载缓存，不过在咱们这个场景是完全没问题的，由于一旦在后台推送了主页后就不会再去改变了。即便改变了也没问题，能够设置写过期为半小时，超越半小时从头加载缓存，这种短时刻内的脏数据咱们是能够承受的。

LoadingCache导致的缓存击穿

尽管说本地缓存和机器上强相关的，尽管代码层面写的是半小时过期，但由于每台机器的发动时刻不同，导致缓存的加载时刻不同，过期时刻也就不同，也就不会一切机器上的恳求在同一时刻缓存失效后都去恳求数据库。可是关于单一一台机器也是会导致缓存穿透的，假如有10台机器，每台1000的qps，只要有一台缓存过期就或许导致这1000个恳求一起打到了数据库。这种问题其实比较好解决，可是容易被忽略，也便是在设置LoadingCache的时分运用LoadingCache的load-miss办法，而不是直接判断cache.getIfPresent()== null然后去恳求db；前者会加虚拟机层面的锁，确保只要一个恳求打到数据库去，然后完美的解决了这个问题。

可是，假如关于实时性要求较高的状况，比方有段时刻要经常做活动，我要确保活动页面能近实时更新，也便是运营在后台装备好了活动信息后，需求在C端近实时展现这次装备的活动信息，此刻运用LoadingCache必定就不能满足了。

ReloadableCache的运用场景和限制性

关于上面说的LoadingCache不能解决的实时问题，能够考虑运用ReloadableCache，这是快手开源的一个本地缓存结构，最大的特点是支撑多机器一起更新缓存，假设咱们修正了主页信息，然后恳求打到的是A机器，这个时分从头加载ReloadableCache，然后它会发出告诉，监听了同一zk节点的其他机器收到告诉后从头更新缓存。运用这个缓存一般的要求是将全量数据加载到本地缓存，所以假如数据量过大必定会对gc形成压力，这种状况就不能运用了。由于小爱同学主页这个主页是带有状况的，一般online状况的就那么两个，所以完全能够运用ReloadableCache来只装载online状况的主页。

小结

到这儿三种缓存根本都介绍完了，做个小结：

  1. 关于非热门的数据拜访，比方用户维度的数据，直接运用redis即可；
  2. 关于热门数据的拜访，假如流量不是很高，无脑运用redis即可；
  3. 关于热门数据，假如答应必定时刻内的脏数据，运用LoadingCache即可；
  4. 关于热门数据，假如一致性要求较高，一起数据量不大的状况，运用ReloadableCache即可；

小技巧

不管哪种本地缓存尽管都带有虚拟机层面的加锁来解决击穿问题，可是意外总有或许以你意想不到的办法产生，稳妥起见你能够运用两级缓存的办法即本地缓存+redis+db。

缓存运用的简略介绍

这儿redis的运用就不再多说了，相信很多人对api的运用比我还熟悉

LoadingCache的运用

这个是guava供给的网上一抓一大把，可是给两点留意事项

  1. 要运用load-miss的话, 要么运用V get(K key, Callable<? extends V> loader)；要么运用build的时分运用的是build(CacheLoader<? super K1, V1> loader)这个时分能够直接运用get()了。此外建议运用load-miss，而不是getIfPresent==null的时分再去查数据库，这或许导致缓存击穿；
  2. 运用load-miss是由于这是线程安全的，假如缓存失效的话，多个线程调用get的时分只会有一个线程去db查询，其他线程需求等候，也便是说这是线程安全的。
LoadingCache<String, String> cache = CacheBuilder.newBuilder()
                .maximumSize(1000L)
                .expireAfterAccess(Duration.ofHours(1L)) // 多久不拜访就过期
                .expireAfterWrite(Duration.ofHours(1L))  // 多久这个key没修正就过期
                .build(new CacheLoader<String, String>() {
                    @Override
                    public String load(String key) throws Exception {
                        // 数据装载办法，一般便是loadDB
                        return key + " world";
                    }
                });
String value = cache.get("hello"); // 返回hello world

reloadableCache的运用

导入三方依靠

<dependency>
  <groupId>com.github.phantomthief</groupId>
  <artifactId>zknotify-cache</artifactId>
  <version>0.1.22</version>
</dependency>

需求看文档，不然无法运用，有爱好自己写一个也行的。

public interface ReloadableCache<T> extends Supplier<T> {
    /**
     * 获取缓存数据
     */
    @Override
    T get();
    /**
     * 告诉全局缓存更新
     * 留意：假如本地缓存没有初始化，本办法并不会初始化本地缓存并从头加载
     *
     * 假如需求初始化本地缓存，请先调用 {@link ReloadableCache#get()}
     */
    void reload();
    /**
     * 更新本地缓存的本地副本
     * 留意：假如本地缓存没有初始化，本办法并不会初始化并刷新本地的缓存
     *
     * 假如需求初始化本地缓存，请先调用 {@link ReloadableCache#get()}
     */
    void reloadLocal();
}

目前有个子类ZkNotifyReloadCache，内部类Builder，供给了withCacheFactory办法，经过该办法来加载数据到缓存；enableAutoReload办法供给缓存自动更新时刻。

public class ZkNotifyReloadCache<T> implements ReloadableCache<T> {
	public static final class Builder<T> {
        public Builder<T> withNotifyZkPath(String notifyZkPath) {
            if (notifyZkPaths == null) {
                notifyZkPaths = new HashSet<>();
            }
            this.notifyZkPaths.add(notifyZkPath);
            return this;
        }
        public Builder<T> withCacheFactoryEx(CacheFactoryEx<T> cacheFactoryEx) {
            this.cacheFactory = cacheFactoryEx;
            return this;
        }
        public Builder<T> enableAutoReload(long timeDuration, TimeUnit unit) {
            return enableAutoReload(() -> ofMillis(unit.toMillis(timeDuration)));
        }
    }
}

一般T咱们会设置为一个Map<K,V>, key为缓存的数据id，value为具体数据。原理简略介绍如下：

  1. 经过withNotifyZkPath指定监听的zkPath, 集群环境下多机器的缓存同步便是经过这个zkPath完成的，比方a机器收到了更新缓存的恳求，那么需求从头加载数据，此刻会向该路径下写入数据，其他机器的zk客户端监听到了事情变更，则也会从头加载数据;
  2. enableAutoReload则表示每隔指定的时刻就从头加载一遍数据；
  3. withCacheFactoryEx则是最核心的办法，指定数据来源，也便是咱们查db的操作了；
  4. build办法会构建ReloadableCache，其中经过ensure办法来构建broadcast（简略了解便是zk实例）;

当恳求获取数据的时分，咱们调用get办法来获取map，也便是全量的缓存数据，然后再根据key来获取咱们想要的数据.

当数据更新的时分咱们根据需求来调用reload办法来更新全局缓存。

陈词滥调的缓存击穿/穿透/雪崩问题

这三个真的是亘古不变的问题，假如流量大的确需求考虑。

缓存击穿

简略说便是缓存失效，导致很多恳求同一时刻打到了数据库。关于缓存击穿问题上面现已给出了很多解决方案了。

  1. 比方运用本地缓存
  2. 本地缓存运用load-miss办法
  3. 运用第三方服务来加载缓存

1.2和都说过，首要来看3。假如事务乐意只能运用redis而无法运用本地缓存，比方数据量过大，实时性要求比较高。那么当缓存失效的时分就得想办法确保只要少量的恳求打到数据库。很天然的就想到了运用分布式锁，理论上说是可行的，但实际上存在危险。咱们的分布式锁相信很多人都是运用redis+lua的办法完成的，而且在while中进行了轮训，这样恳求量大，数据多的话会导致无形中让redis成了危险，而且占了太多事务线程，其实仅仅是引入了分布式锁就加大了复杂度，咱们的原则便是能不必就不必。

那么咱们是不是能够设计一个类似分布式锁，可是更牢靠的rpc服务呢？当调用get办法的时分这个rpc服务确保相同的key打到同一个节点，而且运用synchronized来进行加锁，之后完成数据的加载。在快手供给了一个叫cacheSetter的结构。下面供给一个简易版，自己写也很容易完成。

import com.google.common.collect.Lists;
import org.apache.commons.collections4.CollectionUtils;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
/**
 * @Description 分布式加载缓存的rpc服务，假如部署了多台机器那么调用端最好运用id做一致性hash确保相同id的恳求打到同一台机器。
 **/
public abstract class AbstractCacheSetterService implements CacheSetterService {
    private final ConcurrentMap<String, CountDownLatch> loadCache = new ConcurrentHashMap<>();
    private final Object lock = new Object();
    @Override
    public void load(Collection<String> needLoadIds) {
        if (CollectionUtils.isEmpty(needLoadIds)) {
            return;
        }
        CountDownLatch latch;
        Collection<CountDownLatch> loadingLatchList;
        synchronized (lock) {
            loadingLatchList = excludeLoadingIds(needLoadIds);
            needLoadIds = Collections.unmodifiableCollection(needLoadIds);
            latch = saveLatch(needLoadIds);
        }
        System.out.println("needLoadIds:" + needLoadIds);
        try {
            if (CollectionUtils.isNotEmpty(needLoadIds)) {
                loadCache(needLoadIds);
            }
        } finally {
            release(needLoadIds, latch);
            block(loadingLatchList);
        }
    }
    /**
     * 加锁
     * @param loadingLatchList 需求加锁的id对应的CountDownLatch
     */
    protected void block(Collection<CountDownLatch> loadingLatchList) {
        if (CollectionUtils.isEmpty(loadingLatchList)) {
            return;
        }
        System.out.println("block:" + loadingLatchList);
        loadingLatchList.forEach(l -> {
            try {
                l.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }
    /**
     * 开释锁
     * @param needLoadIds 需求开释锁的id调集
     * @param latch 经过该CountDownLatch来开释锁
     */
    private void release(Collection<String> needLoadIds, CountDownLatch latch) {
        if (CollectionUtils.isEmpty(needLoadIds)) {
            return;
        }
        synchronized (lock) {
            needLoadIds.forEach(id -> loadCache.remove(id));
        }
        if (latch != null) {
            latch.countDown();
        }
    }
    /**
     * 加载缓存，比方根据id从db查询数据，然后设置到redis中
     * @param needLoadIds 加载缓存的id调集
     */
    protected abstract void loadCache(Collection<String> needLoadIds);
    /**
     * 对需求加载缓存的id绑定CountDownLatch，后续相同的id恳求来了从map中找到CountDownLatch，而且await，直到该线程加载完了缓存
     * @param needLoadIds 能够正在去加载缓存的id调集
     * @return 共用的CountDownLatch
     */
    protected CountDownLatch saveLatch(Collection<String> needLoadIds) {
        if (CollectionUtils.isEmpty(needLoadIds)) {
            return null;
        }
        CountDownLatch latch = new CountDownLatch(1);
        needLoadIds.forEach(loadId -> loadCache.put(loadId, latch));
        System.out.println("loadCache:" + loadCache);
        return latch;
    }
    /**
     * 哪些id正在加载数据，此刻持有相同id的线程需求等候
     * @param ids 需求加载缓存的id调集
     * @return 正在加载的id所对应的CountDownLatch调集
     */
    private Collection<CountDownLatch> excludeLoadingIds(Collection<String> ids) {
        List<CountDownLatch> loadingLatchList = Lists.newArrayList();
        Iterator<String> iterator = ids.iterator();
        while (iterator.hasNext()) {
            String id = iterator.next();
            CountDownLatch latch = loadCache.get(id);
            if (latch != null) {
                loadingLatchList.add(latch);
                iterator.remove();
            }
        }
        System.out.println("loadingLatchList:" + loadingLatchList);
        return loadingLatchList;
    }
}

事务完成

import java.util.Collection;
public class BizCacheSetterRpcService extends AbstractCacheSetterService {
    @Override
    protected void loadCache(Collection<String> needLoadIds) {
        // 读取db进行处理
   	// 设置缓存
    }
}

缓存穿透

简略来说便是恳求的数据在数据库不存在，导致无效恳求打穿数据库。

解法也很简略，从db获取数据的办法(getByKey(K key))必定要给个默认值。

比方我有个奖池，金额上限是1W，用户完成使命的时分给他发笔钱，而且运用redis记载下来，而且落表，用户在使命页面能实时看到奖池剩余金额，在使命开端的时分显然奖池金额是不变的，redis和db里边都没有发放金额的记载，这就导致每次必然都去查db，关于这种状况，从db没查出来数据应该缓存个值0到缓存。

缓存雪崩

便是很多缓存会集失效打到了db，当然必定都是一类的事务缓存，归根到底是代码写的有问题。能够将缓存失效的过期时刻打散，别让其会集失效就能够了。