业务背景

有时候日志的信息比较多，怎么样才可以让系统做到自适应采样呢？

日志开源组件（一）java 注解结合 spring aop 实现自动输出日志

日志开源组件（二）java 注解结合 spring aop 实现日志traceId唯一标识

日志开源组件（三）java 注解结合 spring aop 自动输出日志新增拦截器与过滤器

日志开源组件（四）如何动态修改 spring aop 切面信息？让自动日志输出框架更好用

日志开源组件（五）如何将 dubbo filter 拦截器原理运用到日志拦截器中？

自适应采样

系统生成的日志可以包含大量信息，包括错误、警告、性能指标等，但在实际应用中，处理和分析所有的日志数据可能会对系统性能和资源产生负担。

自适应采样在这种情况下发挥作用，它能够根据当前系统状态和日志信息的重要性，智能地决定哪些日志需要被采样记录，从而有效地管理和分析日志数据。

日志采样系统会给业务系统额外增加消耗，很多系统在接入的时候会比较排斥。

给他们一个百分比的选择，或许是一个不错的开始，然后根据实际需要选择合适的比例。

自适应采样是一个对用户透明，同时又非常优雅的方案。

如何通过 java 实现自适应采样?

首先我们定义一个接口，返回 boolean。

根据是否为 true 来决定是否输出日志。

/**
 * 采样条件
 * @author binbin.hou
 * @since 0.5.0
 */
public interface IAutoLogSampleCondition {

    /**
     * 条件
     *
     * @param context 上下文
     * @return 结果
     * @since 0.5.0
     */
    boolean sampleCondition(IAutoLogContext context);

}

我们先实现一个简单的概率采样。

0-100 的值，让用户指定，按照百分比决定是否采样。

public class InnerRandomUtil {

    /**
     * 1. 计算一个 1-100 的随机数 randomVal
     * 2. targetRatePercent 值越大，则返回 true 的概率越高
     * @param targetRatePercent 目标百分比
     * @return 结果
     */
    public static boolean randomRateCondition(int targetRatePercent) {
        if(targetRatePercent <= 0) {
            return false;
        }
        if(targetRatePercent >= 100) {
            return true;
        }

        // 随机
        ThreadLocalRandom threadLocalRandom = ThreadLocalRandom.current();
        int value = threadLocalRandom.nextInt(1, 100);

        // 随机概率
        return targetRatePercent >= value;
    }

}

实现起来也非常简单，直接一个随机数，然后比较大小即可。

思路

我们计算一下当前日志的 QPS，让输出的概率和 QPS 称反比。

/**
 * 自适应采样
 *
 * 1. 初始化采样率为 100%，全部采样
 *
 * 2. QPS 如果越来越高，那么采样率应该越来越低。这样避免 cpu 等资源的损耗。最低为 1%
 * 如果 QPS 越来越低，采样率应该越来越高。增加样本，最高为 100%
 *
 * 3. QPS 如何计算问题
 *
 * 直接设置大小为 100 的队列，每一次在里面放入时间戳。
 * 当大小等于 100 的时候，计算首尾的时间差，currentQps = 100 / (endTime - startTime) * 1000
 *
 * 触发 rate 重新计算。
 *
 * 3.1 rate 计算逻辑
 *
 * 这里我们存储一下 preRate = 100, preQPS = ?
 *
 * newRate = (preQps / currentQps) * rate
 *
 * 范围限制：
 * newRate = Math.min(100, newRate);
 * newRate = Math.max(1, newRate);
 *
 * 3.2 时间队列的清空
 *
 * 更新完 rate 之后，对应的队列可以清空？
 *
 * 如果额外使用一个 count，好像也可以。
 * 可以调整为 atomicLong 的计算器，和 preTime。
 *

代码实现

public class AutoLogSampleConditionAdaptive implements IAutoLogSampleCondition {

    private static final AutoLogSampleConditionAdaptive INSTANCE = new AutoLogSampleConditionAdaptive();

    /**
     * 单例的方式获取实例
     * @return 结果
     */
    public static AutoLogSampleConditionAdaptive getInstance() {
        return INSTANCE;
    }

    /**
     * 次数大小限制，即接收到多少次请求更新一次 adaptive 计算
     *
     * TODO: 这个如何可以让用户可以自定义呢？后续考虑配置从默认的配置文件中读取。
     */
    private static final int COUNT_LIMIT = 1000;

    /**
     * 自适应比率，初始化为 100.全部采集
     */
    private volatile int adaptiveRate = 100;

    /**
     * 上一次的 QPS
     *
     * TODO: 这个如何可以让用户可以自定义呢？后续考虑配置从默认的配置文件中读取。
     */
    private volatile double preQps = 100.0;

    /**
     * 上一次的时间
     */
    private volatile long preTime;

    /**
     * 总数，请求计数器
     */
    private final AtomicInteger counter;

    public AutoLogSampleConditionAdaptive() {
        preTime = System.currentTimeMillis();
        counter = new AtomicInteger(0);
    }

    @Override
    public boolean sampleCondition(IAutoLogContext context) {
        int count = counter.incrementAndGet();

        // 触发一次重新计算
        if(count >= COUNT_LIMIT) {
            updateAdaptiveRate();
        }

        // 直接计算是否满足
        return InnerRandomUtil.randomRateCondition(adaptiveRate);
    }

}

每次累加次数超过限定次数之后，我们就更新一下对应的日志概率。

最后的概率计算和上面的百分比类似，不再赘述。

/**
 * 更新自适应的概率
 *
 * 100 计算一次，其实还好。实际应该可以适当调大这个阈值，本身不会经常变化的东西。
 */
private synchronized void updateAdaptiveRate() {
    //消耗的毫秒数
    long costTimeMs = System.currentTimeMillis() - preTime;
    //qps 的计算，时间差是毫秒。所以次数需要乘以 1000
    double currentQps = COUNT_LIMIT*1000.0 / costTimeMs;
    // preRate * preQps = currentRate * currentQps; 保障采样均衡，服务器压力均衡
    // currentRate = (preRate * preQps) / currentQps;
    // 更新比率
    int newRate = 100;
    if(currentQps > 0) {
        newRate = (int) ((adaptiveRate * preQps) / currentQps);
        newRate = Math.min(100, newRate);
        newRate = Math.max(1, newRate);
    }
    // 更新 rate
    adaptiveRate = newRate;
    // 更新 QPS
    preQps = currentQps;
    // 更新上一次的时间内戳
    preTime = System.currentTimeMillis();
    // 归零
    counter.set(0);
}

问题

上面的自适应算法一般情况下都可以运行的很好。

但是有一种情况会不太好，那就是流量从高峰期到低峰期。

比如凌晨11点是请求高峰期，我们的输出日志概率很低。深夜之后请求数会很少，想达到累计值就会很慢，这个时间段就会导致日志输出很少。

如何解决这个问题呢？

思路

我们可以通过固定时间窗口的方式，来定时调整流量概率。

java 实现

我们初始化一个定时任务，1min 定时更新一次。

public class AutoLogSampleConditionAdaptiveSchedule implements IAutoLogSampleCondition {

    private static final ScheduledExecutorService EXECUTOR_SERVICE = Executors.newSingleThreadScheduledExecutor();

    /**
     * 时间分钟间隔
     */
    private static final int TIME_INTERVAL_MINUTES = 5;

    /**
     * 自适应比率，初始化为 100.全部采集
     */
    private volatile int adaptiveRate = 100;

    /**
     * 上一次的总数
     *
     * TODO: 这个如何可以让用户可以自定义呢？后续考虑配置从默认的配置文件中读取。
     */
    private volatile long preCount;

    /**
     * 总数，请求计数器
     */
    private final AtomicLong counter;

    public AutoLogSampleConditionAdaptiveSchedule() {
        counter = new AtomicLong(0);
        preCount = TIME_INTERVAL_MINUTES * 60 * 100;

        //1. 1min 后开始执行
        //2. 中间默认 5 分钟更新一次
        EXECUTOR_SERVICE.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                updateAdaptiveRate();
            }
        }, 60, TIME_INTERVAL_MINUTES * 60, TimeUnit.SECONDS);
    }

    @Override
    public boolean sampleCondition(IAutoLogContext context) {
        counter.incrementAndGet();

        // 直接计算是否满足
        return InnerRandomUtil.randomRateCondition(adaptiveRate);
    }

}

其中更新概率的逻辑和上面类似：

/**
 * 更新自适应的概率
 *
 * QPS = count / time_interval
 *
 * 其中时间维度是固定的，所以可以不用考虑时间。
 */
private synchronized void updateAdaptiveRate() {
    // preRate * preCount = currentRate * currentCount; 保障采样均衡，服务器压力均衡
    // currentRate = (preRate * preCount) / currentCount;
    // 更新比率
    long currentCount = counter.get();
    int newRate = 100;
    if(currentCount != 0) {
        newRate = (int) ((adaptiveRate * preCount) / currentCount);
        newRate = Math.min(100, newRate);
        newRate = Math.max(1, newRate);
    }
    // 更新自适应频率
    adaptiveRate = newRate;
    // 更新 QPS
    preCount = currentCount;
    // 归零
    counter.set(0);
}