阅读《2024 中国开源开发者报告》赢大奖，扫码申请享特权

管道（执行流）模型由 Pipeline（管道）/ Valve（阀门）/ Context（上下文） 组成

概念

我们把特定的业务，比如订单业务中的临时订单、订单提交以及订单支付等，抽象成一组Pipeline（管道）；

拿生成临时订单业务来说，执行流程包括：1参数校验->2业务数据校验->3业务处理，这里的三段子流程是严格按照顺序执行的，我们用Valve（阀门）定义它们，每一个子流程即一个Valve；

在管道模式中，我们要处理的对象是一组业务数据，即概念中的Context（上下文），Context贯穿于整个执行流程。

意义

管道模式是多步流程业务很好的抽象，内部基于单链表实现顺序执行，具有强顺序性；

管道模式对于整体流程的拆分，使得业务的扩展性大大增强，当业务需求发生变化，只需要确定需要加入/删除的子流程位置即可，就像从单链表中增加/删除一个节点。

实现

模块结构

PipeLineContext 实现管道上下文的概念，内部记录一个所处阀门在管道中的索引；使用HashMap存储业务数据，用户流程进行时的数据传递。

public class PipeLineContext {

    private PipeLineContext() {
    }

    @Getter
    private int index;

    @Getter
    private Map<String, Object> context;

    public PipeLineContext(int size) {
        this.index = 0;
        this.context = new HashMap<>(size);
    }

    public void put(String key, Object value) {
        context.put(key, value);
    }

    public void get(String key) {
        context.get(key);
    }

    @JSONField(serialize = false)
    public int getAndIncrement() {
        this.index++;
        return index;
    }

    @Override
    public String toString() {
        return "{\"index\":\"" + index + "\", \"context\":\"" + JSON.toJSONString(context) + "\"}";
    }

}

PipeLine 管道接口，包括添加阀门方法以及开启管道方法

public interface PipeLine {
    /**
     * 添加阀门
     * @param valve 阀门
     */
    void addValve(Valve valve);

    /**
     * 开启管道
     * @param pipeLineContext 管道上下文
     * @return FlowResult
     */
    FlowResult start(PipeLineContext pipeLineContext);
}

Valve 阀门接口，阀门都需实现该接口或者该接口的扩展接口

public interface Valve {
    /**
     * 获取下一个阀门
     * @return Valve 阀门
     */
    Valve getNext();

    /**
     * 设置下一个阀门
     * @param valve 阀门
     */
    void setNext(Valve valve);

    /**
     * 执行管道
     * @param pipeLineContext 管道上下文
     * @return FlowResult
     */
    FlowResult invoke(PipeLineContext pipeLineContext);
}

NormalPipeLine PipeLine接口通用实现

@Component
public class NormalPipeLine implements PipeLine {
    private Valve head = null;
    private Valve next = null;

    @Override
    public void addValve(Valve valve) {
        if (head == null) {
            head = valve;
            valve.setNext(next);
        } else {
            Valve current = head;
            while (current != null) {
                if (current.getNext() == next) {
                    current.setNext(valve);
                    valve.setNext(next);
                    break;
                }
                current = current.getNext();
            }
        }
    }

    @Override
    public FlowResult start(PipeLineContext pipeLineContext) {
        if (pipeLineContext == null) {
            return FlowResult.fail("pipeLineContext should be not null!");
        }
        if (head == null) {
            return FlowResult.fail("there's no valve in current pipeLine!");
        }
        return head.invoke(pipeLineContext);
    }
}

NormalValve Valve接口通用实现

public class NormalValve implements Valve {
    protected Valve next = null;

    @Override
    public Valve getNext() {
        return next;
    }

    @Override
    public void setNext(Valve valve) {
        this.next = valve;
    }

    @Override
    public FlowResult invoke(PipeLineContext pipeLineContext) {
        return processContinue(pipeLineContext);
    }

    protected FlowResult processContinue(PipeLineContext pipeLineContext) {
        return next == null ? FlowResult.ok() : getNext().invoke(pipeLineContext);
    }
}

Validator 订单-临时订单前置参数校验

@Slf4j
@Component
public class Validator extends NormalValve {
    @Override
    public FlowResult invoke(PipeLineContext pipeLineContext) {
        pipeLineContext.put("param", "1");
        return processContinue(pipeLineContext);
    }
}

使用AOP织入阀门，跟踪执行流

@Slf4j
@Aspect
@Component
public class PipeLineAspect {
    /**
     * 定义阀门invoke切点
     */
    @Pointcut(value = "this(com.nooice.order.common.pipeline.Valve) " +
            "&& execution(* invoke(com.nooice.order.common.pipeline.model.PipeLineContext)) && args((pipeLineContext))",
            argNames = "pipeLineContext")
    public void valveInvokeCutOffPoint(PipeLineContext pipeLineContext) {
    }

    @Before(value = "valveInvokeCutOffPoint(pipeLineContext)", argNames = "point,pipeLineContext")
    public void doBefore(JoinPoint point, PipeLineContext pipeLineContext) {
        int currentIndex = pipeLineContext.getAndIncrement();
        String className = point.getTarget().getClass().getName();
        log.info("管道前置通知-{}号阀门({})进入执行, pipeLineContext={}", currentIndex, className, pipeLineContext.toString());
    }
}

测试

@Slf4j
@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest(classes = Application.class)
public class PipelineTest {

	@Autowired
	private NormalPipeLine normalPipeLine;
	@Autowired
	private Validator validator;
	@Autowired
	private OrderPreviewValidator orderPreviewValidator;
	@Autowired
	private Processor processor;

	@Test
	public void testUserController() {
		// 定义上下文
		PipeLineContext pipeLineContext = new PipeLineContext(0);
		pipeLineContext.put("index", "0");
		// 增加阀门
		normalPipeLine.addValve(validator); // 参数校验阀门
		normalPipeLine.addValve(orderPreviewValidator); // 业务校验阀门
		normalPipeLine.addValve(processor); // 业务处理阀门
		// 管道执行
		FlowResult flowResult = normalPipeLine.start(pipeLineContext);
		log.info(JSON.toJSONString(flowResult));
	}

}

管道前置通知-1号阀门(com.nooice.order.common.pipeline.validator.Validator)进入执行, pipeLineContext=管道前置通知-2号阀门(com.nooice.order.common.pipeline.validator.OrderPreviewValidator)进入执行, pipeLineContext={"index":"2", "context":"{"index":"0","param":"1"}"}
管道前置通知-3号阀门(com.nooice.order.common.pipeline.processor.Processor)进入执行, pipeLineContext={"index":"3", "context":"{"index":"0","param":"2"}"}
PipelineTest: {"code":1,"message":"成功"}