一个搞定责任链的注解

Posted 阿里巴巴淘系技术团队官网博客

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了一个搞定责任链的注解相关的知识，希望对你有一定的参考价值。

本文将介绍注解版的责任链模式框架-foldright/auto-pipeline，解决繁琐业务流程手动编写责任链的开发效率&复用等问题。

背景

在繁琐的业务流程处理中，通常采用面向过程的设计方法将流程拆分成N个步骤，每个步骤执行独立的逻辑。

public void process(params)
    doFirst(params);
    doSecond(params);
    ....
    doLast(params);

但是这样剥离仍然是不彻底的，修改其中一个步骤仍然可能影响其他步骤(同一个类修改，不符合开闭原则)。在这种场景下，有一种经典的设计模式-责任链模式，可以将这些子步骤封装成独立的handler，然后通过pipeline将其串联起来。 

常见的责任链模式会设计如下：

总体来看，纯手动编写有以下问题:

  1. 正确性：实现复杂度较高，短时间手工编写容易出错

  2. 开发效率：涉及多个类的实现，需要花费不少时间进行测试非业务的pipeline流程，ROI不高

  3. 复用性：不同业务流程难以复用同一套pipeline的关键代码

那有没有一套靠谱的框架能够解决上述问题呢？有的，它就是foldright/auto-pipeline，是责任链领域的"lombok"!

Quirk Start

下面以读取系统配置为例，读取逻辑如下：

  1. 从本地配置文件读取，读取成功则直接返回，否则执行下一步

  2. 从系统变量读取，返回对应的值

为了实现这个需求，读取配置接口定义如下：

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>

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

只需为这个接口加上@AutoPipeline

@AutoPipeline
public interface ConfigSource 
    String get(String key);

  1. 实现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);

  1. 使用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源码

场景实战

下面举一个项目中真实的例子-消息分级限流。

消息发送的流量现状：

  1. 同一个请求可能包含有多个AppKey的消息

  2. 同一个请求可能包含多个消息分级的消息

  3. 同一个请求的消息可能经过多个接口

  4. 每个消息都会有对应的Appkey、消息分级

限流规则如下：

  1. 需要对消息所属的AppKey进行单独限流

  2. 仅对营销类消息进行限流，IM&实时类消息无需限流

  3. 如果一个消息已经被一个接口限流过，经过下一个接口时不应该被限流

  4. 对于同一个请求，只有整体限流和整体不限流 两种情况，不允许部分成功部分失败的情况(历史遗留问题)

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

  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: 相关类并未全部列出，仅展示主要逻辑

团队介绍

欢迎加入淘宝移动技术中台，团队成员大牛云集，有阿里移动中间件的创始人员、鹰眼全链路追踪平台核心成员、更有一群热爱技术，期望用技术推动业务的小伙伴。

淘宝移动技术中台，推进大淘宝技术（淘宝、天猫等）架构升级，致力于为淘系、整个集团提供基础核心能力、产品与解决方案：

  1. 业务高可用的解决方案与核心能力（应用高可用：为业务提供自适应限流、隔离与熔断的柔性高可用解决方案，站点高可用：故障自愈、多机房与异地容灾与快速切流恢复）；

  2. 新一代的业务研发模式FaaS（一站式函数研发Gaia平台）；

  3. 下一代网络协议QUIC实现与落地；

  4. 移动中间件（API网关MTop、域名调度AMDC、消息/推送、文件上传AUS、移动配置推送Orange 等等）。

✿  拓展阅读

作者|余敦刚（玄苏）

编辑|橙子君

以上是关于一个搞定责任链的注解的主要内容，如果未能解决你的问题，请参考以下文章

基于双向责任链的接口设计

基于双向责任链的接口设计

模拟javaWeb责任链的设计

5 行为型模式之- 责任链模式

一个注解搞定 Spring Security 忽略拦截

牛X！一个注解搞定接口防刷！