【开源项目】热点监测降级框架Akali源码解读

大家好，欢迎来到IT知识分享网。

项目地址

https://gitee.com/dromara/Akali

项目介绍

Akali（阿卡丽）是一个轻量级本地化热点检测/降级框架，适用于大流量场景，可轻松解决业务中超高流量的并发查询等场景。并且接入和使用极其简单，10秒钟即可接入使用！

Akali框架的理念就是小巧，实用，来无影去无踪，丝血团战，满血退场，所到之处，皆为虚无。

核心功能

对于核心方法，发现该方法高频使用，要么使用原有的数据进行返回（@AkaliHot），要么使用指定的方法进行降级（@AkaliFallback）。

源码拆解

系统启动

AkaliScanner实现了InstantiationAwareBeanPostProcessor，如果注册到Spring容器中的Bean存在AkaliFallback和AkaliHot注解标注的方法，创建代理类AkaliProxy。

public class AkaliScanner implements InstantiationAwareBeanPostProcessor { 
    @Override public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { 
    Class<?> clazz = bean.getClass(); if (AkaliStrategy.class.isAssignableFrom(clazz)){ 
    AkaliStrategyManager.addStrategy((AkaliStrategy) bean); return bean; } AtomicBoolean needProxy = new AtomicBoolean(false); List<Method> fallbackMethodList = new ArrayList<>(); List<Method> hotspotMethodList = new ArrayList<>(); Arrays.stream(clazz.getDeclaredMethods()).forEach(method -> { 
    AkaliFallback akaliFallback = searchAnnotation(method, AkaliFallback.class); if (ObjectUtil.isNotNull(akaliFallback)){ 
    fallbackMethodList.add(method); S.addMethodStr(MethodUtil.resolveMethodName(method), new Tuple2<>(AkaliStrategyEnum.FALLBACK, akaliFallback)); needProxy.set(true); } AkaliHot akaliHot = searchAnnotation(method, AkaliHot.class); if (ObjectUtil.isNotNull(akaliHot)){ 
    hotspotMethodList.add(method); AkaliMethodManager.addMethodStr(MethodUtil.resolveMethodName(method), new Tuple2<>(AkaliStrategyEnum.HOT_METHOD, akaliHot)); needProxy.set(true); } }); if (needProxy.get()){ 
    try{ 
    AkaliProxy akaliProxy = new AkaliProxy(bean, fallbackMethodList, hotspotMethodList); Object obj = akaliProxy.proxy(); return obj; }catch (Exception e){ 
    throw new BeanInitializationException(e.getMessage()); } }else{ 
    return bean; } } } 

系统拦截

AopInvocationHandler，拦截器的核心方法。注册对应的FlowRule，执行SphEngine.process

 public class AopInvocationHandler implements InvocationHandler { 
    @Override public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { 
    String methodStr = MethodUtil.resolveMethodName(method); if (AkaliMethodManager.contain(methodStr)){ 
    AkaliStrategyEnum akaliStrategyEnum = AkaliMethodManager.getAnnoInfo(methodStr).r1; Annotation anno = AkaliMethodManager.getAnnoInfo(methodStr).r2; if (anno instanceof AkaliFallback){ 
    AkaliRuleManager.registerFallbackRule((AkaliFallback) anno, method); }else if (anno instanceof AkaliHot){ 
    AkaliRuleManager.registerHotRule((AkaliHot) anno, method); }else{ 
    throw new RuntimeException("annotation type error"); } return SphEngine.process(bean, method, args, methodStr, akaliStrategyEnum); }else { 
    return method.invoke(bean, args); } } } 

AkaliRuleManager#registerHotRule，利用Sentinel框架注册流控规则。

 public static void registerHotRule(AkaliHot akaliHot, Method method){ 
    String resourceKey = MethodUtil.resolveMethodName(method); if (!ParamFlowRuleManager.hasRules(resourceKey)){ 
    ParamFlowRule rule = new ParamFlowRule(); rule.setResource(MethodUtil.resolveMethodName(method)); rule.setGrade(akaliHot.grade().getGrade()); rule.setCount(akaliHot.count()); rule.setDurationInSec(akaliHot.duration()); rule.setParamIdx(0); ParamFlowRuleManager.loadRules(ListUtil.toList(rule)); log.info("[AKALI] Add Hot Rule [{}]", rule.getResource()); } } 

SphEngine的处理逻辑就是如果流控是允许的，执行核心方法，如果流控不允许，执行对应的策略。

public class SphEngine { 
    private static final Logger log = LoggerFactory.getLogger(SphEngine.class); public static Object process(Object bean, Method method, Object[] args, String methodStr, AkaliStrategyEnum akaliStrategyEnum) throws Exception{ 
    switch (akaliStrategyEnum){ 
    case FALLBACK: if (SphO.entry(methodStr)){ 
    try{ 
    return method.invoke(bean, args); }finally { 
    SphO.exit(); } }else{ 
    log.info("[AKALI]Trigger fallback strategy for [{}]", methodStr); return AkaliStrategyManager.getStrategy(akaliStrategyEnum).process(bean, method, args); } case HOT_METHOD: String convertParam = DigestUtil.md5Hex(JSON.toJSONString(args)); Entry entry = null; try{ 
    entry = SphU.entry(methodStr, EntryType.IN, 1, convertParam); return method.invoke(bean, args); }catch (BlockException e){ 
    log.info("[AKALI]Trigger hotspot strategy for [{}]", methodStr); return AkaliStrategyManager.getStrategy(akaliStrategyEnum).process(bean, method, args); }finally { 
    if (entry != null){ 
    entry.exit(1, convertParam); } } default: throw new Exception("[AKALI] Strategy error!"); } } } 

FallbackStrategy回退策略是获取方法名称是指定方法+Fallback的方法，进行方法调用。

public class FallbackStrategy implements AkaliStrategy{ 
    private final Map<String, Method> fallBackMethodMap = new ConcurrentHashMap<>(); @Override public AkaliStrategyEnum getStrategy() { 
    return AkaliStrategyEnum.FALLBACK; } @Override public Object process(Object bean, Method method, Object[] args) throws Exception{ 
    String fallbackMethodName = StrUtil.format("{}Fallback", method.getName()); Method fallbackMethod; if (fallBackMethodMap.containsKey(fallbackMethodName)){ 
    fallbackMethod = fallBackMethodMap.get(fallbackMethodName); }else{ 
    fallbackMethod = ReflectUtil.getMethod(bean.getClass(), fallbackMethodName, method.getParameterTypes()); fallBackMethodMap.put(fallbackMethodName, fallbackMethod); } if (ObjectUtil.isNull(fallbackMethod)){ 
    throw new RuntimeException(StrUtil.format("[AKALI] Can't find fallback method [{}] in bean [{}]", fallbackMethodName, bean.getClass().getName())); } return fallbackMethod.invoke(bean, args); } } 

MethodHotspotStrategy使用缓存，缓存中有数据就返回，没数据就调用方法。该地方使用的是hutool的缓存类。

public class MethodHotspotStrategy implements AkaliStrategy{ 
    private TimedCache<String, Object> timedCache; public MethodHotspotStrategy() { 
    timedCache = CacheUtil.newTimedCache(1000 * 60); timedCache.schedulePrune(1000); } @Override public AkaliStrategyEnum getStrategy() { 
    return AkaliStrategyEnum.HOT_METHOD; } @Override public Object process(Object bean, Method method, Object[] args) throws Exception{ 
    String hotKey = StrUtil.format("{}-{}", MethodUtil.resolveMethodName(method), DigestUtil.md5Hex(JSON.toJSONString(args))); if (timedCache.containsKey(hotKey)){ 
    return timedCache.get(hotKey); }else{ 
    Object result = method.invoke(bean, args); timedCache.put(hotKey, result); return result; } } }