項目地址

https://gitee.com/dromara/Akali

項目介紹

Akali（阿卡麗）是一個輕量級本地化熱點檢測/降級框架，適用于大流量場景，可輕松解決業務中超高流量的并發查詢等場景。并且接入和使用極其簡單，10秒鐘即可接入使用！

Akali框架的理念就是小巧，實用，來無影去無蹤，絲血團戰，滿血退場，所到之處，皆為虛無。

核心功能

對于核心方法，發現該方法高頻使用，要么使用原有的數據進行返回（@AkaliHot），要么使用指定的方法進行降級（@AkaliFallback）。

源碼拆解

系統啟動

AkaliScanner實現了InstantiationAwareBeanPostProcessor，如果注冊到Spring容器中的Bean存在AkaliFallback和AkaliHot注解標注的方法，創建代理類AkaliProxy。

public class AkaliScanner implements InstantiationAwareBeanPostProcessor {@Overridepublic 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 {@Overridepublic 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<>();@Overridepublic AkaliStrategyEnum getStrategy() {return AkaliStrategyEnum.FALLBACK;}@Overridepublic 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);}@Overridepublic AkaliStrategyEnum getStrategy() {return AkaliStrategyEnum.HOT_METHOD;}@Overridepublic 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;}}
}