登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 业界区 › 业界 ›Pipeline模式应用

Pipeline模式应用

都硎唷 2025-6-9 08:27:15
本文记录Pipeline设计模式在业务流程编排中的应用
前言

Pipeline模式意为管道模式，又称为流水线模式。旨在通过预先设定好的一系列阶段来处理输入的数据，每个阶段的输出即是下一阶段的输入。
本案例通过定义PipelineProduct（管道产品），PipelineJob（管道任务），PipelineNode（管道节点），完成一整条流水线的组装，并将“原材料”加工为“商品”。其中管道产品负责承载各个阶段的产品信息；管道任务负责不同阶段对产品的加工；管道节点约束了管道产品及任务的关系，通过信号量定义了任务的执行方式。
依赖

工具依赖如下
  1.             
  2.             <dependency>
  3.                 <groupId>cn.hutool</groupId>
  4.                 hutool-all</artifactId>
  5.                 <version>最新版本</version>
  6.             </dependency>
复制代码
编程示例

1. 管道产品定义
  1. package com.example.demo.pipeline.model;
  3. /**
  4. * 管道产品接口
  5. *
  6. * @param <S> 信号量
  7. * @author
  8. * @date 2023/05/15 11:49
  9. */
  10. public interface PipelineProduct<S> {
  11. }
复制代码
2. 管道任务定义
  1. package com.example.demo.pipeline.model;
  3. /**
  4. * 管道任务接口
  5. *
  6. * @param <P> 管道产品
  7. * @author
  8. * @date 2023/05/15 11:52
  9. */
  10. @FunctionalInterface
  11. public interface PipelineJob<P> {
  12.     /**
  13.      * 执行任务
  14.      *
  15.      * @param product 管道产品
  16.      * @return {@link P}
  17.      */
  18.     P execute(P product);
  19. }
复制代码
3. 管道节点定义
  1. package com.jd.baoxian.mall.market.service.pipeline.model;
  3. import java.util.function.Predicate;
  5. /**
  6. * 管道节点定义
  7. *
  8. * @param <S> 信号量
  9. * @param <P> 管道产品
  10. * @author
  11. * @date 2023/05/15 11:54
  12. */
  13. public interface PipelineNode<S, P extends PipelineProduct<S>> {
  14.     /**
  15.      * 节点组装，按照上个管道任务传递的信号，执行 pipelineJob
  16.      *
  17.      * @param pipelineJob 管道任务
  18.      * @return {@link PipelineNode}<{@link S},  {@link P}>
  19.      */
  20.     PipelineNode<S, P> flax(PipelineJob<P> pipelineJob);
  22.     /**
  23.      * 节点组装，按照传递的信号,判断当前管道的信号是否相等，执行 pipelineJob
  24.      *
  25.      * @param signal      信号
  26.      * @param pipelineJob 管道任务
  27.      * @return {@link PipelineNode}<{@link S},  {@link P}>
  28.      */
  29.     PipelineNode<S, P> flax(S signal, PipelineJob<P> pipelineJob);
  31.     /**
  32.      * 节点组装，按照传递的信号,判断当前管道的信号是否相等，执行 pipelineJob
  33.      *
  34.      * @param predicate   信号
  35.      * @param pipelineJob 管道任务
  36.      * @return {@link PipelineNode}<{@link S},  {@link P}>
  37.      */
  38.     PipelineNode<S, P> flax(Predicate<S> predicate, PipelineJob<P> pipelineJob);
  40.     /**
  41.      * 管道节点-任务执行
  42.      *
  43.      * @param product 管道产品
  44.      * @return {@link P}
  45.      */
  46.     P execute(P product);
  47. }
复制代码
4. 管道产品、任务，节点的实现

4.1 管道产品
  1. package com.example.demo.pipeline.factory;
  4. import com.example.demo.model.request.DemoReq;
  5. import com.example.demo.model.response.DemoResp;
  6. import com.example.demo.pipeline.model.PipelineProduct;
  7. import lombok.*;
  9. /**
  10. * 样例-管道产品
  11. *
  12. * @author
  13. * @date 2023/05/15 14:04
  14. */
  15. @Data
  16. @Builder
  17. @NoArgsConstructor
  18. @AllArgsConstructor
  19. public class DemoPipelineProduct implements PipelineProduct<DemoPipelineProduct.DemoSignalEnum> {
  20.     /**
  21.      * 信号量
  22.      */
  23.     private DemoSignalEnum signal;
  25.     /**
  26.      * 产品-入参及回参
  27.      */
  28.     private DemoProductData productData;
  30.     /**
  31.      * 异常信息
  32.      */
  33.     private Exception exception;
  35.     /**
  36.      * 流程Id
  37.      */
  38.     private String tradeId;
  40.     @Data
  41.     @Builder
  42.     @NoArgsConstructor
  43.     @AllArgsConstructor
  44.     public static class DemoProductData {
  45.         /**
  46.          * 待验证入参
  47.          */
  48.         private DemoReq userRequestData;
  50.         /**
  51.          * 待验证回参
  52.          */
  53.         private DemoResp userResponseData;
  54.     }
  56.     /**
  57.      * 产品-信号量
  58.      *
  59.      * @author
  60.      * @date 2023/05/15 13:54
  61.      */
  62.     @Getter
  63.     public enum DemoSignalEnum {
  64.         /**
  65.          *
  66.          */
  67.         NORMAL(0, "正常"),
  68.         /**
  69.          *
  70.          */
  71.         CHECK_NOT_PASS(1, "校验不通过"),
  72.         /**
  73.          *
  74.          */
  75.         BUSINESS_ERROR(2, "业务异常"),
  76.         /**
  77.          *
  78.          */
  79.         LOCK_ERROR(3, "锁处理异常"),
  80.         /**
  81.          *
  82.          */
  83.         DB_ERROR(4, "事务处理异常"),
  85.         ;
  86.         /**
  87.          * 枚举码值
  88.          */
  89.         private final int code;
  90.         /**
  91.          * 枚举描述
  92.          */
  93.         private final String desc;
  95.         /**
  96.          * 构造器
  97.          *
  98.          * @param code
  99.          * @param desc
  100.          */
  101.         DemoSignalEnum(int code, String desc) {
  102.             this.code = code;
  103.             this.desc = desc;
  104.         }
  105.     }
  106. }
复制代码
4.2 管道任务（抽象类）
  1. package com.example.demo.pipeline.factory.job;
  3. import cn.hutool.core.util.ClassUtil;
  4. import cn.hutool.json.JSONUtil;
  5. import com.example.demo.pipeline.factory.DemoPipelineProduct;
  6. import com.example.demo.pipeline.model.PipelineJob;
  7. import lombok.extern.slf4j.Slf4j;
  9. /**
  10. * 管道任务-抽象层
  11. *
  12. * @author
  13. * @date 2023/05/15 19:48
  14. */
  15. @Slf4j
  16. public abstract class AbstractDemoJob implements PipelineJob<DemoPipelineProduct> {
  18.     /**
  19.      * 公共执行逻辑
  20.      *
  21.      * @param product 产品
  22.      * @return
  23.      */
  24.     @Override
  25.     public DemoPipelineProduct execute(DemoPipelineProduct product) {
  26.         DemoPipelineProduct.DemoSignalEnum newSignal;
  27.         try {
  28.             newSignal = execute(product.getTradeId(), product.getProductData());
  29.         } catch (Exception e) {
  30.             product.setException(e);
  31.             newSignal = DemoPipelineProduct.DemoSignalEnum.BUSINESS_ERROR;
  32.         }
  33.         product.setSignal(newSignal);
  34.         defaultLogPrint(product.getTradeId(), product);
  35.         return product;
  36.     }
  38.     /**
  39.      * 子类执行逻辑
  40.      *
  41.      * @param tradeId     流程Id
  42.      * @param productData 请求数据
  43.      * @return
  44.      * @throws Exception 异常
  45.      */
  46.     abstract DemoPipelineProduct.DemoSignalEnum execute(String tradeId, DemoPipelineProduct.DemoProductData productData) throws Exception;
  48.     /**
  49.      * 默认的日志打印
  50.      */
  51.     public void defaultLogPrint(String tradeId, DemoPipelineProduct product) {
  52.         if (!DemoPipelineProduct.DemoSignalEnum.NORMAL.equals(product.getSignal())) {
  53.             log.info("流水线任务处理异常：流程Id=【{}】,信号量=【{}】,任务=【{}】,参数=【{}】", tradeId, product.getSignal(),
  54.                     ClassUtil.getClassName(this, true), JSONUtil.toJsonStr(product.getProductData()), product.getException());
  55.         }
  56.     }
  58. }
复制代码
4.3 管道节点
  1. package com.example.demo.pipeline.factory;
  4. import cn.hutool.core.util.ClassUtil;
  5. import cn.hutool.json.JSONUtil;
  6. import com.example.demo.pipeline.model.PipelineJob;
  7. import com.example.demo.pipeline.model.PipelineNode;
  8. import lombok.extern.slf4j.Slf4j;
  10. import java.util.function.Predicate;
  12. /**
  13. * 审核-管道节点
  14. *
  15. * @author
  16. * @date 2023/05/15 14:32
  17. */
  18. @Slf4j
  19. public class DemoPipelineNode implements PipelineNode<DemoPipelineProduct.DemoSignalEnum, DemoPipelineProduct> {
  21.     /**
  22.      * 下一管道节点
  23.      */
  24.     private DemoPipelineNode next;
  26.     /**
  27.      * 当前管道任务
  28.      */
  29.     private PipelineJob<DemoPipelineProduct> job;
  31.     /**
  32.      * 节点组装，按照上个管道任务传递的信号，执行 pipelineJob
  33.      *
  34.      * @param pipelineJob 管道任务
  35.      * @return {@link DemoPipelineNode}
  36.      */
  37.     @Override
  38.     public DemoPipelineNode flax(PipelineJob<DemoPipelineProduct> pipelineJob) {
  39.         return flax(DemoPipelineProduct.DemoSignalEnum.NORMAL, pipelineJob);
  40.     }
  42.     /**
  43.      * 节点组装，按照传递的信号,判断当前管道的信号是否相等，执行 pipelineJob
  44.      *
  45.      * @param signal      信号
  46.      * @param pipelineJob 管道任务
  47.      * @return {@link DemoPipelineNode}
  48.      */
  49.     @Override
  50.     public DemoPipelineNode flax(DemoPipelineProduct.DemoSignalEnum signal, PipelineJob<DemoPipelineProduct> pipelineJob) {
  51.         return flax(signal::equals, pipelineJob);
  52.     }
  54.     /**
  55.      * 节点组装，上个管道过来的信号运行 predicate 后是true的话，执行 pipelineJob
  56.      *
  57.      * @param predicate
  58.      * @param pipelineJob
  59.      * @return
  60.      */
  61.     @Override
  62.     public DemoPipelineNode flax(Predicate<DemoPipelineProduct.DemoSignalEnum> predicate,
  63.                                  PipelineJob<DemoPipelineProduct> pipelineJob) {
  64.         this.next = new DemoPipelineNode();
  65.         this.job = (job) -> {
  66.             if (predicate.test(job.getSignal())) {
  67.                 return pipelineJob.execute(job);
  68.             } else {
  69.                 return job;
  70.             }
  71.         };
  72.         return next;
  73.     }
  75.     /**
  76.      * 管道节点-任务执行
  77.      *
  78.      * @param product 管道产品
  79.      * @return
  80.      */
  81.     @Override
  82.     public DemoPipelineProduct execute(DemoPipelineProduct product) {
  83.         // 执行当前任务
  84.         try {
  85.             product = job == null ? product : job.execute(product);
  86.             return next == null ? product : next.execute(product);
  87.         } catch (Exception e) {
  88.             log.error("流水线处理异常：流程Id=【{}】,任务=【{}】,参数=【{}】", product.getTradeId(), ClassUtil.getClassName(job, true), JSONUtil.toJsonStr(product.getProductData()), product.getException());
  89.             return null;
  90.         }
  92.     }
  93. }
复制代码
5. 业务实现

通过之前的定义，我们已经可以通过Pipeline完成流水线的搭建，接下来以“审核信息提交”这一业务场景，完成应用。
5.1 定义Api、入参、回参
  1. package com.example.demo.api;
  3. import com.example.demo.model.request.DemoReq;
  4. import com.example.demo.model.response.DemoResp;
  5. import com.example.demo.pipeline.factory.PipelineForManagerSubmit;
  6. import org.springframework.stereotype.Service;
  8. import javax.annotation.Resource;
  10. /**
  11. * 演示-API
  12. *
  13. * @author
  14. * @date 2023/08/06 16:27
  15. */
  16. @Service
  17. public class DemoManagerApi {
  19.     /**
  20.      * 管道-审核提交
  21.      */
  22.     @Resource
  23.     private PipelineForManagerSubmit pipelineForManagerSubmit;
  25.     /**
  26.      * 审核提交
  27.      *
  28.      * @param requestData 请求数据
  29.      * @return {@link DemoResp}
  30.      */
  31.     public DemoResp managerSubmit(DemoReq requestData) {
  32.         return pipelineForManagerSubmit.managerSubmitCheck(requestData);
  33.     }
  34. }
  37. package com.example.demo.model.request;
  39. /**
  40. * 演示入参
  41. *
  42. * @author
  43. * @date 2023/08/06 16:33
  44. */
  45. public class DemoReq {
  46. }
  49. package com.example.demo.model.response;
  51. import lombok.Data;
  53. /**
  54. * 演示回参
  55. *
  56. * @author
  57. * @date 2023/08/06 16:33
  58. */
  59. @Data
  60. public class DemoResp {
  61.     /**
  62.      * 成功标识
  63.      */
  64.     private Boolean success = false;
  66.     /**
  67.      * 结果信息
  68.      */
  69.     private String resultMsg;
  71.     /**
  72.      * 构造方法
  73.      *
  74.      * @param message 消息
  75.      * @return {@link DemoResp}
  76.      */
  77.     public static DemoResp buildRes(String message) {
  78.         DemoResp response = new DemoResp();
  79.         response.setResultMsg(message);
  80.         return response;
  81.     }
  82. }
复制代码
5.2 定义具体任务

假定审核提交的流程需要包含：参数验证、加锁、解锁、事务提交
  1. package com.example.demo.pipeline.factory.job;
  3. import cn.hutool.json.JSONUtil;
  4. import com.example.demo.model.request.DemoReq;
  5. import com.example.demo.pipeline.factory.DemoPipelineProduct;
  6. import lombok.extern.slf4j.Slf4j;
  7. import org.springframework.stereotype.Service;
  9. /**
  10. * 加锁-实现层
  11. *
  12. * @author
  13. * @date 2023/05/17 17:00
  14. */
  15. @Service
  16. @Slf4j
  17. public class CheckRequestLockJob extends AbstractDemoJob {
  19.     /**
  20.      * 子类执行逻辑
  21.      *
  22.      * @param tradeId     流程Id
  23.      * @param productData 请求数据
  24.      * @return
  25.      * @throws Exception 异常
  26.      */
  27.     @Override
  28.     DemoPipelineProduct.DemoSignalEnum execute(String tradeId, DemoPipelineProduct.DemoProductData productData) throws Exception {
  29.         DemoReq userRequestData = productData.getUserRequestData();
  30.         log.info("任务[{}]加锁,线程号:{}", JSONUtil.toJsonStr(userRequestData), tradeId);
  31.         return DemoPipelineProduct.DemoSignalEnum.NORMAL;
  32.     }
  33. }
  36. package com.example.demo.pipeline.factory.job;
  38. import cn.hutool.json.JSONUtil;
  39. import com.example.demo.model.request.DemoReq;
  40. import com.example.demo.pipeline.factory.DemoPipelineProduct;
  41. import lombok.extern.slf4j.Slf4j;
  42. import org.springframework.stereotype.Service;
  44. /**
  45. * 解锁-实现层
  46. *
  47. * @author
  48. * @date 2023/05/17 17:00
  49. */
  50. @Service
  51. @Slf4j
  52. public class CheckRequestUnLockJob extends AbstractDemoJob {
  54.     /**
  55.      * 子类执行逻辑
  56.      *
  57.      * @param tradeId     流程Id
  58.      * @param productData 请求数据
  59.      * @return
  60.      * @throws Exception 异常
  61.      */
  62.     @Override
  63.     DemoPipelineProduct.DemoSignalEnum execute(String tradeId, DemoPipelineProduct.DemoProductData productData) throws Exception {
  64.         DemoReq userRequestData = productData.getUserRequestData();
  65.         log.info("任务[{}]解锁,线程号:{}", JSONUtil.toJsonStr(userRequestData), tradeId);
  66.         return DemoPipelineProduct.DemoSignalEnum.NORMAL;
  67.     }
  68. }
  70. package com.example.demo.pipeline.factory.job;
  72. import cn.hutool.json.JSONUtil;
  73. import com.example.demo.model.request.DemoReq;
  74. import com.example.demo.pipeline.factory.DemoPipelineProduct;
  75. import lombok.extern.slf4j.Slf4j;
  76. import org.springframework.stereotype.Component;
  79. /**
  80. * 审核-参数验证-实现类
  81. *
  82. * @author
  83. * @date 2023/05/15 19:50
  84. */
  85. @Slf4j
  86. @Component
  87. public class ManagerCheckParamJob extends AbstractDemoJob {
  89.     /**
  90.      * 执行基本入参验证
  91.      *
  92.      * @param tradeId
  93.      * @param productData 请求数据
  94.      * @return
  95.      */
  96.     @Override
  97.     DemoPipelineProduct.DemoSignalEnum execute(String tradeId, DemoPipelineProduct.DemoProductData productData) {
  98.         /*
  99.          * 入参验证
  100.          */
  101.         DemoReq userRequestData = productData.getUserRequestData();
  102.         log.info("任务[{}]入参验证,线程号:{}", JSONUtil.toJsonStr(userRequestData), tradeId);
  103.         // 非空验证
  105.         // 有效验证
  107.         // 校验通过,退出
  108.         return DemoPipelineProduct.DemoSignalEnum.NORMAL;
  109.     }
  111. }
  114. package com.example.demo.pipeline.factory.job;
  116. import cn.hutool.json.JSONUtil;
  117. import com.example.demo.model.request.DemoReq;
  118. import com.example.demo.model.response.DemoResp;
  119. import com.example.demo.pipeline.factory.DemoPipelineProduct;
  120. import lombok.extern.slf4j.Slf4j;
  121. import org.springframework.stereotype.Service;
  123. /**
  124. * 审核-信息提交-业务实现
  125. *
  126. * @author
  127. * @date 2023/05/12 14:36
  128. */
  129. @Service
  130. @Slf4j
  131. public class ManagerSubmitJob extends AbstractDemoJob {
  133.     /**
  134.      * 子类执行逻辑
  135.      *
  136.      * @param tradeId     流程Id
  137.      * @param productData 请求数据
  138.      * @return
  139.      * @throws Exception 异常
  140.      */
  141.     @Override
  142.     DemoPipelineProduct.DemoSignalEnum execute(String tradeId, DemoPipelineProduct.DemoProductData productData) throws Exception {
  143.         DemoReq userRequestData = productData.getUserRequestData();
  144.         try {
  145.             /*
  146.              * DB操作
  147.              */
  148.             log.info("任务[{}]信息提交,线程号:{}", JSONUtil.toJsonStr(userRequestData), tradeId);
  149.             productData.setUserResponseData(DemoResp.buildRes("成功"));
  150.         } catch (Exception ex) {
  151.             log.error("审核-信息提交-DB操作失败,入参:{}", JSONUtil.toJsonStr(userRequestData), ex);
  152.             throw ex;
  153.         }
  154.         return DemoPipelineProduct.DemoSignalEnum.NORMAL;
  155.     }
  156. }
复制代码
5.3 完成流水线组装

针对入回参转换，管道任务执行顺序及执行信号量的构建
  1. package com.example.demo.pipeline.factory;
  3. import com.example.demo.model.request.DemoReq;
  4. import com.example.demo.model.response.DemoResp;
  5. import com.example.demo.pipeline.factory.job.CheckRequestLockJob;
  6. import com.example.demo.pipeline.factory.job.CheckRequestUnLockJob;
  7. import com.example.demo.pipeline.factory.job.ManagerCheckParamJob;
  8. import com.example.demo.pipeline.factory.job.ManagerSubmitJob;
  9. import lombok.RequiredArgsConstructor;
  10. import lombok.extern.slf4j.Slf4j;
  11. import org.springframework.stereotype.Service;
  13. import javax.annotation.PostConstruct;
  14. import java.util.Objects;
  15. import java.util.UUID;
  17. /**
  18. * 管道工厂入口-审核流水线
  19. *
  20. * @author
  21. * @date 2023/05/15 19:52
  22. */
  23. @Slf4j
  24. @Service
  25. @RequiredArgsConstructor
  26. public class PipelineForManagerSubmit {
  28.     /**
  29.      * 审核-管道节点
  30.      */
  31.     private final DemoPipelineNode managerSubmitNode = new DemoPipelineNode();
  34.     /**
  35.      * 审核-管道任务-提交-防刷锁-加锁
  36.      */
  37.     private final CheckRequestLockJob checkRequestLockJob;
  39.     /**
  40.      * 审核-管道任务-提交-防刷锁-解锁
  41.      */
  42.     private final CheckRequestUnLockJob checkRequestUnLockJob;
  44.     /**
  45.      * 审核-管道任务-参数验证
  46.      */
  47.     private final ManagerCheckParamJob managerCheckParamJob;
  49.     /**
  50.      * 审核-管道任务-事务操作
  51.      */
  52.     private final ManagerSubmitJob managerSubmitJob;
  55.     /**
  56.      * 组装审核的处理链
  57.      */
  58.     @PostConstruct
  59.     private void assembly() {
  60.         assemblyManagerSubmit();
  61.     }
  63.     /**
  64.      * 组装处理链
  65.      */
  66.     private void assemblyManagerSubmit() {
  68.         managerSubmitNode
  69.                 // 参数验证及填充
  70.                 .flax(managerCheckParamJob)
  71.                 // 防刷锁
  72.                 .flax(checkRequestLockJob)
  73.                 // 事务操作
  74.                 .flax(managerSubmitJob)
  75.                 // 锁释放
  76.                 .flax((ignore) -> true, checkRequestUnLockJob);
  77.     }
  79.     /**
  80.      * 审核-提交处理
  81.      *
  82.      * @param requestData 入参
  83.      * @return
  84.      */
  85.     public DemoResp managerSubmitCheck(DemoReq requestData) {
  86.         DemoPipelineProduct initialProduct = managerSubmitCheckInitial(requestData);
  87.         DemoPipelineProduct finalProduct = managerSubmitNode.execute(initialProduct);
  88.         if (Objects.isNull(finalProduct) || Objects.nonNull(finalProduct.getException())) {
  89.             return DemoResp.buildRes("未知异常");
  90.         }
  91.         return finalProduct.getProductData().getUserResponseData();
  92.     }
  94.     /**
  95.      * 审核-初始化申请的流水线数据
  96.      *
  97.      * @param requestData 入参
  98.      * @return 初始的流水线数据
  99.      */
  100.     private DemoPipelineProduct managerSubmitCheckInitial(DemoReq requestData) {
  101.         // 初始化
  102.         return DemoPipelineProduct.builder()
  103.                 .signal(DemoPipelineProduct.DemoSignalEnum.NORMAL)
  104.                 .tradeId(UUID.randomUUID().toString())
  105.                 .productData(DemoPipelineProduct.DemoProductData.builder().userRequestData(requestData).build())
  106.                 .build();
  107.     }
  108. }
复制代码
总结

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

  • 各个任务其实隐含了执行的先后顺序，此项内容可进一步实现；
  • 针对最后“流水线组装”这一步，可通过配置描述的方式，进一步抽象，从而将变动控制在每个“管道任务”的描述上，针对规则项做到“可插拔”式处理。
作者：京东保险 侯亚东
来源：京东云开发者社区 转载请注明来源

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

使用道具 举报
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
都硎唷
2025-6-9 08:27:15

0

粉丝关注

9

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫