背景
在繁琐的业务流程处理中,通常采用面向过程的设计方法将流程拆分成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
- 引入Maven依赖
<dependency>
<groupId>com.foldright.auto-pipeline</groupId>
<artifactId>auto-pipeline-processor</artifactId>
<version>0.2.0</version>
<scope>provided</scope>
</dependency>
- 在需要生成pipeline的接口上加上@AutoPipeline
只需为这个接口加上@AutoPipeline
@AutoPipeline
public 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;
}
@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);
}
}
- 使用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&实时类消息无需限流
- 如果一个消息已经被一个接口限流过,经过下一个接口时不应该被限流
- 对于同一个请求,只有整体限流和整体不限流 两种情况,不允许部分成功部分失败的情况(历史遗留问题)
面对这种的场景,该如何设计呢?
- 首先是将限流规则拆分成三个步骤:消息分级处理、去重处理、请求限流令牌处理
- 将整体限流和整体不限流抽象成合并策略,通过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: 相关类并未全部列出,仅展示主要逻辑