如何将 Apache Camel 路由到内存中的 ActiveMQ JMS 队列

Question

TLDR：我想知道如何使用 apache 骆驼路由连接到使用 spring boot java 和默认配置创建的内存中 JMS 队列。 目前我尝试连接，但它一直显示错误：

连接到 [tcp://localhost:61616] 失败：10 次尝试继续重试。

我尝试通过添加参数来编辑属性以修改配置和路由，但没有成功。任何帮助表示赞赏。我将下面的整个代码留给需要详细信息的人。

MainApplication.java

public class MainApplication{
    public static void main(String[] args) {
        SpringApplication.run(MainApplication.class, args);
    }
}

JmsConfiguration.java：

import org.apache.activemq.command.ActiveMQQueue;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.annotation.EnableJms;
import org.springframework.jms.core.JmsTemplate;

import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Queue;

@Configuration
@EnableJms
public class JmsConfiguration {

@Autowired
private BeanFactory springContextBeanFactory;

@Bean
public Queue queue(){
    return new ActiveMQQueue("inmemory.queue");
}

@Bean
public JmsTemplate jmsTemplate(ConnectionFactory connectionFactory) throws JMSException {
    return new JmsTemplate(connectionFactory);
}

}

Producer.java

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.web.bind.annotation.*;
import javax.jms.Queue;

@RestController
@RequestMapping("publish")
public class Producer {

    @Autowired
    private JmsTemplate jmsTemplate;

    @Autowired
    private Queue queue;

    @GetMapping("/{message}")
    public String publishString(@PathVariable("message") final String message){
        jmsTemplate.convertAndSend(queue, message);
        return "published succesfully";
    }
}

Consumer.java

import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

@Component
public class Consumer {

    @JmsListener(destination = "inmemory.queue")
    public void listener(String message){
        System.out.println("Receiveage: -> " +message);
    }
}

application-default.yml

spring:    
  activemq:
    in-memory: true
    pool:
      enabled: false

pom.xml

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-activemq</artifactId>
</dependency>
<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-camel</artifactId>
</dependency>

然后配置Apache Camel路由：

import org.apache.camel.builder.RouteBuilder;
import org.springframework.stereotype.Component;
@Component
public class FooRoute extends RouteBuilder {
    @Override
    public void configure() {
// sender
    from("foo-bar")
    .to("activemq:queue:inmemory.queue");
// receiver
    from("activemq:queue:inmemory.queue")
    .to("bar-foo");
    }
}

来自服务器的日志：

2018-12-30 15:12:04,549 INFO  [main] org.apache.activemq.broker.BrokerService : Using Persistence Adapter: MemoryPersistenceAdapter
2018-12-30 15:12:04,624 INFO  [JMX connector] org.apache.activemq.broker.jmx.ManagementContext : JMX consoles can connect to service:jmx:rmi:///jndi/rmi://localhost:1099/jmxrmi
2018-12-30 15:12:04,757 INFO  [main] org.apache.activemq.broker.BrokerService : Apache ActiveMQ 5.14.5 (localhost, ID:hostname-12345-1234567890123-0:1) is starting
2018-12-30 15:12:04,770 INFO  [main] org.apache.activemq.broker.BrokerService : Apache ActiveMQ 5.14.5 (localhost, ID:hostname-12345-1234567890123-0:1) started
2018-12-30 15:12:04,770 INFO  [main] org.apache.activemq.broker.BrokerService : For help or more information please see: http://activemq.apache.org
2018-12-30 15:12:04,995 INFO  [main] org.apache.activemq.broker.TransportConnector : Connector vm://localhost started
2018-12-30 16:01:49,872 WARN  [ActiveMQ Task-1] org.apache.activemq.transport.failover.FailoverTransport : Failed to connect to [tcp://localhost:61616] after: 10 attempt(s) continuing to retry.

Answer 1

缺少的部分是 Camel ActiveMQComponent 的配置，在您的 JMS 路由端点中引用为 activemq:

注意 @Qualifier 注释以确保 bean 名称与 JMS 路由端点中的名称匹配。

您将 ConnectionFactory 注入到组件中。这样组件就可以获取代理 URL 等。

这种带有事务性消息消费的配置示例：

@Bean
@Qualifier("activemq")
ActiveMQComponent activeMQComponent(ConnectionFactory connectionFactory) {
    ActiveMQComponent activeMQComponent = new ActiveMQComponent();
    activeMQComponent.connectionFactory = connectionFactory;
    activeMQComponent.transacted = true;
    activeMQComponent.lazyCreateTransactionManager = false;
    activeMQComponent.cacheLevelName = 'CACHE_CONSUMER';
    activeMQComponent.concurrentConsumers = 5;
    return activeMQComponent;
}