开发者社区> 游客o6py746u4tivu> 正文
阿里云
为了无法计算的价值
打开APP
阿里云APP内打开

一个注解搞定责任链,学还是不学?

简介: 责任链作为一种常见的设计模式,是业务流程解耦利器。但是纯手工手编写一个责任链耗时又费力, 有没有快捷方式,就像lombok一样简单?有的,它就是foldright/auto-pipeline,一个注解轻松搞定!
+关注继续查看

背景

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

public void process(params){

    doFirst(params);

    doSecond(params);

    ....

    doLast(params);

}

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

image


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

image

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

  • 正确性:实现复杂度较高,短时间手工编写容易出错
  • 开发效率:涉及多个类的实现,需要花费不少时间进行测试非业务的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>

  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)返回仍然为空,则直接返回空。

image

以获取配置为例:

image

  • 用户实现
    • 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对象的默认实现类

源码解读

目录

image

  • auto-pipeline-annotations
    • 框架包含的注解:AutoPipeline、PipelineDirection

image

      • AutoPipeline
        • 生成pipeline的核心注解
      • PipelineDirection
        • pipeline处理的顺序方向
  • auto-pipeline-processor

image

    • 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

image

生成相关源代码,一个源文件采用一个特定的代码生成器生成,各个类的生成器继承AbstractGenerator

image

源代码生成类: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源码

image

场景实战

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

消息发送的流量现状:

  • 同一个请求可能包含有多个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: 相关类并未全部列出,仅展示主要逻辑

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
一起学习设计模式:责任链模式
每个公司可能都存在自己的采购审批模式,那么如何在软件中设计这样一个采购单的分级审批?
22 0
iOS视频镜像处理
在网上找了很多资料和dome都没有处理镜像的,于是就自己研究了下。在官方说明文档中找到了
27 0
巧妙的运用责任链模式,让你的代码高出一个逼格!
什么是责任链模式?(Chain of Responsibility Pattern),简单的说,为请求者和接受者之间创建一条对象处理链路,避免请求发送者与接收者耦合在一起!
67 0
某学院最全的SSM框架全套学习视频
某学院最全的SSM框架全套学习视频
56 0
【设计模式】责任链模式 ( 简介 | 适用场景 | 优缺点 | 代码示例 )
【设计模式】责任链模式 ( 简介 | 适用场景 | 优缺点 | 代码示例 )
407 0
给一个网页指定编码
给一个网页指定编码
15 0
【Ubuntu使用技巧】在Ubuntu下制作ISO镜像的方法
作者:gnuhpc 出处:http://www.cnblogs.com/gnuhpc/ 1 如果你是直接从cd 压制iso 文件的,执行 sudo umount /dev/cdrom dd if=/dev/cdrom of=file.iso bs=1024 2 如果你要把某个文件或者目录压到iso 文件中,需要使用mkisofs 这个工具。
749 0
文章
问答
文章排行榜
最热
最新
相关电子书
更多
基于云等保的安全责任分担模型及解读
立即下载
低代码开发师(初级)实战教程
立即下载
阿里巴巴DevOps 最佳实践手册
立即下载