java 使用 ActiveMQ、Camel 和 Spring 实现请求-回复模式

Implement Request-Reply pattern using ActiveMQ, Camel and Spring

提问by aviad

I am trying to implement the following functionality:

我正在尝试实现以下功能：

Read CSV file line-by-line then for every line:

逐行读取 CSV 文件，然后读取每一行：

1. Build request based on the values that the line contains
2. Send the request to the message queue
3. Other component needs to pick up the message, process the request and send the response to another message queue (known to producer, so the producer could pick up the response).

1. 基于该行包含的值构建请求
2. 将请求发送到消息队列
3. 其他组件需要接收消息，处理请求并将响应发送到另一个消息队列（生产者已知，因此生产者可以接收响应）。

I believe that the request-reply patternfits the bill. I installed ActiveMQ, downloaded camel and tried to use their jms project.

我相信请求-回复模式符合要求。我安装了 ActiveMQ，下载了骆驼并尝试使用他们的 jms 项目。

After configuring the component, the queue and testing connection (worked), I tried to figure out how actually to implement request-reply? I failed to find any good examples

在配置组件、队列和测试连接（工作）后，我试图弄清楚如何实际实现请求-回复？我没有找到任何好的例子

I have a RouteBuilder

我有一个 RouteBuilder

The RouteBuilder

RouteBuilder

public class MyRouteBuilder extends RouteBuilder {
    public static void main(String[] args) throws Exception {
        new Main().run(args);
    }

    public void configure() {
        from("file:src/data?noop=true")
        .to("activemq:RequestQ");

        from("activemq:RequestQ?exchangePattern=InOut&timeToLive=5000") 
        .inOut("activemq:RequestQ", "bean:myBean?method=someMethod"); 
    }
}

camel-context.xml

骆驼上下文.xml

<beans xmlns="http://www.springframework.org/schema/beans"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="
         http://www.springframework.org/schema/beans 
         http://www.springframework.org/schema/beans/spring-beans.xsd
         http://camel.apache.org/schema/spring
         http://camel.apache.org/schema/spring/camel-spring.xsd">

    <camelContext id="camel" xmlns="http://camel.apache.org/schema/spring">
        <package>org.apache.camel.example.spring</package>
    </camelContext>

    <bean id="jmsConnectionFactory" 
        class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://localhost:61616" />
    </bean>

    <bean id="pooledConnectionFactory" 
        class="org.apache.activemq.pool.PooledConnectionFactory" 
        init-method="start" destroy-method="stop">
        <property name="maxConnections" value="8" />
        <property name="connectionFactory" ref="jmsConnectionFactory" />
    </bean>

    <bean id="jmsConfig" 
        class="org.apache.camel.component.jms.JmsConfiguration">
        <property name="connectionFactory" ref="pooledConnectionFactory"/>
        <property name="concurrentConsumers" value="10"/>
    </bean>

    <bean id="activemq" 
        class="org.apache.activemq.camel.component.ActiveMQComponent">
        <property name="configuration" ref="jmsConfig"/>
    </bean>

    <bean id="myBean" class="org.apache.camel.example.spring.MyBean"/>
</beans>

Questions:

问题：

1. How can I read a file line-by-line construct and post a message based on line content?
2. How to configure the route and how to configure the message header in order to get response in temporary queue that will be deleted after the response was picked up?
3. What quick-start guides for the above can you recommend?

1. 如何逐行读取文件结构并根据行内容发布消息？
2. 如何配置路由以及如何配置消息头以便在响应被拾取后将被删除的临时队列中获得响应？
3. 您可以推荐哪些有关上述内容的快速入门指南？

EDIT

编辑

I got the code below working. Now lets say that in the Processor I create the response. How can I send it back? How can I consume the response?

我得到了下面的代码。现在让我们说在处理器中我创建了响应。 我怎样才能把它寄回去？如何使用响应？

public class MyRouteBuilder extends RouteBuilder {

    public static void main(String[] args) throws Exception {
        new Main().run(args);
    }

    public void configure() {
        from("file:/Users/aviad/ws/integ/src/data?fileName=lines.txt&noop=true&idempotent=true")
        .split()
        .tokenize("\n")
        .inOut("activemq:req");

        from("activemq:req")
        .process(new Processor() {
            public void process(Exchange exchange) throws Exception {
                System.out.println(exchange.getIn().getBody(String.class));
                System.out.println("jmscorrelationid=" + exchange.getIn().getHeader("jmscorrelationid"));
                System.out.println("jmsdestination=" + exchange.getIn().getHeader("jmsdestination"));
            }
        });
    }
}

回答by Petter Nordlander

I just had something similar around, so I alterd it and here it is. Note that the 2nd route does not need to be explicitly aware of a request/reply message, only the producer needs to know that. The 2nd route will reply if there is a reply to destination set (which is handled automagically by camel).

我只是有类似的东西，所以我改变了它，它在这里。请注意，第二条路由不需要明确知道请求/回复消息，只有生产者需要知道。如果有对目的地集的回复（由骆驼自动处理），则第二条路线将回复。

I don't know of any good example, but this doc pageis really comprehensive with small examples.

我不知道有什么好的例子，但是这个文档页面非常全面，有小例子。

 <camelContext id="camel" xmlns="http://camel.apache.org/schema/spring">
        <route>
          <from uri="file://c:/apps/in"/>
          <split>
            <tokenize token="\n"/>
            <to uri="activemq:req" pattern="InOut"/>
            <to uri="stream:out"/><!-- print Hello to console -->
          </split>
        </route>
        <route>
          <from uri="activemq:req"/>
            <transform>
              <simple>Hello ${in.body}</simple>
            </transform>
        </route>
    </camelContext>