3

我的应用程序的工作流程：我已经设置了activemq并编写了spring框架来监听activemq的队列。每当队列中有消息时，侦听器将获取消息然后将消息出列并执行我的业务逻辑。

现在在测试用例中，如果我的业务逻辑中有任何运行时错误，则消息应该回滚到队列中。这样消费者就可以再次消费消息并再次执行我的业务逻辑。

如何使用 spring-camel 实现这一目标？

我为 ActiveMqConsumer 编写的代码

public class ActiveMqConsumer {
public static void main(String[] args){
    try {
        PropertyConfigurator.configure("C:/Users/awsdemo/src/main/resources/log4j.properties");
        ApplicationContext springcontext = new FileSystemXmlApplicationContext("C:/Users/awsdemo/src/main/resources/activecamel.xml");
        CamelContext context = springcontext.getBean("activeContext", CamelContext.class);
        //context.addComponent("activemq", activeMQComponent("tcp://localhost:61616?broker.persistent=false"));
        context.start();
        //Thread.sleep(1000);
        //context.stop();

    } catch ( Exception e ) {
        System.out.println(e);
    }

}
}

ActiveMQRouterBuilder 的代码

public class ActiveMQRouterBuilder extends RouteBuilder {
@Override
public void configure() throws Exception {
    String activeMqURI = "activemq:queue:ThermalMap";
    System.out.print(activeMqURI);
    from( activeMqURI).to("bean:activemqProcessor?method=processMessage");

}
}

ActiveMQProcessor 的代码

public class ActiveMQProcessor{
public void processMessage(Exchange exchange) throws Exception{
    System.out.println("\ninside processMessage :Consumer1");
    //System.out.println(exchange.getIn().getBody());
    Object object = exchange.getIn().getBody();
    FunctionNames functionNamesObject=new FunctionNames();
    //Call Intergration function to execute .exe file
    try {
                 /* my business logic*/
    } catch (IOException e) {
                     /* message should rollback here to activemq*/

        // TODO Auto-generated catch block
        e.printStackTrace();
    } catch (InterruptedException e) {
                     /* or message should rollback here to activemq*/
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
    System.out.println("ActiveMQProcessor: finished");
}

}

以上三个文件组合起来充当消费者。这三个文件是在file.xml中配置的activecamel.xml。包含以下activecamel.xml代码

<camelContext id="activeContext" xmlns="http://camel.apache.org/schema/spring">
    <routeBuilder ref="activeMQRouter" />
</camelContext>

<bean id="activemq" class="org.apache.activemq.camel.component.ActiveMQComponent">
    <property name="brokerURL" value="tcp://localhost:61616?jms.prefetchPolicy.queuePrefetch=1" />
</bean>

<bean id="activeMQRouter" class="main.java.com.aranin.activemq.ActiveMQRouterBuilder"/>

<bean id="activemqProcessor" class="main.java.com.aranin.activemq.ActiveMQProcessor"/>

在 ActiveMQProcessor 中，我编写了我的业务逻辑，如果有任何错误，它会抛出错误来捕获块。在 catch 块中，我应该编写代码来回滚消息。应该有什么代码来回滚消息？

4

1 回答 1

4

如果你在路由中使用 transacted()，那么你应该从 ActiveMQProcessor 抛出一个异常，Camel 会自动回滚 TX。

  from( activeMqURI)
    .transacted()      
    .to("bean:activemqProcessor?method=processMessage");

您还需要将 ActiveMQ 配置为已交易

<bean id="activemq" class="org.apache.activemq.camel.component.ActiveMQComponent">
    <property name="brokerURL" value="tcp://localhost:61616?jms.prefetchPolicy.queuePrefetch=1" />
    <property name="transacted" value="true"/>
</bean>

并设置一个 JMS 事务管理器。在以下位置查看更多详细信息：http ://camel.apache.org/transactional-client.html

虽然是后者，但 Camel 应该默认设置一个，如果你有 transacted=true。但最好还是在xml文件中定义一个事务管理器，并在activemq配置中引用。上述链接中的所有详细信息。

如果你有 Camel in Action 一书，请阅读第 9 章。

于 2013-09-18T16:08:10.693 回答