⭐⭐⭐ Spring Boot 项目实战	⭐⭐⭐ Spring Cloud 项目实战
《Dubbo 实现原理与源码解析 —— 精品合集》	《Netty 实现原理与源码解析 —— 精品合集》
《Spring 实现原理与源码解析 —— 精品合集》	《MyBatis 实现原理与源码解析 —— 精品合集》
《Spring MVC 实现原理与源码解析 —— 精品合集》	《数据库实体设计合集》
《Spring Boot 实现原理与源码解析 —— 精品合集》	《Java 面试题 + Java 学习指南》

摘要: 原创出处 juejin.cn/post/7203993298817384508 「自由的_鱼」欢迎转载，保留摘要，谢谢！

🙂🙂🙂关注微信公众号：【芋道源码】有福利：

RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表

RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址

您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。

新的源码解析文章实时收到通知。每周更新一篇左右。

认真的源码交流微信群。

前言

在电商，保险，餐饮领域我们经常会遇到比较复杂的业务场景，比如在电商领域的下单业务中，需要根据用户的身份（是否会员），身份等级，商品属性，用户所处的位置属性等走不同的业务分支，不同业务分支的逻辑处理可能各不相同，每个分支的逻辑处理特别多且不固定，在后续产品完善，需求迭代过程中，需要频繁对各条业务线增减或修改。这时候如果我们只是在原有业务代码的基础上堆积木，业务类可能会越来越膨胀，越来越复杂，终成shi山，以至于后期难以维护，难以及时响应业务方需求，这是一个技术人所不能忍受的。在此分享一下如何使用Pipeline模式拆分，解耦这种复杂业务。

Spring源码中用到的责任链

Pipeline模式是责任链模式的一个变种，首先从责任链模式说起，看过Spring源码的同学都大概看到过在核心Http请求处理类DispatcherServlet中，在获取handler的时候，具体获取到的其实是一个被包装过的HandlerExecutionChain

HandlerExecutionChain中除了handler以外，还包含一堆实现了HandlerInterceptor接口的拦截器

里面有preHandle()、postHandle()、afterCompletion()三个方法，实现这三个方法可以分别在调用”Controller”方法之前，调用”Controller”方法之后渲染”ModelAndView”之前，以及渲染”ModelAndView”之后执行，每个方法都对应着HandlerExecutionChain中的applyxxx方法，真正执行时通过转化为数组用指针遍历执行。

此处的处理链类似于责任链。

在aop模块的ReflectiveMethodInvocation类中，执行增强逻辑时，也是通过递归调用形成责任链来执行。

public Object proceed() throws Throwable {
   // We start with an index of -1 and increment early.
   // 执行完所有增强后执行切点方法
   if (this.currentInterceptorIndex == this.interceptorsAndDynamicMethodMatchers.size() - 1) {
      return invokeJoinpoint();
   }

   // 获取下一个要执行的拦截器
   Object interceptorOrInterceptionAdvice =
         this.interceptorsAndDynamicMethodMatchers.get(++this.currentInterceptorIndex);
   if (interceptorOrInterceptionAdvice instanceof InterceptorAndDynamicMethodMatcher) {
      // Evaluate dynamic method matcher here: static part will already have
      // been evaluated and found to match.
      // 动态匹配
      InterceptorAndDynamicMethodMatcher dm =
            (InterceptorAndDynamicMethodMatcher) interceptorOrInterceptionAdvice;
      Class<?> targetClass = (this.targetClass != null ? this.targetClass : this.method.getDeclaringClass());
      if (dm.methodMatcher.matches(this.method, targetClass, this.arguments)) {
         return dm.interceptor.invoke(this);
      }
      else {
         // Dynamic matching failed.
         // Skip this interceptor and invoke the next in the chain.
         // 不匹配则不执行拦截器
         return proceed();
      }
   }
   else {
      // 普通拦截器，直接调用拦截器，比如 ExposeInvocationInterceptor,AspectJAfterAdvice
      // It's an interceptor, so we just invoke it: The pointcut will have
      // been evaluated statically before this object was constructed.
      // 传入this，形成调用链条，如MethodBeforeAdviceInterceptor，会先执行增强方法，再执行后面逻辑
      return ((MethodInterceptor) interceptorOrInterceptionAdvice).invoke(this);
   }
}

上面第一种是通过index下标去循环调用，无法终止处理，第二种通过递归，耦合比较严重，容易相互影响而无法调试，两者都不适用于业务编排。

Pipeline模式

上面我们说到Pipeline模式是责任链模式的一个变种，下面看看Pipeline的结构

Pipeline为链表结构，包含了头尾节点，当前上下文，开始-结束方法，对节点的操作方法

 public class DefaultOrderPipeLine implements OrderPipeLine {
//    public static TransmittableThreadLocal<StopWatch> pipeLineStopWatchTTL = new TransmittableThreadLocal<>();

    public OrderContext context;
    public OrderHandlerNode head;
    public OrderHandlerNode tail;

    public DefaultOrderPipeLine(OrderContext context) {
        this.context = context;
        head = new OrderHandlerNode();
        tail = head;
//      pipeLineStopWatchTTL.set(new StopWatch("DefaultCastPipeLine"));
    }
    
        @Override
    public void addFirst(OrderHandler... handlers) {
        OrderHandlerNode handlerNode;
        for (OrderHandler handler : handlers) {
            OrderHandlerNode preNode = head.getNextNode();
            handlerNode = new OrderHandlerNode(handler);
            handlerNode.setNextNode(preNode);
            head.setNextNode(handlerNode);
        }
    }

每个handler-node包含了对下一个节点的引用，部分代码如下

public class OrderHandlerNode {
    private OrderHandler handler;

    private OrderHandlerNode nextNode;

    public OrderHandlerNode(OrderHandler handler) {
        this.handler = handler;
    }
    
    public void execute(OrderContext) {
    }
   }

每个node包含了handler，各个handler之间通过统一上下文handlerContext串联通信

public class OrderContext {
    private String skuId;
    private String goodsName;
    private BigDecimal amount;
}

当触发pipeline的start方法时，handlerContext便会像一艘小船一样在水流中流动执行。上面举例我们只是创建了订单处理的pipeline，而实际真是业务场景在订单创建前可能需要作很多前置性校验，在订单创建后也可能需要作很多后续工作，如此我们可以起多条pipeline来分别处理不同业务。

tomcat和netty中用到的pipeline模式

tomcat中的核心模块也用到了这种模式，和上面我们说的这种很相似，它里面包含了4层管道，上层管道会调用下层管道，而每层管道都有一个特定的尾节点，每个尾部节点继承于ValveBase。

netty中的pipeline使用有些不太一样，他包含了出站事件，入站事件。

以上下文DefaultChannelHandlerContext为节点保存handler, 每个节点都有自己的执行器EventExecutor（其继承了java.util包的ScheduledExecutorService类），存储在DefaultChannelPipeline中，并且结构为双向链表，pipeline通过外界传入不同事件触发不同操作，如果是bind操作时, 通过尾节点tail的prev指针从后往前一个个执行，如果是fireChannelRegistered则通过头节点head的next指针从前往后执行。

我们实际使用的时候也可以借鉴这种结构，在追求性能场景下，可以一次性初始化好所有业务线的pipeline，后面根据参数来选择具体使用哪一种，在业务入口和出口都需要处理的情况下可以使用双向链表，在需要做不同业务线程池隔离或者执行方式隔离的情况下，给每个node添加一个executor。

关于此篇核心介绍的Pipeline完整模版的 demo 代码已经放到Github上，也是现在公司正在使用的模式，需要的同学可以前往https://github.com/jekran/order_pipeline 获取，拉下来即可测试，觉得有用的话，就给本文点个赞吧。

芋道源码 —— 纯源码解析博客

使用Pipeline模式解耦电商保险等复杂业务

前言

Spring源码中用到的责任链

Pipeline模式

tomcat和netty中用到的pipeline模式