一个注解搞定责任链，学还是不学？
阿
余敦刚(玄苏)约 11 分钟
余敦刚(玄苏)约 11 分钟
背景
在繁琐的业务流程处理中，通常采用面向过程的设计方法将流程拆分成N个步骤，每个步骤执行独立的逻辑。
public void process(params){doFirst(params);doSecond(params);....doLast(params);}
- 正确性：实现复杂度较高，短时间手工编写容易出错
- 开发效率：涉及多个类的实现，需要花费不少时间进行测试非业务的pipeline流程，ROI不高
- 复用性：不同业务流程难以复用同一套pipeline的关键代码
Quirk Start
下面以 读取系统配置 为例，读取逻辑如下：
- 从本地配置文件读取，读取成功则直接返回，否则执行下一步
- 从系统变量读取，返回对应的值
public interface ConfigSource {String get(String key);}
以下大部分内容引至auto-pipeline官网： https://github.com/foldright/auto-pipeline
- 引入Maven依赖
<dependency><groupId>com.foldright.auto-pipeline</groupId><artifactId>auto-pipeline-processor</artifactId><version>0.2.0</version><scope>provided</scope></dependency>
- 在需要生成pipeline的接口上加上@AutoPipeline
@AutoPipelinepublic interface ConfigSource {String get(String key);}
- 实现pipeline的handler
public class MapConfigSourceHandler implements ConfigSourceHandler {private final Map<String, String> map;public MapConfigSourceHandler(Map<String, String> map) {this.map = map;}@Overridepublic String get(String key, ConfigSourceHandlerContext context) {String value = map.get(key);if (StringUtils.isNotBlank(value)) {return value;}return context.get(key);}}public class SystemConfigSourceHandler implements ConfigSourceHandler {public static final SystemConfigSourceHandler INSTANCE = new SystemConfigSourceHandler();@Overridepublic String get(String key, ConfigSourceHandlerContext context) {String value = System.getProperty(key);if (StringUtils.isNotBlank(value)) {return value;}return context.get(key);}}
- 使用pipeline
Map<String, String> mapConfig = new HashMap<String, String>();mapConfig.put("hello", "world");ConfigSourceHandler mapConfigSourceHandler = new MapConfigSourceHandler(mapConfig);ConfigSource pipeline = new ConfigSourcePipeline().addLast(mapConfigSourceHandler).addLast(SystemConfigSourceHandler.INSTANCE);pipeline.get("hello");// get value "world"// from mapConfig / mapConfigSourceHandlerpipeline.get("java.specification.version")// get value "1.8"// from system properties / SystemConfigSourceHandler
实现原理
业务接口通过生成的Pipeline构造实现，Pipeline负责责任链的组装及调用链表的首个节点(head)。首个节点如果处理完成有返回值，则直接返回；否则传递给下一个节点。如果处理到最后一个节点(tail)返回仍然为空，则直接返回空。
- 用户实现：
- ConfigSource
- 用户自定义的 获取配置的接口
- Handler实现：
- MapConfigSourceHandler 、SystemConfigSourceHandler
- AutoPipeline生成
- ConfigSourcePipeline
- 含义：责任链管道
- 核心作用：将ConfigSourceHandler 串联成链表
- ConfigSourceHandlerContext
- 含义：Handler的上下文，相比传统责任链，新增了获取全局Pipeline的能力
- AbstractConfigSourceHandlerContext
- 含义：Handler的上下文的抽象类
- 数据结构：主要由三个部分组成：pre、next、handler
- 核心作用：通过handler().get(key , findNextCtx()) 实现了 String get(String key) 方法
- DefaultConfigSourceHandlerContext
- 持有ConfigSourceHandler对象的默认实现类
源码解读
- auto-pipeline-annotations
- 框架包含的注解：AutoPipeline、PipelineDirection
- AutoPipeline
- 生成pipeline的核心注解
- PipelineDirection
- pipeline处理的顺序方向
- auto-pipeline-processor
- AutoPipelineProcessor
- 生成pipeline的入口类
- SourceGeneratorFacade
- 源代码生成器
- auto-pipeline-examples
- 一些实例，比如获取配置、rpc、merger
注册编译时注解
-
编写注解类： AutoPipelineProcessor -
继承JDK的 AbstractProcessor , 实现process 方法 -
在 resources 目录下新建文件夹：META-INF/services -
在META-INF/services 里面新增spi文件： javax.annotation.processing.Processor ，文件写入需要继承 AbstractProcessor 的全类名
- Processor
- 提供注解处理，它遵循SPI规约进行拓展
- AbstractProcessor
- 注解处理器主要拓展处理类
生成源代码
JDK术语介绍：- ProcessingEnvironment
- 注解处理工具的集合
- Element
- 是一个接口，表示一个程序元素，它可以是包、类、方法或者一个变量
- PackageElement
- 表示一个包程序元素，提供对有关包及其成员的信息的访问。
- ExecutableElement
- 表示某个类或接口的方法、构造方法或初始化程序（静态或实例），包括注释类型元素。
- TypeElement
- 表示一个类或接口程序元素，提供对有关类型及其成员的信息的访问。注意，枚举类型是一种类，而注解类型是一种接口。
- VariableElement
- 表示一个字段、enum 常量、方法或构造方法参数、局部变量或异常参数。
- Filer
- 文件管理器，主要负责生成源代码、class 或者辅助文件
- TypeSpec
- 用于生成类、接口、枚举的工具类
- MethodSpec
- 用于生成构造方法或者普通的方法的工具类
关键代码解读
生成入口：AutoPipelineProcessor#process
override fun process(annotations: Set<TypeElement>, roundEnv: RoundEnvironment): Boolean {val elements = roundEnv.getElementsAnnotatedWith(AutoPipeline::class.java)if (elements.isEmpty()) {return false}for (element in elements) {if (element.kind != ElementKind.INTERFACE) {error(element, "${(element as TypeElement).qualifiedName} is not a interface! Only interface can annotated with @${AutoPipeline::class.simpleName}")return false}if (!element.modifiers.contains(Modifier.PUBLIC)) {error(element, "interface ${(element as TypeElement).qualifiedName} is not public! Only public interface can annotated with @${AutoPipeline::class.simpleName}")return false}if (element is TypeElement) {doProcess(element)}}return false}
- 通过roundEnv 获取所有被AutoPipeline注释修饰的类，如果没有则直接返回
- 遍历elements，处理每个element (被注解的类必须是public修饰的接口)
生成源码门户：
SourceGeneratorFacade#genSourceCode
源代码生成类：HandlerGenerator#gen
下面以HandlerGenerator#gen 为例:fun gen() {// 生成类val handlerTypeBuilder = TypeSpec.interfaceBuilder(desc.handlerRawClassName).addTypeVariables(desc.entityDeclaredTypeVariables).addModifiers(Modifier.PUBLIC)// 构建handlerContext参数val contextParam = ParameterSpec.builder(desc.handlerContextTypeName, desc.handlerContextRawClassName.asFieldName()).build()// 为原来接口的每个方法额外添加handlerContext参数desc.entityMethods.forEach {val operationMethod = createMethodSpecBuilder(it.executableElement).addParameter(contextParam).build()handlerTypeBuilder.addMethod(operationMethod)}// 生成源码javaFileBuilder(desc.handlerRawClassName.packageName(), handlerTypeBuilder.build()).build().writeTo(filer)}
- 在项目的maven compile上右键，点击Debug '${moduleName}'
- 在AutoPipelineProcessor#process方法上加上断连，即可断点Debug源码
场景实战
下面举一个项目中真实的例子-消息分级限流。 消息发送的流量现状：
- 同一个请求可能包含有多个AppKey的消息
- 同一个请求可能包含多个消息分级的消息
- 同一个请求的消息可能经过多个接口
- 每个消息都会有对应的Appkey、消息分级
- 需要对消息所属的AppKey进行单独限流
- 仅对营销类消息进行限流，IM&实时类消息无需限流
- 如果一个消息已经被一个接口限流过，经过下一个接口时不应该被限流
- 对于同一个请求，只有整体限流和整体不限流 两种情况，不允许部分成功部分失败的情况(历史遗留问题)
- 首先是将限流规则拆分成三个步骤：消息分级处理、去重处理、请求限流令牌处理
- 将整体限流和整体不限流抽象成合并策略，通过proxy的方式对外暴露
- 限流接口类
/*** 消息限流器*/@AutoPipelinepublic interface MessageThrottler {/*** 节流单个消息** @param messageThrottlerToken 消息限流令牌* @return 是否被节流*/boolean throttle(MessageThrottlerToken messageThrottlerToken);/*** 节流多个消息。任意一个消息被节流将返回true,否则返回false** @param messageThrottlerTokens 多个消息限流令牌* @return 是否被节流*/boolean anyThrottle(List<MessageThrottlerToken> messageThrottlerTokens);/*** 节流多个消息。所有消息被节流才会返回true, 否则返回false** @param messageThrottlerTokens 多个消息限流令牌* @return 是否被节流*/boolean allThrottle(List<MessageThrottlerToken> messageThrottlerTokens);}
- 将限流规则拆分成三个不同的处理类
- ClassificationThrottlerHandler
/*** 消息分类节流器** <p>* 目前仅针对营销消息进行节流*/public class ClassificationThrottlerHandler implements MessageThrottlerHandler {@Overridepublic boolean throttle(MessageThrottlerToken messageThrottlerToken, MessageThrottlerHandlerContext context) {if (!ClassificationConstant.MARKETING.equals(messageThrottlerToken.getClassification())) {return false;}return context.throttle(messageThrottlerToken);}@Overridepublic boolean anyThrottle(List<MessageThrottlerToken> messageThrottlerTokens, MessageThrottlerHandlerContext context) {if (CollectionUtils.isEmpty(messageThrottlerTokens)) {return false;}// 获取营销消息List<MessageThrottlerToken> marketingMessageThrottlerTokens = messageThrottlerTokens.stream().filter(messageThrottlerToken -> {return ClassificationConstant.MARKETING.equals(messageThrottlerToken.getClassification());}).collect(Collectors.toList());// 如果营销消息为空，说明消息均不需要被限流，直接返回falseif (CollectionUtils.isEmpty(marketingMessageThrottlerTokens)) {return false;}return context.anyThrottle(marketingMessageThrottlerTokens);}@Overridepublic boolean allThrottle(List<MessageThrottlerToken> messageThrottlerTokens, MessageThrottlerHandlerContext context) {if (CollectionUtils.isEmpty(messageThrottlerTokens)) {return false;}// 获取营销消息List<MessageThrottlerToken> marketingMessageThrottlerTokens = messageThrottlerTokens.stream().filter(messageThrottlerToken -> {return ClassificationConstant.MARKETING.equals(messageThrottlerToken.getClassification());}).collect(Collectors.toList());// 存在非营销消息，非营销消息不会被限流if (marketingMessageThrottlerTokens.size() < messageThrottlerTokens.size()) {return false;}return context.allThrottle(marketingMessageThrottlerTokens);}}
- DuplicateThrottlerHandler
@Slf4jpublic class DuplicateThrottlerHandler implements MessageThrottlerHandler {@Overridepublic boolean throttle(MessageThrottlerToken messageThrottlerToken, MessageThrottlerHandlerContext context) {if (messageThrottlerToken.isThrottled()) {return false;}boolean throttleResult = context.throttle(messageThrottlerToken);messageThrottlerToken.markThrottled();return throttleResult;}@Overridepublic boolean anyThrottle(List<MessageThrottlerToken> messageThrottlerTokens, MessageThrottlerHandlerContext context) {if (CollectionUtils.isEmpty(messageThrottlerTokens)) {return false;}// 过滤掉已经被限流的消息List<MessageThrottlerToken> needMessageThrottlerTokens = messageThrottlerTokens.stream().filter(messageThrottlerToken -> !messageThrottlerToken.isThrottled()).collect(Collectors.toList());if (CollectionUtils.isEmpty(needMessageThrottlerTokens)) {return false;}boolean throttleResult = context.anyThrottle(needMessageThrottlerTokens);needMessageThrottlerTokens.forEach(messageThrottlerToken -> messageThrottlerToken.markThrottled());return throttleResult;}@Overridepublic boolean allThrottle(List<MessageThrottlerToken> messageThrottlerTokens, MessageThrottlerHandlerContext context) {if (CollectionUtils.isEmpty(messageThrottlerTokens)) {return false;}// 过滤掉已经被限流的消息List<MessageThrottlerToken> needMessageThrottlerTokens = messageThrottlerTokens.stream().filter(messageThrottlerToken -> !messageThrottlerToken.isThrottled()).collect(Collectors.toList());if (CollectionUtils.isEmpty(needMessageThrottlerTokens)) {return false;}boolean throttleResult = context.allThrottle(needMessageThrottlerTokens);needMessageThrottlerTokens.forEach(messageThrottlerToken -> messageThrottlerToken.markThrottled());return throttleResult;}}
- AcquireThrottlerHandler
/*** 请求令牌处理类*/public class AcquireThrottlerHandler implements MessageThrottlerHandler {private static final Logger apiThrottlerLog = LoggerFactory.getLogger("api.throttler.log");@Autowiredprivate ThrottlerProxy throttlerProxy;@Autowiredprivate ThrottlerModeConfiguration throttlerModeConfiguration;private boolean throttle(AcquireToken acquireToken) {// 获取限流模式ThrottlerMode throttlerMode = throttlerModeConfiguration.getThrottlerMode(acquireToken.getAppKey(), acquireToken.getThrottleTag());// 执行限流return !throttlerProxy.tryAcquireWithAppKey(throttlerMode, acquireToken.getAppKey(), acquireToken.getPermits());}@Overridepublic boolean throttle(MessageThrottlerToken messageThrottlerToken, MessageThrottlerHandlerContext context) {boolean throttled = throttle(new AcquireToken(messageThrottlerToken.getThrottleTag(), messageThrottlerToken.getAppKey(), messageThrottlerToken.getPermits()));// 限流日志埋点if (SendSwitch.THROTTLER_ONLY_WATCH || throttled) {log(messageThrottlerToken.getAppKey(), messageThrottlerToken.getPermits(), messageThrottlerToken.getThrottleTag(), throttled);}return throttled;}@Overridepublic boolean anyThrottle(List<MessageThrottlerToken> messageThrottlerTokens, MessageThrottlerHandlerContext context) {return throttle(messageThrottlerTokens, acquireTokens -> acquireTokens.stream().anyMatch(this::throttle));}@Overridepublic boolean allThrottle(List<MessageThrottlerToken> messageThrottlerTokens, MessageThrottlerHandlerContext messageThrottlerHandlerContext) {return throttle(messageThrottlerTokens, acquireTokens -> acquireTokens.stream().allMatch(this::throttle));}private static boolean throttle(List<MessageThrottlerToken> messageThrottlerTokens, Function<List<AcquireToken>, Boolean> function) {if (CollectionUtils.isEmpty(messageThrottlerTokens)) {return false;}List<AcquireToken> acquireTokens = messageThrottlerTokens.stream().collect(Collectors.groupingBy(messageThrottlerToken -> messageThrottlerToken.getAppKey())).entrySet().stream().map(messageEntry -> {String appKey = messageEntry.getKey();int permits = messageEntry.getValue().stream().map(messageThrottlerToken -> messageThrottlerToken.getPermits()).reduce(Integer::sum).orElse(1);String throttlerTag = messageEntry.getValue().get(0).getThrottleTag();return new AcquireToken(throttlerTag, appKey, permits);}).collect(Collectors.toList());boolean throttled = function.apply(acquireTokens);// 限流日志埋点if (SendSwitch.THROTTLER_ONLY_WATCH || throttled) {messageThrottlerTokens.forEach(messageThrottlerToken -> {log(messageThrottlerToken.getAppKey(), messageThrottlerToken.getPermits(), messageThrottlerToken.getThrottleTag(), throttled);});}return throttled;}private static void log(String appKey, int permits, String throttlerTag, boolean throtted) {List<String> metrics = new ArrayList<>();metrics.add(appKey);metrics.add(String.valueOf(permits));metrics.add(throttlerTag);metrics.add(String.valueOf(throtted));String logContent = StringUtils.join(metrics, "|");apiThrottlerLog.info(logContent);}@Data@AllArgsConstructorprivate static class AcquireToken {private final String throttleTag;private final String appKey;private final int permits;}}
- 消息限流代理
/*** 消息限流代理*/@Slf4jpublic class MessageThrottlerProxy {@Autowiredprivate AcquireThrottlerHandler acquireThrottlerHandler;private MessageThrottler messageThrottler;@PostConstructpublic void init() {messageThrottler = new MessageThrottlerPipeline().addLast(new ClassificationThrottlerHandler()).addLast(new DuplicateThrottlerHandler()).addLast(acquireThrottlerHandler);}/*** 限流单个消息** @param messageThrottlerToken 单个消息令牌* @return 是否限流成功*/public boolean throttle(MessageThrottlerToken messageThrottlerToken) {if (!SendSwitch.ENABLE_API_THROTTLER) {return false;}try {boolean throttled = messageThrottler.throttle(messageThrottlerToken);return SendSwitch.THROTTLER_ONLY_WATCH ? false : throttled;} catch (Exception e) {log.error("Failed to throttle messageSendDTO:" + messageThrottlerToken, e);// throttle内部异常不应该影响正常请求，遇到此情况直接降级限流通过return false;}}/*** 限流多个消息, 合并策略可通过 {@link SendSwitch#THROTTLER_MERGE_STRATEGY} 开关控制** @param messageThrottlerTokens 多个消息令牌* @return 是否限流成功*/public boolean throttle(List<MessageThrottlerToken> messageThrottlerTokens) {if (!SendSwitch.ENABLE_API_THROTTLER) {return false;}if (CollectionUtils.isEmpty(messageThrottlerTokens)) {return false;}MergeStrategy mergeStrategy = MergeStrategy.getByName(SendSwitch.THROTTLER_MERGE_STRATEGY);if (mergeStrategy == null) {log.error("illegal throttler mergeStrategy:" + SendSwitch.THROTTLER_MERGE_STRATEGY);return false;}try {boolean throttled = mergeStrategy.throttle(messageThrottler, messageThrottlerTokens);return SendSwitch.THROTTLER_ONLY_WATCH ? false : throttled;} catch (Exception e) {log.error("Failed to throttle messageSendDTO:" + messageThrottlerTokens, e);// throttle内部异常不应该影响正常请求，遇到此情况直接降级限流通过return false;}}public enum MergeStrategy {ALL {@Overridepublic boolean throttle(MessageThrottler messageThrottler, List<MessageThrottlerToken> messageThrottlerTokens) {return messageThrottler.allThrottle(messageThrottlerTokens);}},ANY {@Overridepublic boolean throttle(MessageThrottler messageThrottler, List<MessageThrottlerToken> messageThrottlerTokens) {return messageThrottler.anyThrottle(messageThrottlerTokens);}};public static MergeStrategy getByName(String name) {MergeStrategy[] values = values();for (MergeStrategy value : values) {if (value.name().equalsIgnoreCase(name)) {return value;}}return null;}public abstract boolean throttle(MessageThrottler messageThrottler, List<MessageThrottlerToken> messageThrottlerTokens);}}
推荐阅读
3. 如何做好“防御性编码”？
千万阅读量、百万下载量、上百本电子书，近200位阿里专家参与编写。多元化选择、全领域覆盖，汇聚阿里巴巴技术实践精华，读、学、练一键三连。开发者藏经阁，开发者的工作伴侣~
点击阅读原文查看详情。