标签：res local executors 构造函数 简单 优先级队列 ret 加锁 index
现在记录话单的时候想加一个参数：每秒接口调用的并发量，也就是所谓的QPS（Queries per second）。QPS即每秒请求数，是对一个特定的接口在规定时间内请求流量的衡量标准。那么如何实现QPS的计算呢？我想到的是两种方案：
1、一定时间内（比如一分钟）的请求总量/统计时间段（比如一分钟），最终得出就是每秒的并发量，它是基于某一段时间来统计的
2、直接统计一秒钟内的请求总量，就是按每秒的时间段来统计，简单粗暴
方案一的适用场景应该是报表、运维统计之类的，只关心QPS曲线；如果用来做并发量校验，明显只能用方案二，需要实时获取QPS。那么如何统计一秒内的并发量？假设某一个时间点有接口到来，那么就开始统计该接口，在一秒之内，来多少个累加多少次。一秒之后，统计数清零。之后的某一个时间点，又有接口到来，又开始统计一秒之内的接口调用量，如此循环往复。
那么如何维护一个一秒之内的接口计数器呢？我觉得失效缓存是一个合适的选择，缓存的键即为接口名，值就是接口统计数，过期时间一秒。为了避免引入第三方中间件，我们自己实现该过期缓存，需要维护一个定时器和一个优先级队列，每秒清理一次队列中已过期的缓存。
废话说完了，看代码：
1、缓存的值
import lombok.Getter; import lombok.Setter; import java.util.concurrent.atomic.AtomicLong; /** * 内部类，缓存对象，按失效时间排序，越早失效越前 * @author wulf * @since 20200422 */ @Getter @Setter public class CacheNode implements Comparable<CacheNode> { private String key; private AtomicLong callQuantity; private long expireTime; public CacheNode(String key, AtomicLong callQuantity, long expireTime) { this.key = key; this.callQuantity = callQuantity; this.expireTime = expireTime; } @Override public int compareTo(CacheNode o) { long dif = this.expireTime - o.expireTime; if (dif > 0) { return 1; } else if (dif < 0) { return -1; } return 0; } }
2、过期缓存：
import com.wlf.bean.CacheNode; import java.util.Map; import java.util.PriorityQueue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.ReentrantLock; /** * 带过期时间的缓存 * * @author wulf * @since 2020/04/21 */ public class ExpiredCache { // 缓存key=接口名，value=接口调用量、过期时间戳 private Map<String, CacheNode> cache = new ConcurrentHashMap<>(); // qps private AtomicLong qps = null; // 重入锁 private ReentrantLock lock = new ReentrantLock(); // 失效队列 private PriorityQueue<CacheNode> queue = new PriorityQueue<>(); // 启动定时任务，每秒清理一次过期缓存 private final static ScheduledExecutorService scheduleExe = new ScheduledThreadPoolExecutor(10); // 构造函数中启动定时任务，执行对已过期缓存的清理工作，每秒执行一次 public ExpiredCache() { scheduleExe.scheduleAtFixedRate(new CleanExpireCacheTask(), 1L, 1L, TimeUnit.SECONDS); } /** * 内部类，清理过期缓存对象 */ private class CleanExpireCacheTask implements Runnable { @Override public void run() { long currentTime = System.currentTimeMillis(); // 取出队列中的队头元素，对已过期的元素执行清除计划，剩下没有过期则退出 while (true) { lock.lock(); try { CacheNode cacheNode = queue.peek(); // 已经把队列清空了，或者所有过期元素已清空了，退出 if (cacheNode == null || cacheNode.getExpireTime() > currentTime) { return; } // 开始大清理了 cache.remove(cacheNode.getKey()); queue.poll(); } finally { lock.unlock(); } } } } /** * 根据缓存key获取values * * @param cacheKey * @return */ public CacheNode getCacheNode(String cacheKey) { return cache.get(cacheKey); } /** * 加入缓存，设置存活时间 * * @param cacheKey * @param ttl 缓存的存活时间 * return */ public AtomicLong set(String cacheKey, long ttl) { // 若缓存中已存在缓存节点，不需要更新过期时间，仅更新QPS值 CacheNode oldNode = cache.get(cacheKey); if (oldNode != null) { AtomicLong oldQps = oldNode.getCallQuantity(); oldQps.incrementAndGet(); cache.put(cacheKey, oldNode); } else { // 否则新创建CacheNode对象，失效时间=当前时间+缓存存活时间 AtomicLong qps = new AtomicLong(1); CacheNode newNode = new CacheNode(cacheKey, qps, System.currentTimeMillis() + ttl * 1000); // 放入缓存，加入过期队列 cache.put(cacheKey, newNode); queue.add(newNode); } return cache.get(cacheKey).getCallQuantity(); } }
3、在切面中统计接口QPS：
package com.wlf.cdr; import com.wlf.javabean.ots.TranslateCdr; import com.wlf.utils.ExpiredCache; import com.wlf.utils.IPUtil; import lombok.extern.slf4j.Slf4j; import org.aspectj.lang.ProceedingJoinPoint; import org.aspectj.lang.annotation.Around; import org.aspectj.lang.annotation.Aspect; import org.springframework.stereotype.Component; import org.springframework.web.context.request.RequestContextHolder; import org.springframework.web.context.request.ServletRequestAttributes; import javax.servlet.http.HttpServletRequest; import java.text.SimpleDateFormat; import java.util.Date; @Slf4j @Aspect @Component public class CdrAsept { private final static SimpleDateFormat SF = new SimpleDateFormat("yyyyMMddHHmmss"); // 话单格式：接口名称|话单记录时间|接口时延|调用方IP|本地IP|用户ID|用户名|源语言|目标语言|结果码|QPS private final static String CDR_FORMAT = "{}|{}|{}|{}|{}|{}|{}|{}|{}|{}|{}"; // 过期缓存 private ExpiredCache expiredCache = new ExpiredCache(); @Around("execution(* com.wlf.translateprovider.controller.TranslateController.*(..))") public Object recordCdr(ProceedingJoinPoint joinPoint) throws Throwable { long startTime = System.currentTimeMillis(); String startDate = SF.format(new Date(startTime)); // 白名单校验 ServletRequestAttributes attributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes(); HttpServletRequest httpServletRequest = attributes.getRequest(); String localIp = IPUtil.getLocalIp(); String remoteIp = IPUtil.getRemoteIp(httpServletRequest); TranslateCdr cdr = new TranslateCdr(); cdr.setRemoteIp(remoteIp); CdrThreadLocal.setTranslateCdr(cdr); // 获取接口名 String requestPath = httpServletRequest.getRequestURI(); String cacheKey = requestPath.substring(requestPath.lastIndexOf("/") + 1, requestPath.length()); // 设置过期时间为1秒 long qps = expiredCache.set(cacheKey, 1).get(); Object result = joinPoint.proceed(); long endTime = System.currentTimeMillis(); cdr = CdrThreadLocal.getTranslateCdr(); if (cdr != null) { log.error(CDR_FORMAT, cacheKey, startDate, endTime - startTime, remoteIp, localIp, cdr.getUserId(), cdr.getUserName(), cdr.getFrom(), cdr.getTo(), cdr.getResultCode(), qps); } CdrThreadLocal.delThreadLocal(); return result; } }
在切面中只需set一下，如果这时缓存有数据，就累加统计数，没有就设置统计数为1，再get出来的得到QPS。但这里为了兼顾吞吐量，让接口的调用不受QPS统计的影响，并没有在切面或者过期缓存的set方法加锁，因此对两个并发时间很短的接口，统计数会相同。
标签：res local executors 构造函数 简单 优先级队列 ret 加锁 index
原文地址：https://www.cnblogs.com/wuxun1997/p/12753548.html
