目录
基础概念
管道模式（Pipeline Pattern） 是责任链模式（Chain of Responsibility Pattern）的常用变体之一。在管道模式中，管道扮演着流水线的角色，将数据传递到一个加工处理序列中，数据在每个步骤中被加工处理后，传递到下一个步骤进行加工处理，直到全部步骤处理完毕。
PS：纯的责任链模式在链上只会有一个处理器用于处理数据，而管道模式上多个处理器都会处理数据。
使用场景
任务代码较为复杂，满足以下业务场景下可以考虑使用管道模式。
- 添加新的子步骤
- 删除旧的子步骤
- 交换子步骤顺序
- 替换子步骤实现
场景描述
自定义生成优惠券
- 自定义生成前缀
定义通用管道上下文
@Getter
@Setter
public abstract class PipelineContext {
private LocalDateTime startTime;
private LocalDateTime endTime;
private String errorMsg;
public String getSimpleName() {
return this.getClass().getSimpleName();
}
}
定义管道上下文处理器
public interface ContextHandler<T extends PipelineContext> {
/**
* Handle context
*
* @param context context
* @return result
*/
boolean handle(T context);
}
定义业务上下文类型和所需处理器
业务上下文
@Data
@Builder
public class BatchCouponContext extends PipelineContext {
private Double lastIndex;
private Integer quantity;
private CouponBuildContext couponContext;
@Builder.Default
private List<String> couponCodes = new ArrayList<>();
}
处理器 - 校验
@Component
@Slf4j
public class BatchCouponValidator implements ContextHandler<BatchCouponContext> {
@Override
public boolean handle(BatchCouponContext context) {
log.info("[{}] Start validation", context.getSimpleName());
boolean condition = false;
if (condition) {
context.setErrorMsg("Validation failure");
return false;
}
return true;
}
}
处理器 - 生成模型实例
@Component
@Slf4j
public class BatchCouponCreator implements ContextHandler<BatchCouponContext> {
@Override
public boolean handle(BatchCouponContext context) {
log.info("[{}] Start to create, current dictionary {}", context.getSimpleName(), context.getCouponContext().getDictionary());
for (int i = 0; i < context.getQuantity(); i++) {
boolean result = doSomething();
if (!result) {
return false;
}
}
return true;
}
}
处理器 - 持久化
@Component
@Slf4j
public class BatchCouponSaver implements ContextHandler<BatchCouponContext> {
@Override
public boolean handle(BatchCouponContext context) {
log.info("[{}] Start saving, data: {}", context.getSimpleName(), context.getCouponCodes());
return true;
}
}
处理器 - 通知
@Slf4j
@Component
public class BatchCouponNotifier implements ContextHandler<BatchCouponContext> {
@Override
public boolean handle(BatchCouponContext context) {
log.info("[{}] Start notice, data: {}", context.getSimpleName(), context.getCouponCodes());
return true;
}
}
构建管道路由表
@Configuration
public class PipelineRouteConfig {
HashMap<Class<? extends PipelineContext>, List<Class<? extends ContextHandler<? extends PipelineContext>>>> maps = new HashMap<>();
@Bean("batchCouponExecutor")
public PipelineExecutor getBatchCouponPipelineExecutor() {
setBatchCouponPipeline();
return new PipelineExecutor(maps);
}
public void setBatchCouponPipeline() {
// key为上下文类型，value为处理器列表，按顺序执行
maps.put(BatchCouponContext.class,
Arrays.asList(
BatchCouponValidator.class,
BatchCouponCreator.class,
BatchCouponSaver.class,
BatchCouponNotifier.class
));
}
}
定义管道执行器
@Slf4j
public class PipelineExecutor implements ApplicationContextAware, InitializingBean {
// 配置表
private Map<Class<? extends PipelineContext>, List<Class<? extends ContextHandler<? extends PipelineContext>>>> contextMaps;
// 路由表
private Map<Class<? extends PipelineContext>, List<? extends ContextHandler<? super PipelineContext>>> routeMaps;
private ApplicationContext applicationContext;
public PipelineExecutor(Map<Class<? extends PipelineContext>, List<Class<? extends ContextHandler<? extends PipelineContext>>>> contextMaps) {
this.contextMaps = contextMaps;
}
private Map<Class<? extends PipelineContext>, List<? extends ContextHandler<? super PipelineContext>>> generateRouteMaps(Map<Class<? extends PipelineContext>, List<Class<? extends ContextHandler<? extends PipelineContext>>>> contextClassMaps) {
return contextClassMaps.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, this::getHandlerBean));
}
public List<? extends ContextHandler<? super PipelineContext>> getHandlerBean(Map.Entry<Class<? extends PipelineContext>, List<Class<? extends ContextHandler<? extends PipelineContext>>>> entry) {
return entry.getValue()
.stream()
.map(item -> (ContextHandler<PipelineContext>) applicationContext.getBean(item))
.collect(Collectors.toList());
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.applicationContext = applicationContext;
}
@Override
public void afterPropertiesSet() {
// 根据配置生成路由表
routeMaps = generateRouteMaps(contextMaps);
}
public boolean acceptSync(PipelineContext context) {
Assert.notNull(context, "Pipeline context is required");
// 获取处理类型
Class<? extends PipelineContext> contextType = context.getClass();
// 获取该类型所有管道
List<? extends ContextHandler<? super PipelineContext>> contextHandlers = routeMaps.get(contextType);
if (CollectionUtils.isEmpty(contextHandlers)) {
log.error("Pipeline {} is null ", contextType);
return false;
}
boolean lastSuccess = true;
for (ContextHandler<? super PipelineContext> handler : contextHandlers) {
try {
// 拿到当前管道处理结果
lastSuccess = handler.handle(context);
} catch (Throwable ex) {
lastSuccess = false;
log.error("[{}] occur error，handler={}", context.getSimpleName(), handler.getClass().getSimpleName(), ex);
}
if (!lastSuccess) {
break;
}
}
return lastSuccess;
}
}
测试
2022-12-01 23:55:06.053 INFO 7336 --- [nio-9090-exec-9] c.t.t.p.p.c.batch.BatchCouponValidator : [BatchCouponContext] Start validation
2022-12-01 23:55:06.053 INFO 7336 --- [nio-9090-exec-9] c.t.t.p.p.c.batch.BatchCouponCreator : [BatchCouponContext] Start to create, current dictionary CP05YVR8ZXG9FNMESWK71QLJUBHO4ID3A2T6
2022-12-01 23:55:06.054 INFO 7336 --- [nio-9090-exec-9] c.t.t.p.p.coupon.batch.BatchCouponSaver : [BatchCouponContext] Start saving, data: [PRE-CCC-CCP-SUF, PRE-CCC-CC0-SUF, PRE-CCC-CC5-SUF, PRE-CCC-CCY-SUF, PRE-CCC-CCV-SUF, PRE-CCC-CCR-SUF, PRE-CCC-CC8-SUF, PRE-CCC-CCZ-SUF, PRE-CCC-CCX-SUF, PRE-CCC-CCG-SUF]
2022-12-01 23:55:06.054 INFO 7336 --- [nio-9090-exec-9] c.t.t.p.p.c.batch.BatchCouponNotifier : [BatchCouponContext] Start notice, data: [PRE-CCC-CCP-SUF, PRE-CCC-CC0-SUF, PRE-CCC-CC5-SUF, PRE-CCC-CCY-SUF, PRE-CCC-CCV-SUF, PRE-CCC-CCR-SUF, PRE-CCC-CC8-SUF, PRE-CCC-CCZ-SUF, PRE-CCC-CCX-SUF, PRE-CCC-CCG-SUF]