Pipeline模式应用

2023-12-08 09:34 由京东云开发者发表于 #软件设计

本文记录Pipeline设计模式在业务流程编排中的应用

前言

Pipeline模式意为管道模式，又称为流水线模式。旨在通过预先设定好的一系列阶段来处理输入的数据，每个阶段的输出即是下一阶段的输入。

本案例通过定义PipelineProduct（管道产品），PipelineJob（管道任务），PipelineNode（管道节点），完成一整条流水线的组装，并将“原材料”加工为“商品”。其中管道产品负责承载各个阶段的产品信息；管道任务负责不同阶段对产品的加工；管道节点约束了管道产品及任务的关系，通过信号量定义了任务的执行方式。

依赖

工具依赖如下

            <!-- 工具类大全 -->
            <dependency>
                <groupId>cn.hutool</groupId>
                <artifactId>hutool-all</artifactId>
                <version>最新版本</version>
            </dependency>

编程示例

1. 管道产品定义

package com.example.demo.pipeline.model;

/**
 * 管道产品接口
 *
 * @param <S> 信号量
 * @author 
 * @date 2023/05/15 11:49
 */
public interface PipelineProduct<S> {
}

2. 管道任务定义

package com.example.demo.pipeline.model;

/**
 * 管道任务接口
 *
 * @param <P> 管道产品
 * @author 
 * @date 2023/05/15 11:52
 */
@FunctionalInterface
public interface PipelineJob<P> {
    /**
     * 执行任务
     *
     * @param product 管道产品
     * @return {@link P}
     */
    P execute(P product);
}

3. 管道节点定义

package com.jd.baoxian.mall.market.service.pipeline.model;

import java.util.function.Predicate;

/**
 * 管道节点定义
 *
 * @param <S> 信号量
 * @param <P> 管道产品
 * @author 
 * @date 2023/05/15 11:54
 */
public interface PipelineNode<S, P extends PipelineProduct<S>> {
    /**
     * 节点组装，按照上个管道任务传递的信号，执行 pipelineJob
     *
     * @param pipelineJob 管道任务
     * @return {@link PipelineNode}<{@link S},  {@link P}>
     */
    PipelineNode<S, P> flax(PipelineJob<P> pipelineJob);

    /**
     * 节点组装，按照传递的信号,判断当前管道的信号是否相等，执行 pipelineJob
     *
     * @param signal      信号
     * @param pipelineJob 管道任务
     * @return {@link PipelineNode}<{@link S},  {@link P}>
     */
    PipelineNode<S, P> flax(S signal, PipelineJob<P> pipelineJob);

    /**
     * 节点组装，按照传递的信号,判断当前管道的信号是否相等，执行 pipelineJob
     *
     * @param predicate   信号
     * @param pipelineJob 管道任务
     * @return {@link PipelineNode}<{@link S},  {@link P}>
     */
    PipelineNode<S, P> flax(Predicate<S> predicate, PipelineJob<P> pipelineJob);

    /**
     * 管道节点-任务执行
     *
     * @param product 管道产品
     * @return {@link P}
     */
    P execute(P product);
}

4. 管道产品、任务，节点的实现

4.1 管道产品

package com.example.demo.pipeline.factory;


import com.example.demo.model.request.DemoReq;
import com.example.demo.model.response.DemoResp;
import com.example.demo.pipeline.model.PipelineProduct;
import lombok.*;

/**
 * 样例-管道产品
 *
 * @author 
 * @date 2023/05/15 14:04
 */
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class DemoPipelineProduct implements PipelineProduct<DemoPipelineProduct.DemoSignalEnum> {
    /**
     * 信号量
     */
    private DemoSignalEnum signal;

    /**
     * 产品-入参及回参
     */
    private DemoProductData productData;

    /**
     * 异常信息
     */
    private Exception exception;

    /**
     * 流程Id
     */
    private String tradeId;

    @Data
    @Builder
    @NoArgsConstructor
    @AllArgsConstructor
    public static class DemoProductData {
        /**
         * 待验证入参
         */
        private DemoReq userRequestData;

        /**
         * 待验证回参
         */
        private DemoResp userResponseData;
    }

    /**
     * 产品-信号量
     *
     * @author 
     * @date 2023/05/15 13:54
     */
    @Getter
    public enum DemoSignalEnum {
        /**
         *
         */
        NORMAL(0, "正常"),
        /**
         *
         */
        CHECK_NOT_PASS(1, "校验不通过"),
        /**
         *
         */
        BUSINESS_ERROR(2, "业务异常"),
        /**
         *
         */
        LOCK_ERROR(3, "锁处理异常"),
        /**
         *
         */
        DB_ERROR(4, "事务处理异常"),

        ;
        /**
         * 枚举码值
         */
        private final int code;
        /**
         * 枚举描述
         */
        private final String desc;

        /**
         * 构造器
         *
         * @param code
         * @param desc
         */
        DemoSignalEnum(int code, String desc) {
            this.code = code;
            this.desc = desc;
        }
    }
}

4.2 管道任务（抽象类）

package com.example.demo.pipeline.factory.job;

import cn.hutool.core.util.ClassUtil;
import cn.hutool.json.JSONUtil;
import com.example.demo.pipeline.factory.DemoPipelineProduct;
import com.example.demo.pipeline.model.PipelineJob;
import lombok.extern.slf4j.Slf4j;

/**
 * 管道任务-抽象层
 *
 * @author 
 * @date 2023/05/15 19:48
 */
@Slf4j
public abstract class AbstractDemoJob implements PipelineJob<DemoPipelineProduct> {

    /**
     * 公共执行逻辑
     *
     * @param product 产品
     * @return
     */
    @Override
    public DemoPipelineProduct execute(DemoPipelineProduct product) {
        DemoPipelineProduct.DemoSignalEnum newSignal;
        try {
            newSignal = execute(product.getTradeId(), product.getProductData());
        } catch (Exception e) {
            product.setException(e);
            newSignal = DemoPipelineProduct.DemoSignalEnum.BUSINESS_ERROR;
        }
        product.setSignal(newSignal);
        defaultLogPrint(product.getTradeId(), product);
        return product;
    }

    /**
     * 子类执行逻辑
     *
     * @param tradeId     流程Id
     * @param productData 请求数据
     * @return
     * @throws Exception 异常
     */
    abstract DemoPipelineProduct.DemoSignalEnum execute(String tradeId, DemoPipelineProduct.DemoProductData productData) throws Exception;

    /**
     * 默认的日志打印
     */
    public void defaultLogPrint(String tradeId, DemoPipelineProduct product) {
        if (!DemoPipelineProduct.DemoSignalEnum.NORMAL.equals(product.getSignal())) {
            log.info("流水线任务处理异常：流程Id=【{}】,信号量=【{}】,任务=【{}】,参数=【{}】", tradeId, product.getSignal(),
                    ClassUtil.getClassName(this, true), JSONUtil.toJsonStr(product.getProductData()), product.getException());
        }
    }

}

4.3 管道节点

package com.example.demo.pipeline.factory;


import cn.hutool.core.util.ClassUtil;
import cn.hutool.json.JSONUtil;
import com.example.demo.pipeline.model.PipelineJob;
import com.example.demo.pipeline.model.PipelineNode;
import lombok.extern.slf4j.Slf4j;

import java.util.function.Predicate;

/**
 * 审核-管道节点
 *
 * @author 
 * @date 2023/05/15 14:32
 */
@Slf4j
public class DemoPipelineNode implements PipelineNode<DemoPipelineProduct.DemoSignalEnum, DemoPipelineProduct> {

    /**
     * 下一管道节点
     */
    private DemoPipelineNode next;

    /**
     * 当前管道任务
     */
    private PipelineJob<DemoPipelineProduct> job;

    /**
     * 节点组装，按照上个管道任务传递的信号，执行 pipelineJob
     *
     * @param pipelineJob 管道任务
     * @return {@link DemoPipelineNode}
     */
    @Override
    public DemoPipelineNode flax(PipelineJob<DemoPipelineProduct> pipelineJob) {
        return flax(DemoPipelineProduct.DemoSignalEnum.NORMAL, pipelineJob);
    }

    /**
     * 节点组装，按照传递的信号,判断当前管道的信号是否相等，执行 pipelineJob
     *
     * @param signal      信号
     * @param pipelineJob 管道任务
     * @return {@link DemoPipelineNode}
     */
    @Override
    public DemoPipelineNode flax(DemoPipelineProduct.DemoSignalEnum signal, PipelineJob<DemoPipelineProduct> pipelineJob) {
        return flax(signal::equals, pipelineJob);
    }

    /**
     * 节点组装，上个管道过来的信号运行 predicate 后是true的话，执行 pipelineJob
     *
     * @param predicate
     * @param pipelineJob
     * @return
     */
    @Override
    public DemoPipelineNode flax(Predicate<DemoPipelineProduct.DemoSignalEnum> predicate,
                                 PipelineJob<DemoPipelineProduct> pipelineJob) {
        this.next = new DemoPipelineNode();
        this.job = (job) -> {
            if (predicate.test(job.getSignal())) {
                return pipelineJob.execute(job);
            } else {
                return job;
            }
        };
        return next;
    }

    /**
     * 管道节点-任务执行
     *
     * @param product 管道产品
     * @return
     */
    @Override
    public DemoPipelineProduct execute(DemoPipelineProduct product) {
        // 执行当前任务
        try {
            product = job == null ? product : job.execute(product);
            return next == null ? product : next.execute(product);
        } catch (Exception e) {
            log.error("流水线处理异常：流程Id=【{}】,任务=【{}】,参数=【{}】", product.getTradeId(), ClassUtil.getClassName(job, true), JSONUtil.toJsonStr(product.getProductData()), product.getException());
            return null;
        }

    }
}

5. 业务实现

通过之前的定义，我们已经可以通过Pipeline完成流水线的搭建，接下来以“审核信息提交”这一业务场景，完成应用。

5.1 定义Api、入参、回参

package com.example.demo.api;

import com.example.demo.model.request.DemoReq;
import com.example.demo.model.response.DemoResp;
import com.example.demo.pipeline.factory.PipelineForManagerSubmit;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

/**
 * 演示-API
 *
 * @author 
 * @date 2023/08/06 16:27
 */
@Service
public class DemoManagerApi {

    /**
     * 管道-审核提交
     */
    @Resource
    private PipelineForManagerSubmit pipelineForManagerSubmit;

    /**
     * 审核提交
     *
     * @param requestData 请求数据
     * @return {@link DemoResp}
     */
    public DemoResp managerSubmit(DemoReq requestData) {
        return pipelineForManagerSubmit.managerSubmitCheck(requestData);
    }
}


package com.example.demo.model.request;

/**
 * 演示入参
 *
 * @author 
 * @date 2023/08/06 16:33
 */
public class DemoReq {
}


package com.example.demo.model.response;

import lombok.Data;

/**
 * 演示回参
 *
 * @author 
 * @date 2023/08/06 16:33
 */
@Data
public class DemoResp {
    /**
     * 成功标识
     */
    private Boolean success = false;

    /**
     * 结果信息
     */
    private String resultMsg;

    /**
     * 构造方法
     *
     * @param message 消息
     * @return {@link DemoResp}
     */
    public static DemoResp buildRes(String message) {
        DemoResp response = new DemoResp();
        response.setResultMsg(message);
        return response;
    }
}

5.2 定义具体任务

假定审核提交的流程需要包含：参数验证、加锁、解锁、事务提交

package com.example.demo.pipeline.factory.job;

import cn.hutool.json.JSONUtil;
import com.example.demo.model.request.DemoReq;
import com.example.demo.pipeline.factory.DemoPipelineProduct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

/**
 * 加锁-实现层
 *
 * @author 
 * @date 2023/05/17 17:00
 */
@Service
@Slf4j
public class CheckRequestLockJob extends AbstractDemoJob {

    /**
     * 子类执行逻辑
     *
     * @param tradeId     流程Id
     * @param productData 请求数据
     * @return
     * @throws Exception 异常
     */
    @Override
    DemoPipelineProduct.DemoSignalEnum execute(String tradeId, DemoPipelineProduct.DemoProductData productData) throws Exception {
        DemoReq userRequestData = productData.getUserRequestData();
        log.info("任务[{}]加锁,线程号:{}", JSONUtil.toJsonStr(userRequestData), tradeId);
        return DemoPipelineProduct.DemoSignalEnum.NORMAL;
    }
}


package com.example.demo.pipeline.factory.job;

import cn.hutool.json.JSONUtil;
import com.example.demo.model.request.DemoReq;
import com.example.demo.pipeline.factory.DemoPipelineProduct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

/**
 * 解锁-实现层
 *
 * @author 
 * @date 2023/05/17 17:00
 */
@Service
@Slf4j
public class CheckRequestUnLockJob extends AbstractDemoJob {

    /**
     * 子类执行逻辑
     *
     * @param tradeId     流程Id
     * @param productData 请求数据
     * @return
     * @throws Exception 异常
     */
    @Override
    DemoPipelineProduct.DemoSignalEnum execute(String tradeId, DemoPipelineProduct.DemoProductData productData) throws Exception {
        DemoReq userRequestData = productData.getUserRequestData();
        log.info("任务[{}]解锁,线程号:{}", JSONUtil.toJsonStr(userRequestData), tradeId);
        return DemoPipelineProduct.DemoSignalEnum.NORMAL;
    }
}

package com.example.demo.pipeline.factory.job;

import cn.hutool.json.JSONUtil;
import com.example.demo.model.request.DemoReq;
import com.example.demo.pipeline.factory.DemoPipelineProduct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;


/**
 * 审核-参数验证-实现类
 *
 * @author 
 * @date 2023/05/15 19:50
 */
@Slf4j
@Component
public class ManagerCheckParamJob extends AbstractDemoJob {

    /**
     * 执行基本入参验证
     *
     * @param tradeId
     * @param productData 请求数据
     * @return
     */
    @Override
    DemoPipelineProduct.DemoSignalEnum execute(String tradeId, DemoPipelineProduct.DemoProductData productData) {
        /*
         * 入参验证
         */
        DemoReq userRequestData = productData.getUserRequestData();
        log.info("任务[{}]入参验证,线程号:{}", JSONUtil.toJsonStr(userRequestData), tradeId);
        // 非空验证

        // 有效验证

        // 校验通过,退出
        return DemoPipelineProduct.DemoSignalEnum.NORMAL;
    }

}


package com.example.demo.pipeline.factory.job;

import cn.hutool.json.JSONUtil;
import com.example.demo.model.request.DemoReq;
import com.example.demo.model.response.DemoResp;
import com.example.demo.pipeline.factory.DemoPipelineProduct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

/**
 * 审核-信息提交-业务实现
 *
 * @author 
 * @date 2023/05/12 14:36
 */
@Service
@Slf4j
public class ManagerSubmitJob extends AbstractDemoJob {

    /**
     * 子类执行逻辑
     *
     * @param tradeId     流程Id
     * @param productData 请求数据
     * @return
     * @throws Exception 异常
     */
    @Override
    DemoPipelineProduct.DemoSignalEnum execute(String tradeId, DemoPipelineProduct.DemoProductData productData) throws Exception {
        DemoReq userRequestData = productData.getUserRequestData();
        try {
            /*
             * DB操作
             */
            log.info("任务[{}]信息提交,线程号:{}", JSONUtil.toJsonStr(userRequestData), tradeId);
            productData.setUserResponseData(DemoResp.buildRes("成功"));
        } catch (Exception ex) {
            log.error("审核-信息提交-DB操作失败,入参:{}", JSONUtil.toJsonStr(userRequestData), ex);
            throw ex;
        }
        return DemoPipelineProduct.DemoSignalEnum.NORMAL;
    }
}

5.3 完成流水线组装

针对入回参转换，管道任务执行顺序及执行信号量的构建

package com.example.demo.pipeline.factory;

import com.example.demo.model.request.DemoReq;
import com.example.demo.model.response.DemoResp;
import com.example.demo.pipeline.factory.job.CheckRequestLockJob;
import com.example.demo.pipeline.factory.job.CheckRequestUnLockJob;
import com.example.demo.pipeline.factory.job.ManagerCheckParamJob;
import com.example.demo.pipeline.factory.job.ManagerSubmitJob;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.Objects;
import java.util.UUID;

/**
 * 管道工厂入口-审核流水线
 *
 * @author 
 * @date 2023/05/15 19:52
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class PipelineForManagerSubmit {

    /**
     * 审核-管道节点
     */
    private final DemoPipelineNode managerSubmitNode = new DemoPipelineNode();


    /**
     * 审核-管道任务-提交-防刷锁-加锁
     */
    private final CheckRequestLockJob checkRequestLockJob;

    /**
     * 审核-管道任务-提交-防刷锁-解锁
     */
    private final CheckRequestUnLockJob checkRequestUnLockJob;

    /**
     * 审核-管道任务-参数验证
     */
    private final ManagerCheckParamJob managerCheckParamJob;

    /**
     * 审核-管道任务-事务操作
     */
    private final ManagerSubmitJob managerSubmitJob;


    /**
     * 组装审核的处理链
     */
    @PostConstruct
    private void assembly() {
        assemblyManagerSubmit();
    }

    /**
     * 组装处理链
     */
    private void assemblyManagerSubmit() {

        managerSubmitNode
                // 参数验证及填充
                .flax(managerCheckParamJob)
                // 防刷锁
                .flax(checkRequestLockJob)
                // 事务操作
                .flax(managerSubmitJob)
                // 锁释放
                .flax((ignore) -> true, checkRequestUnLockJob);
    }

    /**
     * 审核-提交处理
     *
     * @param requestData 入参
     * @return
     */
    public DemoResp managerSubmitCheck(DemoReq requestData) {
        DemoPipelineProduct initialProduct = managerSubmitCheckInitial(requestData);
        DemoPipelineProduct finalProduct = managerSubmitNode.execute(initialProduct);
        if (Objects.isNull(finalProduct) || Objects.nonNull(finalProduct.getException())) {
            return DemoResp.buildRes("未知异常");
        }
        return finalProduct.getProductData().getUserResponseData();
    }

    /**
     * 审核-初始化申请的流水线数据
     *
     * @param requestData 入参
     * @return 初始的流水线数据
     */
    private DemoPipelineProduct managerSubmitCheckInitial(DemoReq requestData) {
        // 初始化
        return DemoPipelineProduct.builder()
                .signal(DemoPipelineProduct.DemoSignalEnum.NORMAL)
                .tradeId(UUID.randomUUID().toString())
                .productData(DemoPipelineProduct.DemoProductData.builder().userRequestData(requestData).build())
                .build();
    }
}

总结

本文重点为管道模式的抽象与应用，上述示例仅为个人理解。实际应用中，此案例长于应对各种规则冗杂的业务场景，便于规则编排。
待改进点：

各个任务其实隐含了执行的先后顺序，此项内容可进一步实现；
针对最后“流水线组装”这一步，可通过配置描述的方式，进一步抽象，从而将变动控制在每个“管道任务”的描述上，针对规则项做到“可插拔”式处理。

作者：京东保险侯亚东

来源：京东云开发者社区转载请注明来源

设计模式（十六）迭代器

一、定义提供一种方法顺序访问一个聚合对象中的各个元素，且不用暴露该对象的内部表示。迭代器模式是一种对象行为型模式，又称为游标（Cursor）模式。二、描述在软件系统中，聚合对象拥有两个职责：一是存储数据，二是遍历数据。从依赖性来看，前者是聚合对象的基本职责，而后者既是可变化的，又是可分离的。因 ...阅读全文

Kafka集群调优+能力探底

一、前言我们需要对4个规格的kafka能力进行探底，即其可以承载的最大吞吐；4个规格对应的单节点的配置如下：标准版： 2C4G 铂金版： 4C8G 专业版： 8C16G 企业版： 16C32G 另外，一般来讲，在同配置下，kafka的读性能是要优于写性能的，写操作时，数据要从网卡拷贝至堆内存，然 ...阅读全文

Arcpy使用入门

2.7Python(目前ArcGIS使用)代码转化为3.5Python(目前ArcGIS Pro使用)代码 Analyze Tools For Pro (2to3命令) 基本操作调用ArcToolbox的两种形式 #arcpy.ToolboxAlias.ToolName() #arcpy.Tool ...阅读全文

DDD学习与感悟——总是觉得自己在CRUD怎么办？

我们有时候也会看到一些博客看到或者听到一些同事在说：这个业务有什么难的，不就是CRUD么？在软件生命周期初期，我们通过CRUD这种方式我们可以快速的实现业务规则，交付项目，但随着业务逐渐复杂，通过CRUD这种粗暴方式不可避免地会淹没业务核心规则，产生很多祖传（屎山）代码，系统交接的时候我们经常会听到... ...阅读全文

从物理机到K8S：应用系统部署方式的演进及其影响

公众号「架构成长指南」，专注于生产实践、云原生、分布式系统、大数据技术分享。概述随着科技的进步，软件系统的部署架构也在不断演进，从以前传统的物理机到虚拟机、Docker和Kubernetes，我们经历了一系列变化。这些技术的引入给我们带来了更高的资源利用率、更快的部署速度和更强大的扩展性，下面 ...阅读全文

Batrix企业能力库之物流交易域能力建设实践

Batrix企业能力库，是京东物流战略级项目-技术中台架构升级项目的基础底座。致力于建立企业级业务复用能力平台，依托能力复用业务框架Batrix，通过通用能力/扩展能力的定义及复用，灵活支持业务差异化场景的快速能力编排组装，从而通过技术驱动的方式助力业务整体交付吞吐率。本文重点为大家介绍交易域如何使... ...阅读全文

设计模式（十五）解释器

一、定义定义一个语言的文法，并且建立一个解释器来解释该语言中的句子，这里的“语言”是指使用规定格式和语法的代码。解释器模式是一种行为型模式。二、描述解释器模式是一种使用频率相对较低但学习难度较大的设计模式，它主要用于描述如何使用面向对象语言构成一个简单的语言解释器，包含以下四个角色： 1、Ab ...阅读全文

[编程]UML语言:理论之光与实践之惑

UML是软件工程的重要工具,但维护成本过高,导致实践中表现不佳。本文提出在人工智能支持下实现UML由“表达工具”向“认知工具”转型的路径,使之摆脱过高的人工学习和维护成本,重现辅助软件开发的效用。 ...阅读全文

设计规则：模块化的力量

这是一本比较冷门的书《设计规则：模块化的力量》，虽然豆瓣上只有58个评价，但是确实能学到很多东西。这本书对我非常深远。不是是投资，创业，还是其他领域，模块化思想都能帮上你。这本书告诉我们生万物的规则。书籍电子版PDF（建议及时保存，避免被和谐）：https://pan.quark.cn/s/aa ...阅读全文

写代码不用"if"行不行，曾经的反 if 运动

如果在IT行业的时间够长的话，可能还记得大约10几年前，设计模式风靡一时的时候，有过一段反 "if" 的运动。所谓的反"if"运动，其实是夸大了"if"语句带来的问题，比如当时提出的问题有：代码不好维护，特别是if或者else中的代码比较多的时候 if和 else if分支太多的时候，代码难以阅 ...阅读全文