一个注解搞定责任链，学还是不学？

　　背景
　　在繁琐的业务流程处理中，通常采用面向过程的设计方法将流程拆分成N个步骤，每个步骤执行独立的逻辑。 public void process(params){     doFirst(params);     doSecond(params);     ....     doLast(params); }
　　但是这样剥离仍然是不彻底的，修改其中一个步骤仍然可能影响其他步骤(同一个类修改，不符合开闭原则)。在这种场景下，有一种经典的设计模式-责任链模式，可以将这些子步骤封装成独立的handler，然后通过pipeline将其串联起来。
　　常见的责任链模式会设计如下：
　　总体来看，纯手动编写有以下问题: 正确性：实现复杂度较高，短时间手工编写容易出错 开发效率：涉及多个类的实现，需要花费不少时间进行测试非业务的pipeline流程，ROI不高 复用性：不同业务流程难以复用同一套pipeline的关键代码
　　那有没有一套靠谱的框架能够解决上述问题呢？有的，它就是foldright/auto-pipeline，是责任链领域的＂lombok＂!
　　Quirk Start
　　下面以读取系统配置为例，读取逻辑如下： 从本地配置文件读取，读取成功则直接返回，否则执行下一步 从系统变量读取，返回对应的值
　　为了实现这个需求，读取配置接口定义如下： public interface ConfigSource {     String get(String key); }
　　如果使用auto-pipeline，该如何
　　以下大部分内容引至auto-pipeline官网： https://github.com/foldright/auto-pipeline
　　1.引入Maven依赖      com.foldright.auto-pipeline     auto-pipeline-processor     0.2.0     provided 
　　2.在需要生成pipeline的接口上加上@AutoPipeline
　　只需为这个接口加上@AutoPipeline @AutoPipeline public interface ConfigSource {     String get(String key);
　　3.实现pipeline的handler public class MapConfigSourceHandler implements ConfigSourceHandler {     private final Map map;       public MapConfigSourceHandler(Map map) {         this.map = map;     }       @Override     public 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();       @Override     public String get(String key, ConfigSourceHandlerContext context) {         String value = System.getProperty(key);         if (StringUtils.isNotBlank(value)) {             return value;         }         return context.get(key);     } }
　　4.使用pipeline Map mapConfig = new HashMap(); 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 / mapConfigSourceHandler   pipeline.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
　　生成原理
　　通过SPI的方式注册编译时注解@AutoPipelineProcessor，在编译过程中通过javapoet框架生成业务pipeline源代码。 注册编译时注解编写注解类：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 或者辅助文件
　　JavaPoet技术介绍： TypeSpec 用于生成类、接口、枚举的工具类 MethodSpec 用于生成构造方法或者普通的方法的工具类 关键代码解读生成入口：AutoPipelineProcessor#processoverride fun process(annotations: Set, 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
　　生成相关源代码，一个源文件采用一个特定的代码生成器生成，各个类的生成器继承AbstractGenerator
　　源代码生成类：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) }
　　编译Debug探秘
　　可以通过Idea Maven自带的Debug工具 调试编译过程 在项目的maven compile上右键，点击Debug ＂${moduleName}＂ 在AutoPipelineProcessor#process方法上加上断连，即可断点Debug源码
　　场景实战
　　下面举一个项目中真实的例子-消息分级限流。
　　消息发送的流量现状： 同一个请求可能包含有多个AppKey的消息 同一个请求可能包含多个消息分级的消息 同一个请求的消息可能经过多个接口 每个消息都会有对应的Appkey、消息分级
　　限流规则如下： 需要对消息所属的AppKey进行单独限流 仅对营销类消息进行限流，IM&实时类消息无需限流 如果一个消息已经被一个接口限流过，经过下一个接口时不应该被限流 对于同一个请求，只有整体限流和整体不限流 两种情况，不允许部分成功部分失败的情况(历史遗留问题)
　　面对这种的场景，该如何设计呢？ 首先是将限流规则拆分成三个步骤：消息分级处理、去重处理、请求限流令牌处理 将整体限流和整体不限流抽象成合并策略，通过proxy的方式对外暴露
　　代码设计如下： 限流接口类 /**  * 消息限流器  */ @AutoPipeline public interface MessageThrottler {       /**      * 节流单个消息      *      * @param messageThrottlerToken 消息限流令牌      * @return 是否被节流      */     boolean throttle(MessageThrottlerToken messageThrottlerToken);       /**      * 节流多个消息。任意一个消息被节流将返回true,否则返回false      *      * @param messageThrottlerTokens 多个消息限流令牌      * @return 是否被节流      */     boolean anyThrottle(List messageThrottlerTokens);       /**      * 节流多个消息。所有消息被节流才会返回true, 否则返回false      *      * @param messageThrottlerTokens 多个消息限流令牌      * @return 是否被节流      */     boolean allThrottle(List messageThrottlerTokens); }将限流规则拆分成三个不同的处理类 ClassificationThrottlerHandler   /**  * 消息分类节流器  *  * 
  * 目前仅针对营销消息进行节流  */ public class ClassificationThrottlerHandler implements MessageThrottlerHandler {       @Override     public boolean throttle(MessageThrottlerToken messageThrottlerToken, MessageThrottlerHandlerContext context) {         if (!ClassificationConstant.MARKETING.equals(messageThrottlerToken.getClassification())) {             return false;         }         return context.throttle(messageThrottlerToken);     }       @Override     public boolean anyThrottle(List messageThrottlerTokens, MessageThrottlerHandlerContext context) {         if (CollectionUtils.isEmpty(messageThrottlerTokens)) {             return false;         }           // 获取营销消息         List marketingMessageThrottlerTokens = messageThrottlerTokens.stream().filter(messageThrottlerToken -> {             return ClassificationConstant.MARKETING.equals(messageThrottlerToken.getClassification());         }).collect(Collectors.toList());           // 如果营销消息为空，说明消息均不需要被限流，直接返回false         if (CollectionUtils.isEmpty(marketingMessageThrottlerTokens)) {             return false;         }           return context.anyThrottle(marketingMessageThrottlerTokens);     }       @Override     public boolean allThrottle(List messageThrottlerTokens, MessageThrottlerHandlerContext context) {         if (CollectionUtils.isEmpty(messageThrottlerTokens)) {             return false;         }           // 获取营销消息         List 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  @Slf4j public class DuplicateThrottlerHandler implements MessageThrottlerHandler {       @Override     public boolean throttle(MessageThrottlerToken messageThrottlerToken, MessageThrottlerHandlerContext context) {         if (messageThrottlerToken.isThrottled()) {             return false;         }         boolean throttleResult = context.throttle(messageThrottlerToken);         messageThrottlerToken.markThrottled();         return throttleResult;     }       @Override     public boolean anyThrottle(List messageThrottlerTokens, MessageThrottlerHandlerContext context) {         if (CollectionUtils.isEmpty(messageThrottlerTokens)) {             return false;         }           // 过滤掉已经被限流的消息         List 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;     }       @Override     public boolean allThrottle(List messageThrottlerTokens, MessageThrottlerHandlerContext context) {         if (CollectionUtils.isEmpty(messageThrottlerTokens)) {             return false;         }           // 过滤掉已经被限流的消息         List 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＂);     @Autowired     private ThrottlerProxy throttlerProxy;       @Autowired     private ThrottlerModeConfiguration throttlerModeConfiguration;       private boolean throttle(AcquireToken acquireToken) {         // 获取限流模式         ThrottlerMode throttlerMode = throttlerModeConfiguration.getThrottlerMode(acquireToken.getAppKey(), acquireToken.getThrottleTag());         // 执行限流         return !throttlerProxy.tryAcquireWithAppKey(throttlerMode, acquireToken.getAppKey(), acquireToken.getPermits());     }       @Override     public 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;     }       @Override     public boolean anyThrottle(List messageThrottlerTokens, MessageThrottlerHandlerContext context) {         return throttle(messageThrottlerTokens, acquireTokens -> acquireTokens.stream().anyMatch(this::throttle)         );     }       @Override     public boolean allThrottle(List messageThrottlerTokens, MessageThrottlerHandlerContext messageThrottlerHandlerContext) {         return throttle(messageThrottlerTokens, acquireTokens -> acquireTokens.stream().allMatch(this::throttle)         );     }       private static boolean throttle(List messageThrottlerTokens, Function function) {         if (CollectionUtils.isEmpty(messageThrottlerTokens)) {             return false;         }           List 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 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     @AllArgsConstructor     private static class AcquireToken {         private final String throttleTag;         private final String appKey;         private final int permits;     } }消息限流代理 /**  * 消息限流代理  */ @Slf4j public class MessageThrottlerProxy {       @Autowired     private AcquireThrottlerHandler acquireThrottlerHandler;     private MessageThrottler messageThrottler;       @PostConstruct     public 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 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 {             @Override             public boolean throttle(MessageThrottler messageThrottler, List messageThrottlerTokens) {                 return messageThrottler.allThrottle(messageThrottlerTokens);             }         },         ANY {             @Override             public boolean throttle(MessageThrottler messageThrottler, List 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 messageThrottlerTokens);     } }
　　ps: 相关类并未全部列出，仅展示主要逻辑
　　原文链接：https://mp.weixin.qq.com/s/wR0NByDFWs5RDgWS2Jn6yA