Apache Camel框架之JMS路由

繼上次Camel如何在做項目集成類型的項目中用于從FTP取文件和傳文件之后,我們在系統集成中經常遇到的另一個應用就是將數據通過JMS傳到消息中間件的queue里,或者從消息中間件的queue里取消息.

本文簡單的介紹和示例一個用Camel實現這樣的需求:監聽某一個文件夾是否有文件,取到文件后發送到另外一個系統監聽的queue.

1,因為要用JMS,這里介紹一個open source的activeMQ,可以從http://activemq.apache.org/download.html 下載,下載后解壓,bin目錄有一個activemq.bat文件,在命令行里運行activemq 啟動activeMQ,如果能從從瀏覽器里訪問 http://localhost:8161/admin/則activeMQ成功啟動了.

2,在Camel里實現上圖所示的路由:JAVA項目里需要將activeMQ的jar包配置到classpath下,Java代碼如下:

    private static String user = ActiveMQConnection.DEFAULT_USER;
    private static String password = ActiveMQConnection.DEFAULT_PASSWORD;
    private static String url = ActiveMQConnection.DEFAULT_BROKER_URL;


public static void main(String args[]) throws Exception {        
    CamelContext context = new DefaultCamelContext();        
    ConnectionFactory connectionFactory =
        new ActiveMQConnectionFactory(user, password, url);
    context.addComponent("jms",
        JmsComponent.jmsComponentAutoAcknowledge(connectionFactory));
    System.out.println(url + " " + user + password);        
    context.addRoutes(new RouteBuilder() {
        public void configure() {                
            from("file:d:/temp/inbox").to(
            "jms:queue:TOOL.DEFAULT");
        }
    });
    context.start();
    boolean loop = true;
    while (loop) {
        Thread.sleep(25000);
    }

    context.stop();
}</pre> 

 
  
Camel會在路由的時候將文件的內容以binary message發到activeMQ的名為'TOOL.DEFAULT'的queue .
 
  
用下面的代碼可以從Camel發送的queue里取到消息.
 
  
    private static String user = ActiveMQConnection.DEFAULT_USER;
    private static String password = ActiveMQConnection.DEFAULT_PASSWORD;
    private static String url = ActiveMQConnection.DEFAULT_BROKER_URL;
    private static boolean transacted;
    private static int ackMode = Session.AUTO_ACKNOWLEDGE;


public static void main(String[] args) throws Exception {
    ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, url);
    Connection connection = connectionFactory.createConnection();
    connection.start();
    Session session = connection.createSession(transacted, ackMode);
    Destination destination = session.createQueue("TOOL.DEFAULT");
    MessageConsumer consumer = session.createConsumer(destination);
    Message message = consumer.receive(1000);

    System.out.println("Received: " + message);
    BytesMessage bytesMsg = (BytesMessage) message;
    byte[] bytes = new byte[(int) bytesMsg.getBodyLength()];
    bytesMsg.readBytes(bytes);
    System.out.println("contents: " + new String(bytes));
} </pre>