java pipeline

<link rel="stylesheet" href="https://js.how234.com/third-party/SyntaxHighlighter/shCoreDefault.css" type="text/css" /><script type="text/javascript" src="https://js.how234.com/third-party/SyntaxHighlighter/shCore.js"></script><script type="text/javascript"> SyntaxHighlighter.all(); </script>

java pipeline是什么，让我们一起了解一下？

pipeline又称为管道，是一种在计算机普遍使用的技术。在分布式处理领域，由于管道模式是数据驱动，而目前流行的Spark分布式处理平台也是数据驱动的，两者非常合拍，于是在spark的新的api里面pipeline模式得到了广泛的应用。还有java web中的struct的filter、netty的pipeline，无处不见pipeline模式。

管道模式设计其实和责任链模式很像，都是按照顺序往下执行不同的方法，管道只是负责顺序执行，不管是否执行不同方法。

管道入口：

Map context = new HashMap<>();                context.put("BusinessType", BusinessType.CREDIT_FLOW.getBusinessType());                context.put("CheckType", CertificateBusinessTypeEnum.THREE_ELEMENTS.name());                context.put("ZaUser", user);                context.put("RequestParam", relavants);                //管道入口                context = certificateElementService.checkAdmittance(context);                JSONObject jsonObject = (JSONObject) context.get("ResponseParam");

AaaThreeElementsCheckValve 的实现：

@Component("aaaThreeElementsCheckValve")public class AaaThreeElementsCheckValve extends AbstractLogableValve<Map, I18NSupportException> {     @Autowired    private ICodeLibraryService codeLibraryService;    @Autowired    private IJunYuCertificateService junYuCertificateService;     @Override    public void handle(Map context, ValveChain<Map, I18NSupportException> chain) throws I18NSupportException {        String checkType = context.get("CheckType").toString();        //aaa的校验不通过,直接执行下一个        if (!Objects.equals(CertificateBusinessTypeEnum.THREE_ELEMENTS.name(), checkType)) {            chain.handleNext(context);        }        //执行aaa的代码逻辑         if (Objects.equals(CertificateResponseLevelEnum.FORBIDDEN.getCode(), finalLevel)) {            //aaa处理了请求,返回                return;          }            //放开,让bbb去处理            chain.handleNext(context);        } catch (Exception e) {            logger.error("骏聿三要素接口异常：", e);            I18NSupportException.Builder builder = new I18NSupportException.Builder();            builder.setTargetObject(this);            builder.setMessage(e.getMessage());            throw builder.build();        }    }}

BbbThreeElementsCheckValve 的实现：

@Component("bbbThreeElementsCheckValve")public class BbbThreeElementsCheckValve extends AbstractLogableValve<Map, I18NSupportException> {     @Autowired    private ICodeLibraryService codeLibraryService;    @Autowired    private IShuJuBaoCreditService shuJuBaoCreditService;     @Override    public void handle(Map context, ValveChain<Map, I18NSupportException> chain) throws I18NSupportException {         JSONObject jsonObject = (JSONObject) context.get("ResponseParam");        if (Objects.equals(Integer.valueOf(jsonObject.get(WebUtil.JSON_RESULT_STATUS_CODE).toString()), WebUtil.ERROR)) {            //aaa三要素接口已强控，因此不继续调用bbb三要素接口,也不调用后续阀门进行校验        } else {            ZaUser user = (ZaUser) context.get("ZaUser");            Object requestParam = context.get("RequestParam");            List relavants = JSON.parseArray(JSON.toJSONString(requestParam), CustomerCreditRelavant.class);            CodeLibrary codeLibrary = codeLibraryService.queryLibraryNoException("shujubaoCompanyIdAndCobankId", user.getCompanyId() + "-" + relavants.get(0).getInquryBankId());            if (Objects.nonNull(codeLibrary)) {                String s = shuJuBaoCreditService.sjbThreeElmentVerify(relavants);                if (StringUtils.isNotBlank(s)) {                    jsonObject.put(WebUtil.JSON_RESULT_STATUS_CODE, WebUtil.ERROR);                    jsonObject.put(WebUtil.JSON_RESULT_DATA, JSONObject.toJSONString(s));                }            }else{                //bbb三要素校验通过，调用后续阀门进行校验                chain.handleNext(context);            }        }    }}