阿帕奇骆驼 ?transacted=true答案

【问题标题】：Apache Camel ?transacted=true阿帕奇骆驼 ?transacted=true
【发布时间】：2019-01-15 11:59:15
【问题描述】：

您好 Camel/jms 开发人员。使用 Apache Camel amqp jms 连接器。并作为代理 ActiveMQ。

我的配置是完全默认的。

这是一个消费者代码示例：

 public static void main(String[] args) throws Exception {
    AMQPComponent amqpComponent = AMQPComponent.amqpComponent(HOST, USER, PWD);
    CamelContext context = new DefaultCamelContext();
    context.addComponent("amqp", amqpComponent);
    context.start();
    context.addRoutes(new RouteBuilder() {
        @Override
        public void configure() {
            from("amqp:queue:1test.queue?transacted=true")
                .to("stream:out")
            .end();
        }
    });
    Thread.sleep(20*1000);
    context.stop();
}

很容易看到，我已经配置了交易消费者。对于 1test.queue。当我运行它时，在日志中看到：

[main] INFO org.apache.camel.impl.DefaultCamelContext - 路由：route1 开始并使用：amqp://queue:1test.queue?transacted=true
[AmqpProvider :(1):[amqp:HOST2]] INFO org.apache.qpid.jms.sasl.SaslMechanismFinder - SASL 身份验证的最佳匹配是：SASL-PLAIN
[AmqpProvider :(1):[amqp:HOST2]] 信息 org.apache.qpid.jms.JmsConnection - 连接 ID:...:1 连接到远程代理：amqp:HOST2
[AmqpProvider :(2):[amqp:HOST2]] INFO org.apache.qpid.jms.sasl.SaslMechanismFinder - SASL 身份验证的最佳匹配是：SASL-PLAIN
[AmqpProvider :(2):[amqp:HOST2]] 信息 org.apache.qpid.jms.JmsConnection - 连接 ID:...:2 连接到远程代理：amqp:HOST2
[AmqpProvider :(3):[amqp:HOST2]] INFO org.apache.qpid.jms.sasl.SaslMechanismFinder - SASL 身份验证的最佳匹配是：SASL-PLAIN
[AmqpProvider :(3):[amqp:HOST2]] INFO org.apache.qpid.jms.JmsConnection - 连接 ID:...:3 连接到远程代理：amqp:HOST2

如果我从消费者那里删除 ?transacted=true

[AmqpProvider :(1):[amqp:HOST2]] INFO org.apache.qpid.jms.sasl.SaslMechanismFinder - SASL 身份验证的最佳匹配是：SASL-PLAIN
[AmqpProvider :(1):[amqp:HOST2]] INFO org.apache.qpid.jms.JmsConnection - 连接 ID:...:1 连接到远程代理：amqp:HOST2

它只出现一次。

如何解释这种行为？这通常适用于骆驼交易的消费者？

谢谢。

P.S 检查了this topic，但不确定如何将其映射到 Camel 现实。

【问题讨论】：

标签： java apache-camel jms activemq amqp

【解决方案1】：

使用 amqp 的 jms。

我有类似的问题。但是当我使用 JMS 而不是 AMQP 时，它运行良好，只有一次日志，即创建了单个连接。

AMQ 组件似乎存在一些问题。

谢谢，拉胡尔

【讨论】：

【解决方案2】：

AMQP Camel 组件尚未定义池连接工厂，因此它为每次迭代创建一个新连接以检查代理中是否有消息。

为避免这种情况，您应该将 CachingConnectionFactory 定义为 AMQPComponent：

import org.apache.camel.component.amqp.AMQPComponent;
import org.apache.camel.component.jms.JmsConfiguration;

import org.apache.qpid.jms.JmsConnectionFactory;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Primary;
import org.springframework.jms.connection.CachingConnectionFactory;

@Bean
public JmsConnectionFactory jmsConnectionFactory() {
    JmsConnectionFactory jmsConnectionFactory = new JmsConnectionFactory(brokerUser, brokerPassword, brokerUrl);

    return jmsConnectionFactory;
}

@Bean
@Primary
public CachingConnectionFactory jmsCachingConnectionFactory(JmsConnectionFactory jmsConnectionFactory) {
    CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(jmsConnectionFactory);

    return cachingConnectionFactory;
}

@Bean
public JmsConfiguration jmsConfig(CachingConnectionFactory cachingConnectionFactory) {
    JmsConfiguration jmsConfiguration = new JmsConfiguration();

    jmsConfiguration.setConnectionFactory(cachingConnectionFactory);
    jmsConfiguration.setCacheLevelName("CACHE_CONSUMER");

    return jmsConfiguration;
}

@Bean
public AMQPComponent amqpComponent(JmsConfiguration jmsConfiguration) {
    AMQPComponent amqpComponent = new AMQPComponent();

    amqpComponent.setConfiguration(jmsConfiguration);

    return amqpComponent;
}

您将获得与 JMS Camel 组件相同的行为。

更多信息请访问AMQP Camel Component 页面

【讨论】：