聊聊如何利用管道模式來進行業務編排（下篇）
前言
上篇文章我們介紹 利用管道模式來進行業務編排 的2種實現方式。本文又來介紹其他實現方式
實現方式
方式一：利用springboot自動裝配
1、新建管道實體
@Data
@AllArgsConstructor
@NoArgsConstructor
public class PipelineDefinition {
public static final String PREFIX = "lybgeek_pipeline_";
private String comsumePipelineName;
private List<String> pipelineClassNames;
}
@Data
@AllArgsConstructor
@NoArgsConstructor
@ConfigurationProperties(prefix = PipelineDefinitionProperties.PREFIX)
public class PipelineDefinitionProperties {
public final static String PREFIX = "lybgeek.pipeline";
private List<PipelineDefinition> chain;
}
2、編寫自動裝配類
@Configuration
@EnableConfigurationProperties(PipelineDefinitionProperties.class)
public class PipelineAutoConfiguration implements BeanFactoryAware,InitializingBean, SmartInitializingSingleton {
@Autowired
private PipelineDefinitionProperties pipelineDefinitionProperties;
private DefaultListableBeanFactory defaultListableBeanFactory;
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
defaultListableBeanFactory = (DefaultListableBeanFactory)beanFactory;
}
private void registerPipeline(DefaultListableBeanFactory defaultListableBeanFactory, PipelineDefinition pipelineDefinition) {
LinkedBlockingDeque linkedBlockingDeque = buildPipelineQuque(pipelineDefinition);
GenericBeanDefinition beanDefinition = (GenericBeanDefinition) BeanDefinitionBuilder.genericBeanDefinition(ChannelPipeline.class).getBeanDefinition();
beanDefinition.getPropertyValues().addPropertyValue("channelHandlers",linkedBlockingDeque);
defaultListableBeanFactory.registerBeanDefinition(PipelineDefinition.PREFIX + pipelineDefinition.getComsumePipelineName() ,beanDefinition);
}
@SneakyThrows
private LinkedBlockingDeque buildPipelineQuque(PipelineDefinition pipelineDefinition) {
List<String> pipelineClassNames = pipelineDefinition.getPipelineClassNames();
if(CollectionUtil.isEmpty(pipelineClassNames)){
throw new PipelineException("pipelineClassNames must config");
}
LinkedBlockingDeque linkedBlockingDeque = new LinkedBlockingDeque();
for (String pipelineClassName : pipelineClassNames) {
Class<?> pipelineClassClass = Class.forName(pipelineClassName);
if(!AbstactChannelHandler.class.isAssignableFrom(pipelineClassClass)){
throw new PipelineException("pipelineClassNames must be 【com.github.lybgeek.pipeline.handler.AbstactChannelHandler】 subclass");
}
Object pipeline = pipelineClassClass.getDeclaredConstructor().newInstance();
linkedBlockingDeque.addLast(pipeline);
}
return linkedBlockingDeque;
}
@Override
public void afterPropertiesSet() throws Exception {
if(CollectionUtil.isNotEmpty(pipelineDefinitionProperties.getChain())){
for (PipelineDefinition pipelineDefinition : pipelineDefinitionProperties.getChain()) {
registerPipeline(defaultListableBeanFactory, pipelineDefinition);
}
}
}
@Override
public void afterSingletonsInstantiated() {
Map<String, ChannelPipeline> pipelineBeanMap = defaultListableBeanFactory.getBeansOfType(ChannelPipeline.class);
pipelineBeanMap.forEach((key,bean)->{
bean.setHandlerContext(ChannelHandlerContext.getCurrentContext());
});
}
}
3、編寫spring.factories
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ com.github.lybgeek.pipeline.spring.autoconfigure.PipelineAutoConfiguration\
業務專案如何使用該方式實現業務編排
示例：
1、建立管道執行器
@Slf4j
public class UserCheckChannelHandler extends AbstactChannelHandler {
@Override
public boolean handler(ChannelHandlerContext chx) {
ChannelHandlerRequest channelHandlerRequest = chx.getChannelHandlerRequest();
System.out.println("yml------------------------------------步驟一：使用者資料校驗【"+channelHandlerRequest.getRequestId()+"】");
Object params = channelHandlerRequest.getParams();
if(params instanceof User){
User user = (User)params;
if(StringUtils.isBlank(user.getFullname())){
log.error("使用者名稱不能為空");
return false;
}
return true;
}
return false;
}
}
@Slf4j
public class UserFillUsernameAndEmailChannelHandler extends AbstactChannelHandler {
@SneakyThrows
@Override
public boolean handler(ChannelHandlerContext chx) {
ChannelHandlerRequest channelHandlerRequest = chx.getChannelHandlerRequest();
System.out.println("yml------------------------------------步驟二：使用者名稱以及郵箱填充【將漢語轉成拼音填充】【"+channelHandlerRequest.getRequestId()+"】");
Object params = channelHandlerRequest.getParams();
if(params instanceof User){
User user = (User)params;
String fullname = user.getFullname();
HanyuPinyinOutputFormat hanyuPinyinOutputFormat = new HanyuPinyinOutputFormat();
hanyuPinyinOutputFormat.setToneType(HanyuPinyinToneType.WITHOUT_TONE);
String username = PinyinHelper.toHanYuPinyinString(fullname, hanyuPinyinOutputFormat);
user.setUsername(username);
user.setEmail(username + "@qq.com");
return true;
}
return false;
}
}
。。。其他執行器具體檢視連結程式碼
2、配置yml檔案
lybgeek:
pipeline:
chain:
- comsumePipelineName: userYmlService
pipelineClassNames:
- com.github.lybgeek.pipeline.spring.test.yml.handler.UserCheckChannelHandler
- com.github.lybgeek.pipeline.spring.test.yml.handler.UserFillUsernameAndEmailChannelHandler
- com.github.lybgeek.pipeline.spring.test.yml.handler.UserPwdEncryptChannelHandler
- com.github.lybgeek.pipeline.spring.test.yml.handler.UserMockSaveChannelHandler
- com.github.lybgeek.pipeline.spring.test.yml.handler.UserPrintChannleHandler
3、具體業務service引入管道bean
@Service
public class UserYmlServiceImpl implements UserYmlService {
@Autowired
private ApplicationContext applicationContext;
@Override
public boolean save(User user) {
ChannelPipeline pipeline = applicationContext.getBean(ChannelPipeline.class,PipelineDefinition.PREFIX + StringUtils.uncapitalize(UserYmlService.class.getSimpleName()));
return pipeline.start(ChannelHandlerRequest.builder().params(user).build());
}
}
4、測試
@Test
public void testPipelineYml(){
boolean isOk = userYmlService.save(user);
Assert.assertTrue(isOk);
}
方式二：利用spring自定義標籤
1、定義xsd約束檔案pipeline.xsd
<?xml version="1.0" encoding="UTF-8" standalone="no"?>
<xsd:schema xmlns:xsd="http://www.w3.org/2001/XMLSchema"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:tool="http://www.springframework.org/schema/tool"
xmlns="http://lybgeek.github.com/schema/pipeline"
targetNamespace="http://lybgeek.github.com/schema/pipeline">
<xsd:import namespace="http://www.w3.org/XML/1998/namespace"/>
<xsd:import namespace="http://www.springframework.org/schema/beans"
schemaLocation="http://www.springframework.org/schema/beans/spring-beans.xsd"/>
<xsd:import namespace="http://www.springframework.org/schema/tool"/>
<xsd:annotation>
<xsd:documentation>
<![CDATA[ Namespace support for pipeline services ]]></xsd:documentation>
</xsd:annotation>
<xsd:complexType name="pipelineType">
<xsd:choice>
<xsd:element ref="pipelineHandler" minOccurs="1" maxOccurs="unbounded"/>
</xsd:choice>
<xsd:attribute name="id" type="xsd:ID">
<xsd:annotation>
<xsd:documentation><![CDATA[ The unique identifier for a bean. ]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="consumePipelinesServiceClassName" type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation><![CDATA[ consumePipelinesService class name ]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="consumePipelinesMethod" type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation><![CDATA[ consumePipelinesMethod name ]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="argsType" type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation><![CDATA[ consumePipelinesMethod args type , multiple args types are separated by commas ]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
<xsd:complexType name="pipelineHandlerType">
<xsd:attribute name="className" type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation><![CDATA[ pipelineHanlder class name]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="order" type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation><![CDATA[ pipeline class name]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
<xsd:element name="pipelineHandler" type="pipelineHandlerType">
<xsd:annotation>
<xsd:documentation><![CDATA[ The pipelineHandler config ]]></xsd:documentation>
</xsd:annotation>
</xsd:element>
<xsd:element name="pipeline" type="pipelineType">
<xsd:annotation>
<xsd:documentation><![CDATA[ The pipeline config ]]></xsd:documentation>
</xsd:annotation>
</xsd:element>
</xsd:schema>
2、配置xsd約束檔案
在classpath下的resources資料夾新建META-INF資料夾，再建立一個檔案spring.schemas，內容如下
http\://lybgeek.github.com/schema/pipeline/pipeline.xsd=META-INF/pipeline.xsd
3、定義解析自定義標籤的類
public class PipelineNamespaceHandler extends NamespaceHandlerSupport {
@Override
public void init() {
registerBeanDefinitionParser("pipeline",new PipelineBeanDefinitionParser());
}
}
public class PipelineBeanDefinitionParser implements BeanDefinitionParser {
@Override
public BeanDefinition parse(Element element, ParserContext parserContext) {
PipelineConfig pipelineConfig = buildPipelineConfig(element);
List<HandlerInvotation> handlerInvotations = this.buildHandlerInvotations(pipelineConfig);
GenericBeanDefinition beanDefinition = getGenericBeanDefinition(element, parserContext, pipelineConfig, handlerInvotations);
return beanDefinition;
}
private GenericBeanDefinition getGenericBeanDefinition(Element element, ParserContext parserContext, PipelineConfig pipelineConfig, List<HandlerInvotation> handlerInvotations) {
GenericBeanDefinition beanDefinition = (GenericBeanDefinition) BeanDefinitionBuilder.genericBeanDefinition().getBeanDefinition();
beanDefinition.getPropertyValues().addPropertyValue("pipelineServiceClz",pipelineConfig.getConsumePipelinesService());
beanDefinition.getPropertyValues().addPropertyValue("handlerInvotations",handlerInvotations);
beanDefinition.getPropertyValues().addPropertyValue("createByXml",true);
beanDefinition.setBeanClass(ComsumePipelineFactoryBean.class);
String beanName = BeanUtils.generateBeanName(element,"id",parserContext,pipelineConfig.getConsumePipelinesService().getSimpleName());
parserContext.getRegistry().registerBeanDefinition(beanName,beanDefinition);
return beanDefinition;
}
@SneakyThrows
private List<HandlerInvotation> buildHandlerInvotations(PipelineConfig pipelineConfig){
List<HandlerInvotation> handlerInvotations = new ArrayList<>();
for (PipelineHandlerConfig pipelineHandlerConfig : pipelineConfig.getPipelineChain()) {
if(!AbstactChannelHandler.class.isAssignableFrom(pipelineHandlerConfig.getPipelineClass())){
throw new PipelineException("pipelineHandler className must be 【com.github.lybgeek.pipeline.handler.AbstactChannelHandler】 subclass");
}
AbstactChannelHandler channelHandler = (AbstactChannelHandler) pipelineHandlerConfig.getPipelineClass().getDeclaredConstructor().newInstance();
HandlerInvotation invotation = HandlerInvotation.builder()
.args(pipelineConfig.getArgs())
.handler(channelHandler)
.order(pipelineHandlerConfig.getOrder())
.consumePipelinesMethod(pipelineConfig.getConsumePipelinesMethod())
.build();
handlerInvotations.add(invotation);
}
return handlerInvotations;
}
@SneakyThrows
private PipelineConfig buildPipelineConfig(Element element){
String argsType = element.getAttribute("argsType");
String[] argsTypeArr = trimArrayElements(commaDelimitedListToStringArray(argsType));
String consumePipelinesMethod = element.getAttribute("consumePipelinesMethod");
String consumePipelinesServiceClassName = element.getAttribute("consumePipelinesServiceClassName");
Class[] args = null;
if(ArrayUtil.isNotEmpty(argsTypeArr)){
args = new Class[argsTypeArr.length];
for (int i = 0; i < argsTypeArr.length; i++) {
Class argType = Class.forName(argsTypeArr[i]);
args[i] = argType;
}
}
List<PipelineHandlerConfig> pipelineHandlerConfigs = buildPipelineHandlerConfig(element);
return PipelineConfig.builder().args(args)
.consumePipelinesMethod(consumePipelinesMethod)
.consumePipelinesService(Class.forName(consumePipelinesServiceClassName))
.pipelineChain(pipelineHandlerConfigs)
.build();
}
@SneakyThrows
private List<PipelineHandlerConfig> buildPipelineHandlerConfig(Element element){
NodeList nodeList = element.getChildNodes();
if (nodeList == null) {
return Collections.emptyList();
}
List<PipelineHandlerConfig> pipelineHandlerConfigs = new ArrayList<>();
for (int i = 0; i < nodeList.getLength(); i++) {
if (!(nodeList.item(i) instanceof Element)) {
continue;
}
Element childElement = (Element) nodeList.item(i);
if ("pipelineHandler".equals(childElement.getNodeName()) || "pipelineHandler".equals(childElement.getLocalName())) {
String pipelineHanlderClassName = childElement.getAttribute("className");
String pipelineHanlderOrder = childElement.getAttribute("order");
Class pipelineHanlderClass = Class.forName(pipelineHanlderClassName);
PipelineHandlerConfig pipelineHandlerConfig = PipelineHandlerConfig.builder()
.PipelineClass(pipelineHanlderClass)
.order(Integer.valueOf(pipelineHanlderOrder))
.build();
pipelineHandlerConfigs.add(pipelineHandlerConfig);
}
}
return pipelineHandlerConfigs;
}
}
4、註冊解析類
在META-INF資料夾新建spring.handlers檔案，內容如下
http\://lybgeek.github.com/schema/pipeline=com.github.lybgeek.pipeline.spring.shema.PipelineNamespaceHandler
業務專案如何使用該方式實現業務編排
示例：
1、建立管道執行器
@Slf4j
public class UserCheckChannelHandler extends AbstactChannelHandler {
@Override
public boolean handler(ChannelHandlerContext chx) {
ChannelHandlerRequest channelHandlerRequest = chx.getChannelHandlerRequest();
System.out.println("XML------------------------------------步驟一：使用者資料校驗【"+channelHandlerRequest.getRequestId()+"】");
String json = JSON.toJSONString(channelHandlerRequest.getParams());
List<User> users = JSON.parseArray(json,User.class);
if(CollectionUtil.isEmpty(users) || StringUtils.isBlank(users.get(0).getFullname())){
log.error("使用者名稱不能為空");
return false;
}
return true;
}
}
@Slf4j
public class UserFillUsernameAndEmailChannelHandler extends AbstactChannelHandler {
@SneakyThrows
@Override
public boolean handler(ChannelHandlerContext chx) {
ChannelHandlerRequest channelHandlerRequest = chx.getChannelHandlerRequest();
System.out.println("XML------------------------------------步驟二：使用者名稱以及郵箱填充【將漢語轉成拼音填充】【"+channelHandlerRequest.getRequestId()+"】");
String json = JSON.toJSONString(channelHandlerRequest.getParams());
List<User> users = JSON.parseArray(json,User.class);
if(CollectionUtil.isNotEmpty(users)){
User user = users.get(0);
String fullname = user.getFullname();
HanyuPinyinOutputFormat hanyuPinyinOutputFormat = new HanyuPinyinOutputFormat();
hanyuPinyinOutputFormat.setToneType(HanyuPinyinToneType.WITHOUT_TONE);
String username = PinyinHelper.toHanYuPinyinString(fullname, hanyuPinyinOutputFormat);
user.setUsername(username);
user.setEmail(username + "@qq.com");
return true;
}
return false;
}
}
。。。其他執行器具體檢視連結程式碼
2、定義管道xml
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:lybgeek="http://lybgeek.github.com/schema/pipeline"
xmlns="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.3.xsd
http://lybgeek.github.com/schema/pipeline http://lybgeek.github.com/schema/pipeline/pipeline.xsd">
<lybgeek:pipeline consumePipelinesServiceClassName="com.github.lybgeek.pipeline.spring.test.xml.service.UserXmlService" consumePipelinesMethod="save" argsType="com.github.lybgeek.pipeline.spring.test.model.User">
<lybgeek:pipelineHandler className="com.github.lybgeek.pipeline.spring.test.xml.handler.UserCheckChannelHandler" order="1"/>
<lybgeek:pipelineHandler className="com.github.lybgeek.pipeline.spring.test.xml.handler.UserFillUsernameAndEmailChannelHandler" order="2"/>
<lybgeek:pipelineHandler className="com.github.lybgeek.pipeline.spring.test.xml.handler.UserPwdEncryptChannelHandler" order="3"/>
<lybgeek:pipelineHandler className="com.github.lybgeek.pipeline.spring.test.xml.handler.UserMockSaveChannelHandler" order="4"/>
<lybgeek:pipelineHandler className="com.github.lybgeek.pipeline.spring.test.xml.handler.UserPrintChannleHandler" order="5"/>
</lybgeek:pipeline>
3、建立業務管道類
public interface UserXmlService {
boolean save(User user);
}
直接定義介面即可
4、專案啟動類上加上@ImportResource("classpath:/pipeline.xml")
@SpringBootApplication
@ImportResource("classpath:/pipeline.xml")
public class SpringPipelineApplication {
public static void main(String[] args) {
SpringApplication.run(SpringPipelineApplication.class);
}
}
5、測試
@Test
public void testPipelineXml(){
boolean isOk = userXmlService.save(user);
Assert.assertTrue(isOk);
}
總結
本文的管道模式的核心邏輯核心和上篇文章是一樣，只是把管道執行器通過配置檔案集中管理起來，這個後續維護也比較不容易出錯
demo連結
http://github.com/lyb-geek/springboot-learning/tree/master/springboot-pipeline
「其他文章」
- 即刻報名！SegmentFault AIGC Hackathon 黑客馬拉松全新出發！
- SegmentFault 2022 年社群週報 Vol.9
- 社群精選 | 不容錯過的9個冷門css屬性
- 2022最新版 Redis大廠面試題總結（附答案）
- 手寫一個mini版本的React狀態管理工具
- 【vue3原始碼】十三、認識Block
- 天翼雲全場景業務無縫替換至國產原生作業系統CTyunOS！
- JavaScript 設計模式 —— 代理模式
- MobTech簡訊驗證ApiCloud端SDK
- 以羊了個羊為例，淺談小程式抓包與響應報文修改
- 這幾種常見的 JVM 調優場景，你知道嗎？
- 聊聊如何利用管道模式來進行業務編排（下篇）
- 通用ORM的設計與實現
- 如此狂妄，自稱高效能佇列的Disruptor有啥來頭？
- 為什麼要學習GoF設計模式？
- 827. 最大人工島 : 簡單「並查集 列舉」運用題
- 介紹 Preact Signals
- 手把手教你如何使用 Timestream 實現物聯網時序資料儲存和分析
- 850. 矩形面積 II : 掃描線模板題
- Java 併發程式設計解析 | 基於JDK原始碼解析Java領域中的併發鎖，我們可以從中學習到什麼內容？