java - 如果 ActiveMQ 未启动，Spring 容器将挂起

Question

我目前正在使用 DefaultMessageListenerContainer 来创建监听器，并使用 JmsTemplate 将消息(生产者)发送到队列。

Spring 配置片段:

@Bean
public ActiveMQConnectionFactory connectionFactory() {
    ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_BROKER_URL);
    factory.setRedeliveryPolicy(desiredRedeliveryPolicy());

    return factory;
}

@Bean
public DefaultMessageListenerContainer requestMessageListenerContainer() {
    DefaultMessageListenerContainer requestMessageListenerContainer = new DefaultMessageListenerContainer();
    requestMessageListenerContainer.setConcurrentConsumers(noOfconcurrentConsumers);
    requestMessageListenerContainer.setConnectionFactory(connectionFactory());
    requestMessageListenerContainer.setDestinationName(requestQueueName);
    requestMessageListenerContainer.setMessageListener(requestMessageListener());
    requestMessageListenerContainer.setSessionAcknowledgeMode(Session.CLIENT_ACKNOWLEDGE);
    requestMessageListenerContainer.setSessionTransacted(false);

    return requestMessageListenerContainer;
}


@Bean
public JmsTemplate requestJmsTemplate() {
    JmsTemplate jmsTemplate = new JmsTemplate();
    jmsTemplate.setConnectionFactory(connectionFactory());
    jmsTemplate.setDefaultDestination(requestMqQueue());

    return jmsTemplate;
}

我目前遇到的问题是，如果在运行应用程序之前未启动 ActiveMQ，我的 spring 容器加载过程就会卡住。

我相信 DefaultMessageListenerContainer 和 JmsTemplate 正在尝试创建它们与 ActiveMQConnectionFactory 的连接和 session 。

在 spring 之外，如果提供的 activemq 没有运行，我知道，

activeMQConnection.createSession()

是执行会卡住的地方。在常规 Java 代码中，我可以使一些长时间处理/可能卡住的进程超时。但是我怎样才能在 spring 容器中做这样的事情呢？

我想知道有没有更好的方法来声明这些bean，这样我就知道activemq卡住了，容器不卡了？

在此先感谢您的帮助。

更新 1:

我更新了连接工厂的连接 URL，还添加了一个 ExceptionListener:

@Bean
public ActiveMQConnectionFactory connectionFactory() {
    ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(String.format("failover://(%s)?startupMaxReconnectAttempts=1&maxReconnectAttempts=2", ActiveMQConnectionFactory.DEFAULT_BROKER_BIND_URL));
    factory.setRedeliveryPolicy(desiredRedeliveryPolicy());
    factory.setExceptionListener(factoryExceptionListener());

    return factory;
}

public FactoryExceptionListener factoryExceptionListener(){
    return new FactoryExceptionListener();
}


public class FactoryExceptionListener implements ExceptionListener {
private static XLogger LOG =  XLoggerFactory.getXLogger(FactoryExceptionListener.class);

@Override
public void onException(JMSException exception) {
    LOG.error("Factory Exception Caught: "+exception.getMessage());
    System.exit(1);
}
}

现在是一个愚蠢的问题。

我可以看到正在打印错误日志，但应用程序在 System.exit(1) 后并未退出。我在这里做错了什么吗？

此更改有助于阻塞调用不再阻塞。但是我无法退出，这意味着应用程序开始执行并抛出一堆异常，因为 activeMQ 不可用。

我反而希望它(现在)使应用程序崩溃。我该怎么做？

更新 2: 我没有退出应用程序(仍然无法正常工作 - 可能与监听器有关)，而是更改了异常监听器以使我的实现更有意义。如果异常监听器被触发，我现在正试图让 Broker Up。

public void onException(JMSException exception) {
    LOG.error("Factory Exception Caught: "+exception.getMessage());

    try {
        BrokerService brokerService = new BrokerService();
        brokerService.addConnector("tcp://localhost:61616");
        brokerService.setDataDirectory("C:/temp/data");
        brokerService.setEnableStatistics(true);
        brokerService.setPersistent(true);
        brokerService.start();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

但我遇到以下异常:

2014-07-16 10:24:35.009 [ActiveMQ Task-1] ERROR o.a.a.t.failover.FailoverTransport - Failed to connect to [tcp://localhost:61616] after: 1 attempt(s)
2014-07-16 10:24:35.012 [ActiveMQ Connection Executor: unconnected] ERROR c.b.s.o.b.m.FactoryExceptionListener - Factory Exception Caught: Connection refused: connect
Exception in thread "ActiveMQ Connection Executor: unconnected" java.lang.NoSuchMethodError: org.apache.activemq.transport.TransportFactory.bind(Lorg/apache/activemq/broker/BrokerService;Ljava/net/URI;)Lorg/apache/activemq/transport/TransportServer;
at org.apache.activemq.broker.BrokerService.createTransportConnector(BrokerService.java:2249)
at org.apache.activemq.broker.BrokerService.addConnector(BrokerService.java:291)
at org.apache.activemq.broker.BrokerService.addConnector(BrokerService.java:281)
at com.bhn.service.ordermgmt.bulkorder.mq.FactoryExceptionListener.onException(FactoryExceptionListener.java:19)
at org.apache.activemq.ActiveMQConnection$5.run(ActiveMQConnection.java:1998)
at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
at java.lang.Thread.run(Unknown Source)

但是，当我尝试从另一个项目执行相同的代码时。我成功地启动并运行了 BrokerService。 我不确定这个错误是什么意思以及如何解决它？

更新 3: 不确定之前出了什么问题，但现在可以使用相同的代码。感谢您的帮助@Tim

Answer 1

这是因为您指定的默认 URL 使用的是 Failover运输。默认情况下，传输将尝试连接到代理，直到您关闭您的应用程序。 createSession 调用触发客户端尝试将其连接信息请求发送到 Broker，但在客户端连接到 Broker 之前不会发生。

评论中提到的一种解决方案是禁用自动启动功能，这样 session 创建调用就不会在启动时执行。但是，如果您稍后在代理仍处于关闭状态时触发 session 创建，您仍然会遇到挂起。您可以使用 failover 上显示的选项，使用一定数量的连接尝试配置故障转移传输。传输页面。