ActiveMQ-Camel的使用

           在一个电子系统中可能接受来自不同供应商的各种订单信息，不同类型的订单走的流程不尽相同，为了快速处理各种不同的订单完成不同的业务。特定义不同的路由信息。根据路由信息的不同，将消息进行不同的处理。如果采用ActiveMQ那么最好采用apache-camel整合，使不同的消息根据不同的流程自动处理到不同的队列中去。

 

采用的jar文件如下：

<?xml version="1.0" encoding="UTF-8"?>
<classpath>
	<classpathentry kind="src" path="src"/>
	<classpathentry kind="src" path="resources"/>
	<classpathentry kind="src" path="message"/>
	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER"/>
	<classpathentry kind="con" path="melibrary.com.genuitec.eclipse.springframework.MYECLIPSE_SPRING20_CORE"/>
	<classpathentry kind="con" path="melibrary.com.genuitec.eclipse.j2eedt.core.MYECLIPSE_JAVAEE_5_CONTAINER"/>
	<classpathentry kind="lib" path="src/activemq-all-5.5.0.jar"/>
	<classpathentry kind="lib" path="src/activemq-pool-5.5.0.jar"/>
	<classpathentry kind="lib" path="src/camel-core-2.7.0.jar"/>
	<classpathentry kind="lib" path="src/camel-jetty-2.7.0.jar"/>
	<classpathentry kind="lib" path="src/camel-jms-2.7.0.jar"/>
	<classpathentry kind="lib" path="src/log4j-1.2.14.jar"/>
	<classpathentry kind="lib" path="src/slf4j-log4j12-1.5.11.jar"/>
	<classpathentry kind="lib" path="src/commons-management-1.0.jar"/>
	<classpathentry kind="lib" path="resources/commons-logging-1.1.1.jar"/>
	<classpathentry kind="lib" path="src/camel-spring-2.7.0.jar"/>
	<classpathentry kind="lib" path="src/org.springframework.aop-3.0.4.RELEASE.jar"/>
	<classpathentry kind="lib" path="src/org.springframework.asm-3.0.4.RELEASE.jar"/>
	<classpathentry kind="lib" path="src/org.springframework.aspects-3.0.4.RELEASE.jar"/>
	<classpathentry kind="lib" path="src/org.springframework.beans-3.0.4.RELEASE.jar"/>
	<classpathentry kind="lib" path="src/org.springframework.context-3.0.4.RELEASE.jar"/>
	<classpathentry kind="lib" path="src/org.springframework.context.support-3.0.4.RELEASE.jar"/>
	<classpathentry kind="lib" path="src/org.springframework.core-3.0.4.RELEASE.jar"/>
	<classpathentry kind="lib" path="src/org.springframework.expression-3.0.4.RELEASE.jar"/>
	<classpathentry kind="lib" path="src/org.springframework.instrument-3.0.4.RELEASE.jar"/>
	<classpathentry kind="lib" path="src/org.springframework.instrument.tomcat-3.0.4.RELEASE.jar"/>
	<classpathentry kind="lib" path="src/org.springframework.jdbc-3.0.4.RELEASE.jar"/>
	<classpathentry kind="lib" path="src/org.springframework.jms-3.0.4.RELEASE.jar"/>
	<classpathentry kind="lib" path="src/org.springframework.orm-3.0.4.RELEASE.jar"/>
	<classpathentry kind="lib" path="src/org.springframework.oxm-3.0.4.RELEASE.jar"/>
	<classpathentry kind="lib" path="src/org.springframework.transaction-3.0.4.RELEASE.jar"/>
	<classpathentry kind="lib" path="src/org.springframework.web-3.0.4.RELEASE.jar"/>
	<classpathentry kind="lib" path="src/org.springframework.web.portlet-3.0.4.RELEASE.jar"/>
	<classpathentry kind="lib" path="src/org.springframework.web.servlet-3.0.4.RELEASE.jar"/>
	<classpathentry kind="lib" path="src/org.springframework.web.struts-3.0.4.RELEASE.jar"/>
	<classpathentry kind="lib" path="src/jetty-all-server-7.1.6.v20100715.jar"/>
	<classpathentry kind="lib" path="src/jetty-websocket-7.1.6.v20100715.jar"/>
	<classpathentry kind="output" path="bin"/>
</classpath>

 

package easyeway.mq.app.demo1;

import javax.jms.ConnectionFactory;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.jms.JmsComponent;
import org.apache.camel.impl.DefaultCamelContext;

/**
 *  设置一系列观察一个目录中新订单情况的路由信息，获取文件，并转换为相关的JMS消息
 *，发送他们到不同的对应的不同队列中。进行相关的处理。
 *
 * @author longgangbai
 *
 */
public class OrderRouterWithRecipientListBean {

    public static void main(String args[]) throws Exception {
        // create CamelContext
    	//创建一个CamelContext
        CamelContext context = new DefaultCamelContext();
        
        // connect to embedded ActiveMQ JMS broker
        //创建一个ActiveMQ类型 连接工厂
        ConnectionFactory connectionFactory = 
            new ActiveMQConnectionFactory("vm://localhost");
        //采用JMS组件发送消息
        context.addComponent("jms",
            JmsComponent.jmsComponentAutoAcknowledge(connectionFactory));

        // add our route to the CamelContext
        //添加自定义路由信息到CamelContext
        context.addRoutes(new RouteBuilder() {
            @Override
            public void configure() {
                // load file orders from src/data into the JMS queue
            	//从src/data加载文件订单生成生成jms，将消息放到jms队列中
            	//src/data为定义的接受订单的目录
                from("file:src/data?noop=true").to("jms:incomingOrders");
        
                // content-based router
                //根据路由信息将消息从队列总分配到本地不同的其他消息中
                from("jms:incomingOrders")
                .choice()
                    .when(header("CamelFileName").endsWith(".xml"))
                        .to("jms:xmlOrders")  
                    .when(header("CamelFileName").regex("^.*(csv|csl)$"))
                        .to("jms:csvOrders")
                    .otherwise()
                        .to("jms:badOrders");        
                //将消息进行相关的业务处理
                from("jms:xmlOrders").bean(RecipientListBean.class);
                
                // test that our route is working
                //业务逻辑的处理过程
                from("jms:accounting").process(new Processor() {
                    public void process(Exchange exchange) throws Exception {
                        System.out.println("Accounting received order: " 
                                + exchange.getIn().getHeader("CamelFileName"));   
                    }
                });                
                from("jms:production").process(new Processor() {
                    public void process(Exchange exchange) throws Exception {
                        System.out.println("Production received order: " 
                                + exchange.getIn().getHeader("CamelFileName"));   
                    }
                });
            }
        });

        // start the route and let it do its work
        context.start();
        Thread.sleep(2000);

        // stop the CamelContext
        context.stop();
    }
}

 

 

package easyeway.mq.app.demo1;

import org.apache.camel.RecipientList;
import org.apache.camel.language.XPath;

/**
 * Recipient List bean that sends orders to the production and 
 * accounting queues if the order originated from a gold customer. 
 * Otherwise, the order is only sent to the accounting queue.
 * 
 * The recipient list annotation is used to accomplish this.
 *
 * @author longgangbai
 *
 */
public class RecipientListBean {
	/**
	 * 根据路由中订单信息用户的信息判断是否是金卡用户
	 * @param customer
	 * @return
	 */
    @RecipientList
    public String[] route(@XPath("/order/@customer") String customer) {
        if (isGoldCustomer(customer)) {
            return new String[] {"jms:accounting", "jms:production"};
        } else {
            return new String[] {"jms:accounting"};
        }
    }

    private boolean isGoldCustomer(String customer) {
        return customer.equals("honda");
    }
}