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

Posted 2023-03-10

【中文标题】使用 ActiveMQ、Camel 和 Spring 实现请求-回复模式

【英文标题】：Implement Request-Reply pattern using ActiveMQ, Camel and Spring

【发布时间】：2013-04-21 00:58:36

【问题描述】：

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

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

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

我相信request-reply pattern 符合要求。 我安装了 ActiveMQ，下载了骆驼并尝试使用他们的 jms 项目。

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

我有一个 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

<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>

问题：

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

编辑

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

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"));
            
        );
    

【参考方案1】：

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

我不知道有什么好的例子，但this doc page 用小例子真的很全面。

 <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>

【讨论】：

谢谢！看起来很简单。我会检查一下。顺便说一句：如何将某种处理程序附加到第二条路线？

处理程序，像java方法？只需使用 bean 组件 (camel.apache.org/bean) 或处理器。而不是<transform> 放入<bean ref="someSpringBean"/>

我确实也回答了您编辑的问题。 “stream:out”端点将响应“Hello 不管”打印到屏幕。当然，它应该是别的东西，比如豆子之类的。

嗯......“你好”的东西从来没有奏效......猜猜这是混乱。

好的。这个概念仍然适用，我测试了 XML dsl，它按预期工作。