SpringBoot实现QPS监控的原理与高性能实战

package com.example.qps.monitor;

import java.util.concurrent.atomic.LongAdder;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * 基于 RingBuffer 和 LongAdder 实现的高性能滑动窗口计数器
 */
public class SlidingWindowCounter {

    // 窗口大小（秒）
    private final int windowSize;
    // 槽位数量，默认 1 秒一个槽位
    private final int slotCount;
    // 环形数组，存储每个时间片的计数
    private final LongAdder[] slots;
    // 读写锁，用于周期性清理过期数据时的并发控制
    private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
    // 上次清理时间（纳秒）
    private volatile long lastClearTime;

    public SlidingWindowCounter(int windowSize) {
        this.windowSize = windowSize;
        this.slotCount = windowSize; // 假设粒度为 1s
        this.slots = new LongAdder[slotCount];
        for (int i = 0; i < slotCount; i++) {
            slots[i] = new LongAdder();
        }
        this.lastClearTime = System.currentTimeMillis();
    }

    /**
     * 记录一次请求
     */
    public void record() {
        // 1. 检查是否需要清理过期数据
        checkAndClearExpiredSlots();

        // 2. 获取当前槽位并累加
        int index = getCurrentSlotIndex();
        slots[index].increment();
    }

    /**
     * 获取当前窗口的总 QPS
     */
    public long getQps() {
        long total = 0;
        try {
            // 读锁：允许并发读取，但禁止在清理时读取
            lock.readLock().lock();
            checkAndClearExpiredSlots();
            for (LongAdder slot : slots) {
                total += slot.longValue();
            }
        } finally {
            lock.readLock().unlock();
        }
        // 注意：这里返回的是窗口内的总请求数。
        // 如果要算平均 QPS，应除以有效时间片数量。
        // 为了简化，此处通常暴露的是“最近 N 秒的总请求数”，由 Prometheus rate() 计算 QPS。
        // 或者我们可以直接算平均 QPS = total / validSlotCount。
        return total;
    }

    /**
     * 获取当前槽位索引
     */
    private int getCurrentSlotIndex() {
        long now = System.currentTimeMillis();
        long second = now / 1000;
        return (int) (second % slotCount);
    }

    /**
     * 清理过期数据（防止 RingBuffer 数据重叠）
     * 简单判断：如果当前时间与上次清理时间跨过了一个窗口周期，则重置数组
     */
    private void checkAndClearExpiredSlots() {
        long now = System.currentTimeMillis();
        // 如果已经过了一个完整的窗口周期
        if (now - lastClearTime >= windowSize * 1000L) {
            lock.writeLock().lock();
            try {
                // 双重检查
                if (now - lastClearTime >= windowSize * 1000L) {
                    // 重置所有槽位
                    // 注意：在高并发下直接 new LongAdder[] 或者遍历 reset()
                    // 这里为了极致性能，采用遍历 reset
                    for (LongAdder slot : slots) {
                        slot.reset();
                    }
                    lastClearTime = now;
                }
            } finally {
                lock.writeLock().unlock();
            }
        }
    }
}

package com.example.qps.monitor;

import io.micrometer.core.instrument.Gauge;
import io.micrometer.core.instrument.MeterRegistry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.filter.OncePerRequestFilter;
import org.springframework.web.util.UrlPathHelper;

import javax.servlet.FilterChain;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Component
public class QpsMonitorFilter extends OncePerRequestFilter {

    private static final Logger log = LoggerFactory.getLogger(QpsMonitorFilter.class);

    // 存储每个 URI 的计数器
    private final Map<String, SlidingWindowCounter> counterMap = new ConcurrentHashMap<>();
    
    @Autowired
    private MeterRegistry meterRegistry;

    // 窗口大小 60s
    private static final int WINDOW_SIZE = 60;

    @Override
    protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, FilterChain filterChain)
            throws ServletException, IOException {
        
        String uri = request.getRequestURI();
        
        // 1. 获取或创建该 URI 的计数器
        // computeIfAbsent 保证线程安全
        SlidingWindowCounter counter = counterMap.computeIfAbsent(uri, key -> {
            SlidingWindowCounter newCounter = new SlidingWindowCounter(WINDOW_SIZE);
            // 2. 注册到 Micrometer
            Gauge.builder("app.request.qps.total", newCounter, SlidingWindowCounter::getQps)
                    .tags("uri", key) // 维度标签
                    .description("Total requests in sliding window")
                    .register(meterRegistry);
            return newCounter;
        });

        // 3. 记录请求
        counter.record();

        // 4. 放行
        filterChain.doFilter(request, response);
    }
}

// 简单 LRU 改造示例
public class LruCounterMap<K, V> extends LinkedHashMap<K, V> {
    private static final int MAX_CAPACITY = 500;
    
    public LruCounterMap() {
        super(MAX_CAPACITY, 0.75f, true);
    }

    @Override
    protected boolean removeEldestEntry(Map.Entry<K, V> eldest) {
        return size() > MAX_CAPACITY;
    }
}