Camel流程扩展探索——多个流程是否可行？

程序员文章站 2022-04-17 20:45:48

...

camel框架是一个成熟的流程框架，一般而言我们只是把它应用到一个完整的流程中，而一些逻辑的分支也是在“一个”流程中去控制的。现在如果在流程的源头就需要分支，即拿到源数据，

但是我们需要走不同的流程。（这里不再局限在“一个”流程了！），是否可行呢？下面来探讨。

首先，建立流程的配置：

context-route.xml配置文件

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:drools="http://drools.org/schema/drools-spring"
	xmlns:camel="http://camel.apache.org/schema/spring"
	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
http://drools.org/schema/drools-spring http://anonsvn.jboss.org/repos/labs/labs/jbossrules/trunk/drools-container/drools-spring/src/main/resources/org/drools/container/spring/drools-spring-1.0.0.xsd
http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">
	
	<bean id="fileConverter" class="com.kayak.itmon.web.demo.camel.FileConvertProcessor" />
	<bean id="fileConverter2" class="com.kayak.itmon.web.demo.camel.FileConvertProcessor2" />
	<bean id="fileConverter3" class="com.kayak.itmon.web.demo.camel.FileConvertProcessor3" />
	
	<camelContext id="camel" autoStartup="true" xmlns="http://camel.apache.org/schema/spring">
		 <template id="producerTemplate" />
		<threadPool id="pool" threadName="Thread-dataformat"
			poolSize="50" maxPoolSize="200" maxQueueSize="250" rejectedPolicy="CallerRuns" />
		
		 
		 <route>  
            <from uri="direct://AstartData"/>  
            <process ref="A1"/>  
            <to uri="direct://startA" />
            
        </route> 
        
         <route> 
	        <from uri="direct://startA"/>  
         	<threads executorServiceRef="pool">
         		<process ref="A2" />
         	</threads> 
        </route>
        
        <!-- B流程 -->
         <route>  
            <from uri="direct://BstartData"/>  
            <process ref="B1"/>  
            <to uri="direct://startB" />
        </route> 
        
         <route> 
	        <from uri="direct://startB"/>  
         	<threads executorServiceRef="pool">
         		<process ref="B2" />
         	</threads> 
        </route>
	</camelContext>
	
	<!-- camel的一些bean -->
	<bean id="A1" class="com.kayak.itmon.web.demo.camel.ACamel1"></bean>
	<bean id="A2" class="com.kayak.itmon.web.demo.camel.ACamel2"></bean>
	<bean id="B1" class="com.kayak.itmon.web.demo.camel.BCamel1"></bean>
	<bean id="B2" class="com.kayak.itmon.web.demo.camel.BCamel2"></bean>
</beans>

其中，流程类如下

public class ACamel1 implements Processor {

	@Override
	public void process(Exchange exchange) throws Exception {
		String ret = (String) exchange.getIn().getBody();
		System.out.println("进入到A流程的第一步。。 " + ret);
		ret = ret.toUpperCase();
		exchange.getIn().setBody(ret);
	}

}

public class ACamel2 implements Processor {

	@Override
	public void process(Exchange exchange) throws Exception {
		String ret = (String) exchange.getIn().getBody();
		System.out.println("进入到A流程的第二步。。" + ret);
	}

}

public class BCamel1 implements Processor {

	@Override
	public void process(Exchange exchange) throws Exception {
		String ret = (String) exchange.getIn().getBody();
		System.out.println("进入到B流程的第一步。。  " + ret);
		ret = ret + "hahha";
		exchange.getIn().setBody(ret);
	}

}

public class BCamel2 implements Processor {

	@Override
	public void process(Exchange exchange) throws Exception {
		String ret = (String) exchange.getIn().getBody();
		System.out.println("进入到B流程的第二步。。 " + ret);
	}

}

在测试类里发起流程：

public static void main(String[] args) {
		ApplicationContext ctx = new FileSystemXmlApplicationContext("WebRoot/WEB-INF/conf/xxxxx/context-route.xml");
		System.out.println(ctx);
		
		ProducerTemplate template = ctx.getBean("producerTemplate", ProducerTemplate.class);
		//template.sendBody("direct://AstartData");
		template.sendBody("direct://AstartData", "okoko");
        template.sendBody("direct://BstartData", "okoko");
		try {
			template.start();
		} catch (Exception e) {
			e.printStackTrace();
		}
}

会发现，打印结果如下：

进入到A流程的第一步。。 okoko

进入到A流程的第二步。。OKOKO

进入到B流程的第一步。。 okoko

进入到B流程的第二步。。 okokohahha

说明，同一个数据源（这个demo用的是okoko模拟），可以“分发”给若干个（这里是用2个测试）流程来处理，且流程与流程之间拿到的同一数据源，

但他们的处理互不干涉。只要在ProducerTemplate.sendBody("direct://xxx", data); send多个即可

（注：不要嵌套流程，否则容易出错。即在本例中，

<route>  
            <from uri="direct://AstartData"/>  
            <process ref="A1"/>  
            <to uri="direct://startA" />
            <to uri="direct://BstartData" />
</route>

构成了嵌套，这是不合理的！direct://AstartData和direct://BstartData（这两个都是流程的源头uri）都应该在java类中启动！）

btw, 说下在测试的过程中，出现的一些异常及解决办法吧，记录一下。

1)xml里from uri="xxx"里面要加direct://前缀形如from uri="direct://xxx"，否则会出现异常：

Exception in thread "main" org.apache.camel.RuntimeCamelException: org.apache.camel.FailedToStartRouteException: Failed to start route route4 because of Multiple consumers for the same endpoint is not allowed: Endpoint[direct://start]

at org.apache.camel.util.ObjectHelper.wrapRuntimeCamelException(ObjectHelper.java:1316)

at org.apache.camel.spring.SpringCamelContext.onApplicationEvent(SpringCamelContext.java:120)

at org.apache.camel.spring.CamelContextFactoryBean.onApplicationEvent(CamelContextFactoryBean.java:280)

at org.springframework.context.event.SimpleApplicationEventMulticaster.invokeListener(SimpleApplicationEventMulticaster.java:151)

at org.springframework.context.event.SimpleApplicationEventMulticaster.multicastEvent(SimpleApplicationEventMulticaster.java:128)

at org.springframework.context.support.AbstractApplicationContext.publishEvent(AbstractApplicationContext.java:331)

at org.springframework.context.support.AbstractApplicationContext.finishRefresh(AbstractApplicationContext.java:773)

at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:483)

at org.springframework.context.support.FileSystemXmlApplicationContext.<init>(FileSystemXmlApplicationContext.java:140)

at org.springframework.context.support.FileSystemXmlApplicationContext.<init>(FileSystemXmlApplicationContext.java:84)

at com.kayak.itmon.web.demo.camel.CamelStart.main(CamelStart.java:9)

Caused by: org.apache.camel.FailedToStartRouteException: Failed to start route route4 because of Multiple consumers for the same endpoint is not allowed: Endpoint[direct://start]

at org.apache.camel.impl.DefaultCamelContext.doStartOrResumeRouteConsumers(DefaultCamelContext.java:2019)

at org.apache.camel.impl.DefaultCamelContext.doStartRouteConsumers(DefaultCamelContext.java:1995)

at org.apache.camel.impl.DefaultCamelContext.safelyStartRouteServices(DefaultCamelContext.java:1923)

at org.apache.camel.impl.DefaultCamelContext.doStartOrResumeRoutes(DefaultCamelContext.java:1702)

at org.apache.camel.impl.DefaultCamelContext.doStartCamel(DefaultCamelContext.java:1583)

at org.apache.camel.impl.DefaultCamelContext.doStart(DefaultCamelContext.java:1444)

at org.apache.camel.spring.SpringCamelContext.doStart(SpringCamelContext.java:179)

at org.apache.camel.support.ServiceSupport.start(ServiceSupport.java:60)

at org.apache.camel.impl.DefaultCamelContext.start(DefaultCamelContext.java:1412)

at org.apache.camel.spring.SpringCamelContext.maybeStart(SpringCamelContext.java:228)

at org.apache.camel.spring.SpringCamelContext.onApplicationEvent(SpringCamelContext.java:118)

... 9 more

2)不要有相同的uri="direct://xxx" 否则会出现异常：

Exception in thread "main" org.apache.camel.RuntimeCamelException: org.apache.camel.FailedToCreateRouteException: Failed to create route route1: Route[[From[AstartData]] -> [process[ref:A1], To[direct://st... because of No endpoint could be found for: AstartData, please check your classpath contains the needed Camel component jar.