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

项目地址

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;
        }
    }
}