背景

在繁琐的业务流程处理中，通常采用面向过程的设计方法将流程拆分成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依赖

<dependency>

   <groupId>com.foldright.auto-pipeline</groupId>

   <artifactId>auto-pipeline-processor</artifactId>

   <version>0.2.0</version>

   <scope>provided</scope>

</dependency>

2. 在需要生成pipeline的接口上加上@AutoPipeline

只需为这个接口加上@AutoPipeline

@AutoPipeline

public interface ConfigSource {

   String get(String key);

}

3. 实现pipeline的handler

public class MapConfigSourceHandler implements ConfigSourceHandler {

   private final Map<String, String> map;

   public MapConfigSourceHandler(Map<String, String> 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<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 / 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#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

生成相关源代码，一个源文件采用一个特定的代码生成器生成，各个类的生成器继承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&实时类消息无需限流
• 如果一个消息已经被一个接口限流过，经过下一个接口时不应该被限流
• 对于同一个请求，只有整体限流和整体不限流 两种情况，不允许部分成功部分失败的情况(历史遗留问题)

面对这种的场景，该如何设计呢？

1. 首先是将限流规则拆分成三个步骤：消息分级处理、去重处理、请求限流令牌处理
2. 将整体限流和整体不限流抽象成合并策略，通过proxy的方式对外暴露

代码设计如下：

• 限流接口类

/**

* 消息限流器

*/

@AutoPipeline

public 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 {

   @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<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());

       // 如果营销消息为空，说明消息均不需要被限流，直接返回false

       if (CollectionUtils.isEmpty(marketingMessageThrottlerTokens)) {

           return false;

       }

       return context.anyThrottle(marketingMessageThrottlerTokens);

   }

   @Override

   public 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

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

   }

   @Override

   public 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");

   @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<MessageThrottlerToken> messageThrottlerTokens, MessageThrottlerHandlerContext context) {

       return throttle(messageThrottlerTokens, acquireTokens -> acquireTokens.stream().anyMatch(this::throttle)

       );

   }

   @Override

   public 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

   @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<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 {

           @Override

           public boolean throttle(MessageThrottler messageThrottler, List<MessageThrottlerToken> messageThrottlerTokens) {

               return messageThrottler.allThrottle(messageThrottlerTokens);

           }

       },

       ANY {

           @Override

           public 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);

   }

}

ps: 相关类并未全部列出，仅展示主要逻辑