基于异步+缓存的请求日志批量保存方案，拿去写在你的简历上吧！

站长

2023年02月28日 20:38 · 阅读数 14

前言

为了便于分析和排查问题，我希望可以保存下每一条请求日志。那既然每一条都要保存，把这个功能添加到网关服务中我觉得是比较合适的了。

但是也正因为所有的请求都要经过这个被保存的过程，所以我希望这个过程要尽可能的简短不费时间，不要影响我的访问速度。

正巧我学过一点 Redis 的知识，简单了解过Redis持久化的思路，于是我参考了它的思路，给自己的网关服务写了一个使用 缓存 + 异步 的存储策略，当然我这里持久化就是保存到MySQL了。

本文所介绍的内容均来自我的开源项目校园博客中，开源地址：stick-i/scblogs: 🎉校园博客，基于微服务架构且前后端分离的博客社区系统。项目后端技术栈：SpringBoot + SpringCloud + Mybatis-Plus + Nacos + MySQL + Redis + MQ + ElasticSearch + Docker。前端主要是基于Vue2和ElementUI进行开发的。 (github.com)

基本思路

我的大致思路是这样的：

先把每一条请求信息存到缓存中，可以直接存 List，或者使用 Redis 。
每隔一段时间，创建一个子线程，去读取缓存中的数据，并将数据存入硬盘，这里我存MySQL。
如果没有新的访问记录，那就不用去定时执行了，所以最好可以主动调节，通过请求触发子进程的保存。
程序正常退出的情况下，最好能够再主动保存一次缓存中的数据，这里可以注册一个Hook来执行，防止定时任务没到执行时间。

由于这个功能我已经在项目中实现好了，并且已经使用一段时间了，所以下面我会直接就着已经写好的代码来跟大家分析讲解。

代码实现

首先介绍一下基本情况：

项目的网关使用的是Spring Cloud Gateway，保存请求记录的逻辑通过全局过滤器 GlobalFilter 来调用，也就是用一个单独的过滤器去保存访问记录。
用于存储到数据库的实体类为 VisitRecord ，内容包括ip地址、uri、请求方法、请求参数、状态码等信息。这些信息可以从ServerWebExchange的对象中获取，解析之后放到 VisitRecord的对象里就行了。

有上面这些条件后，下面我们就可以只关注 请求日志 的实现了，这部分的实现我放在了VisitRecordService类里面了，源码所在位置：scblogs/VisitRecordService.java at main · stick-i/scblogs (github.com)。

在下面的讲解中，我剔除了大部分业务相关的东西，但是我保留了一部分。

是故意的还是不小心的？

入口

通过调用下面的方法，可以将经过网关的访问记录进行储存。

/**
 * 保存访问记录
 *
 * @param exchange gateway访问合同
 */
public void add(ServerWebExchange exchange) {
    // 获取信息
    ServerHttpResponse response = exchange.getResponse();
    ServerHttpRequest request = exchange.getRequest();
    // 构建VisitRecord
    VisitRecord visitRecord = getOrBuild(exchange);

    // 打印访问情况
    log.info(visitRecord.toString());

    // 添加访问记录
    addRecord(visitRecord);
}

这段代码很简单，就是先拿到了要被存储的访问记录信息，然后再去调用了另一个方法addRecord()。

存入缓存

我们接着上面的addRecord()方法继续往下看：

private void addRecord(VisitRecord record) {
    // 添加记录到缓存中
    visitCache.add(record);
    // 执行任务，保存数据
    doTask();
}

这个方法也很简单，就是往缓存里添加了这条新的记录，然后调用了doTask()方法去执行存储的任务。

先看看这个visitCache是个什么东西？

/**
 * 缓存，在插入数据库前先存入此。
 * 为防止数据被重复插入，故使用Set，但不能确保100%不会被重复存储。
 */
private HashSet<VisitRecord> visitCache = new HashSet<>();

其实就是个HashSet，不过我这里用Set是有原因的：

在我的这个项目中有个Gateway专用的全局异常处理器GlobalExceptionHandler，如果发生异常的话，会被这个处理器捕获，并且会打断过滤器的执行。基于这个逻辑，可能会出现两种情况：

已经执行了保存的方法，在保存请求记录过滤器的后面抛出了异常，这样的话不需要再重新保存日志了。
还没执行过保存的方法，也就是在它前面抛出了异常，这样的话肯定是需要重新保存日志的。

于是综合这两种情况，我选择了使用Set，并且在异常处理器里加入了保存访问记录的逻辑（就是调用最上面那个入口方法），这样可以保证不会出现漏掉访问记录的情况，也可以尽量避免重复保存的情况，但不能完全保证不会被重复保存。

多讲了几局题外话，这个跟主题关系不大了，感兴趣的朋友可以去GitHub看我的项目源码继续了解：链接。

执行任务

数据已经存到缓存了，我们接着上面的 doTask(); 方法看：

private final ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNamePrefix("visit-record-").build();

private final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(1, 3, 15, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());

/**
 * 信号量，用于标记当前是否有任务正在执行，{@code true}表示当前无任务进行。
 */
private volatile boolean taskFinish = true;

/**
 * 单次批量插入的数据量
 */
private final int BATCH_SIZE = 500;

private void doTask() {
    if (taskFinish) {
        // 当前没有任务的情况下，加锁并执行任务
        synchronized (this) {
            if (taskFinish) {
                taskFinish = false;
                threadPool.execute(() -> {
                    try {
                        // 当数据量较小时，则等待一段时间再插入数据，从而做到将数据尽可能的批量插入数据库
                        if (visitCache.size() <= BATCH_SIZE) {
                            Thread.sleep(500);
                        }
                        batchSave();
                    } catch (InterruptedException e) {
                        log.error("休眠时发生了异常: {}", e.getMessage());
                    } finally {
                        // 任务执行完毕后修改标志位
                        taskFinish = true;
                    }
                });
            }
        }
    }
}

这部分就有点东西了：

首先是经典的双重检查锁定，使用taskFinish信号量，并且被volatile修饰，估计在面试资料的单例模式写法上见过吧🤡。这样可以保证第二个if里面的东西只会被单独执行，而不会并发执行。
有一个被final修饰的线程池，并且用了线程工厂，这样在打印日志的时候可以看到哪些日志是由这部分代码打印的噢。线程池的核心线程数是 1 ，这跟下面的任务提交有关，每次最多只会存在一个任务。
在第二个if里面，是这个方法的核心逻辑。首先把信号量改为了false，表示当前已经有任务在执行了，然后向线程池提交了一个任务，在任务中通过 finally 保证任务执行完毕后再恢复信号量。
考虑到数据量比较小的时候，可能会不停的创建任务，数据保存完之后马上又需要保存新的数据，而且可能每次都只保存了一条两条数据，这样有点违背了我们批量插入的初心，也浪费了性能。

所以我设定了一个常量BATCH_SIZE = 500，用来表示我希望单次批量插入的数据量。如果当前缓存里的数据量小于该数据量，那么让线程在此等待那么一会，再去执行真正的跟数据库交互的操作batchSave()。

这里需要考虑的是，如果我有另一个服务会去读取并展示这些请求日志，那我肯定希望请求日志是能够实时更新的，所以我选择sleep 0.5秒，而不是等到数据量达到500才存入。

这样既减轻了系统负担，又可以尽量做到即时更新，可谓一举两得。

存入数据库

经历了这么几个步骤，终于要存数据库了，也就是上文任务中的最后一个方法batchSave();，先来看看代码：

/**
 * 单次批量插入的数据量
 */
private final int BATCH_SIZE = 500;

/**
 * 缩减因子，每次更新缓存Set时缩小的倍数，对应HashSet的扩容倍数
 */
private final float REDUCE_FACTOR = 0.5f;

private void batchSave() {
    log.debug("访问记录准备插入数据库，当前数据量：{}", visitCache.size());
    if (visitCache.size() == 0) {
        return;
    }
    // 构造新对象来存储数据，旧对象保存到数据库后不再使用
    HashSet<VisitRecord> oldCache = visitCache;
    visitCache = new HashSet<>((int) (oldCache.size() * REDUCE_FACTOR));
    boolean isSave = false;
    try {
        // 存入数据库
        isSave = visitLogService.saveBatch(oldCache, BATCH_SIZE);
    } finally {
        if (!isSave) {
            // 如果插入失败，则重新添加所有数据
            visitCache.addAll(oldCache);
        }
    }
}

这段代码也是有亮点的，我们来分析一下：

首先检查了一下缓存中的数据量，这没什么好说的。
然后将缓存对象visitCache使用了一个新的变量oldCache来引用，然后new了一个新的HashSet对象，并且让visitCache去引用了这个新对象，再把oldCache批量插入数据库，这里的saveBatch是用的Mybatis-Plus的方法，就是批量插入到数据库里的。

这里是有说法的：

为什么我不直接保存visitCache到数据库，还要多创建一个新缓存对象，再去保存旧对象？

结合本文存入缓存的代码，我无法保证在把这些数据存入数据库的期间没有新的请求被存入缓存，也就是visitCache对象。那在visitLogService.saveBatch();执行完毕后，我就无法保证此时的visitCache全部被存到数据库了，那我到底还要不要调用visitCache.clear()方法呢？
创建新对象时我是这么写的visitCache = new HashSet<>((int) (oldCache.size() * REDUCE_FACTOR));，为什么我给HashSet的初始大小要使用 旧缓存的大小 * 0.5 呢？
- 首先，我不希望visitCache去慢慢扩容到合适的大小，这样浪费性能。
- 其次，我希望它不要有过多的冗余容量，如果我的初始化大小直接就是 oldCahce.size()，那它的容量永远都不会降下来了。
- 至于为什么是0.5，因为HashSet的底层其实就是个HashMap，而HashMap每次扩容都是上一次容量大小的两倍，HashMap初始化容量大小的值，也必须是2的次方。如果不是2的次方，则会自动帮你调整为向上取的第一个2的次方的数，比如我给的参数是10，那它的初始容量就是16咯。
  
  这里我乘0.5，其实也不过就是给它降了一次扩容的空间罢了，听懂掌声。
其实这里我也考虑过使用两个HashSet去做一个滚筒的设计，就跟JVM内存中的from区to区一样。但是我还是希望它的容量是可以降下来的，也算是自动调节吧，所以采用了这种方案。

注册Hook

最后我希望程序在正常退出的情况下，能够立马执行一次保存数据的任务，所以我在构造函数这里添加一个ShutdownHook，让它去执行存入数据库的操作，尽量保证数据不丢失。

public VisitRecordService() {
   Runtime.getRuntime().addShutdownHook(new Thread(() -> {
      this.batchSave();
      threadPool.shutdown();
   }));
}

后记

本文向大家分享了一种使用 缓存 + 异步 来存储访问日志的方式，其实不止是访问日志，有其他类似场景的地方，也可以使用这种方案，我个人觉得是非常棒的。

如果有什么意见或者建议，欢迎在评论区留言告诉我，毕竟我也是菜鸡，大家相互学习相互进步嘛。

如果你觉得我的思路还不错的话，麻烦在评论去告诉我一下，让我也开心开心。

基于异步+缓存的请求日志批量保存方案，拿去写在你的简历上吧！