骆驼JMS交易不起作用

我正在尝试获取一个骆驼路由JMS->HTTP4，事务工作，但是当异常时消息没有传输到ActiveMQ.DLQ，我看不出为什么。

下面的示例说明了如果 REST 服务的服务器关闭并且无法传递路由，会发生什么情况。

我得到正确的异常：

2018-01-18 12:30:50:962-[Camel (LRM-Relay) thread #5 - JmsConsumer[myIncomingQueue]] WARN  o.a.c.s.s.TransactionErrorHandler - Transaction rollback (0x30a1c779) redelivered(false) for (MessageId: ID:MGR-MacBook-Pro.local-51837-1516262355358-4:2:1:1:16 on ExchangeId: ID-MGR-MacBook-Pro-local-1516275047663-0-1) caught: java.net.ConnectException: Cannot connect to CORE REST  
2018-01-18 12:30:50:965-[Camel (LRM-Relay) thread #5 - JmsConsumer[myIncomingQueue]] WARN  o.a.c.c.j.EndpointMessageListener - Execution of JMS message listener failed. Caused by: [org.apache.camel.RuntimeCamelException - java.net.ConnectException: Cannot connect to CORE REST] 
org.apache.camel.RuntimeCamelException: java.net.ConnectException: Cannot connect to CORE REST …

但消息将被使用并从队列中删除。我的假设是使用事务/事务处理Camel和AMQ将解决此问题并将消息移动到ActiveMQ.DLQ。

我已经阅读了骆驼在行动第1版的第9章，并在谷歌上搜索，但没有找到我的问题的任何解决方案。

我知道我可以创建/定义我自己的 TransactionErrorHandler() 并将消息存储在我选择的队列中，但我的印象是这是使用事务处理时的默认......

我正在使用安装和配置的独立 ActiveMQ 5.15.2 原版。
Camel 2.20.1
Mac OS 10.13.2 上的 Java 8_144

我的配置：

@Configuration
public class Config {
/**
* The Camel context.
*/
final CamelContext camelContext;

/**
* The Broker url.
*/
@Value("${jms.broker.url}")
private String brokerURL;

/**
* Instantiates a new Config.
*
* @param camelContext   the sisyfos context
* @param metricRegistry the metric registry
*/
@Autowired
public Config(final CamelContext camelContext, final MetricRegistry metricRegistry) {
this.camelContext = camelContext;
this.metricRegistry = metricRegistry;
}
@Bean
public ActiveMQConnectionFactory activeMQConnectionFactory() {
final ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory();
activeMQConnectionFactory.setBrokerURL(brokerURL);
return activeMQConnectionFactory;
}
/**
* Pooled connection factory pooled connection factory.
*
* @return the pooled connection factory
*/
@Bean
@Primary
public PooledConnectionFactory pooledConnectionFactory() {
final PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory();
pooledConnectionFactory.setMaxConnections(8);
pooledConnectionFactory.setMaximumActiveSessionPerConnection(500);
pooledConnectionFactory.setConnectionFactory(activeMQConnectionFactory());
return pooledConnectionFactory;
}
/**
* Jms configuration jms configuration.
*
* @return the jms configuration
*/
@Bean
public JmsConfiguration jmsConfiguration() {
final JmsConfiguration jmsConfiguration = new JmsConfiguration();
jmsConfiguration.setConnectionFactory(pooledConnectionFactory());
jmsConfiguration.setTransacted(true);
jmsConfiguration.setTransactionManager(transactionManager());
jmsConfiguration.setConcurrentConsumers(10);
return jmsConfiguration;
}
/**
* Transaction manager jms transaction manager.
*
* @return the jms transaction manager
*/
@Bean
public JmsTransactionManager transactionManager() {
final JmsTransactionManager transactionManager = new JmsTransactionManager();
transactionManager.setConnectionFactory(pooledConnectionFactory());
return transactionManager;
}
/**
* Active mq component active mq component.
*
* @return the active mq component
*/
@Bean
public ActiveMQComponent activeMQComponent(JmsConfiguration jmsConfiguration,
PooledConnectionFactory pooledConnectionFactory,
JmsTransactionManager transactionManager) {
final ActiveMQComponent activeMQComponent = new ActiveMQComponent();
activeMQComponent.setConfiguration(jmsConfiguration);
activeMQComponent.setTransacted(true);
activeMQComponent.setUsePooledConnection(true);
activeMQComponent.setConnectionFactory(pooledConnectionFactory);
activeMQComponent.setTransactionManager(transactionManager);
return activeMQComponent;
}

}

我的路线：

@Component
public class SendToCore extends SpringRouteBuilder {
@Override
public void configure() throws Exception {
Logger.getLogger(SendToCore.class).info("Sending to CORE");

//No retries if first fails due to connection error
interceptSendToEndpoint("http4:*")
.choice()
.when(header("JMSRedelivered").isEqualTo("false"))
.throwException(new ConnectException("Cannot connect to CORE REST"))
.end();
from("activemq:queue:myIncomingQueue")
.transacted()
.setHeader(Exchange.CONTENT_TYPE, constant("application/xml"))
.to("http4:localhost/myRESTservice")
.log("${header.CamelHttpResponseCode}")
.end();
}
}

您可能会在某些豆子中找到多余的声明，这就是我试图解决问题......

添加一个指向我的 Github 存储库的链接，其中包含一个小型测试项目，说明这一点：
https://github.com/hakuseki/transacted

这可能是 SpringBoot 自动配置的问题。

如果消息丢失而不是转到DLQ，Camel的ActiveMQ组件会自动提交它们，而不是等到工作完成。

更新：使您的示例与 Java Config 配合使用的步骤

注意：我的配置没有事务管理器，因为您的案例不需要它。相反，只需将ActiveMQComponenttransacted设置为true，lazyCreateTransactionManager设置为false。然后，您与经纪人进行了"本地"交易，这就是您所需要的。

我从您的路由中删除了.transacted()(需要一个事务管理器，但不需要有一个"JMS 本地事务处理"路由)
我在路由类中注释掉了您的错误处理程序(需要一个事务管理器，您可以使用默认的错误处理程序)
在MainApplication中禁用 JMS 和 ActiveMQ 的自动配置：@SpringBootApplication(exclude = { JmsAutoConfiguration.class, ActiveMQAutoConfiguration.class})
将你的 Java 配置替换为以下配置(改编自这个问题：ConnectionFactory 在骆驼之前被摧毁)

Java配置：

@Value("${jms.broker.url}") 
String brokerURL;
@Bean
public ActiveMQConnectionFactory connectionFactory() {
final ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory();
activeMQConnectionFactory.setBrokerURL(brokerURL);
return activeMQConnectionFactory;
}
@Bean
@Primary
public PooledConnectionFactory pooledConnectionFactory(ConnectionFactory cf) {
final PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory();
pooledConnectionFactory.setMaxConnections(1);
pooledConnectionFactory.setConnectionFactory(cf);
return pooledConnectionFactory;
}
@Bean(name = "activemq")
@ConditionalOnClass(ActiveMQComponent.class)
public ActiveMQComponent activeMQComponent(ConnectionFactory connectionFactory) {
ActiveMQComponent activeMQComponent = new ActiveMQComponent();
activeMQComponent.setConnectionFactory(connectionFactory);
activeMQComponent.setTransacted(true);
activeMQComponent.setLazyCreateTransactionManager(false);
return activeMQComponent;
}

最后，只是为了"运行"路线，我添加了一个小的骆驼路线测试

@RunWith(CamelSpringBootRunner.class)
@SpringBootTest(classes = MainApplication.class)
public class SampleCamelApplicationTest {
@Produce(uri = "activemq:queue:myIncomingQueue")
protected ProducerTemplate template;
@Test
public void shouldProduceMessages() throws Exception {
template.sendBody("test");
Thread.sleep(20000); //wait for ActiveMQ redeliveries
}
}

如果我运行此测试，消息将ActiveMQ.DLQ.

希望这有帮助

刚刚注意到，如果您希望 Spring Boot 处理这些池和配置的生命周期，那么您不应该直接调用它们的方法，而是让它们作为方法签名中的参数提供

例如这个

public ActiveMQComponent activeMQComponent() {

应该是

public ActiveMQComponent activeMQComponent(JmsConfiguration config, ConnectionFactory cf, ...) {

然后 Spring Boot 会为你提供这些豆子。

关于为什么你的交易不起作用，然后你可以看看骆驼在行动第2版书中的一些交易示例：https://github.com/camelinaction/camelinaction2/tree/master/chapter12

更新：使您的示例与 Java Config 配合使用的步骤

相关内容

最新更新

热门标签：