MQTT Apache Camel路由中的错误导致应用程序停止

Question

我将Apache Camel和Spring（而不是Spring Boot）与XML一起使用。

我有camel-context.xml配置文件，其中包含一个从MQTT服务器到JMS服务器和viceversa的路由示例，只是为了分派消息。

下面是我的camel-context.xml：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:context="http://www.springframework.org/schema/context"
    xmlns:util="http://www.springframework.org/schema/util"
    xmlns:camel="http://camel.apache.org/schema/spring"
    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
                        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
                        http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd
                        http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">

    <bean id="quadtreeProcessor" class="es.gateway.router.QuadtreeProcessor" />

    <camelContext id="camelContext" xmlns="http://camel.apache.org/schema/spring">
        <route id="quadTreeConsumerProducer">
            <from uri="jms:topic:T_ETSI_PRODUCER"/>
            <process ref="quadtreeProcessor"/>
            <to uri="mqtt:quadtree?host=tcp://localhost:1883&amp;publishTopicName=${header.publishTopicName}"/>
        </route>

        <route id="quadTreeConsumerRoute">
            <from uri="mqtt:quadtree?host=tcp://localhost:1883&amp;subscribeTopicName=CONSUMER/DENM/#"/>
            <to uri="jms:topic:T_ETSI_CONSUMER"/>
        </route>
    </camelContext>

    <bean id="jmsConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://localhost:10011"/>
    </bean>
    <bean id="activemq" class="org.apache.activemq.camel.component.ActiveMQComponent" />

package es.conncar.main;

import org.apache.camel.RuntimeCamelException;
import org.apache.camel.main.MainListener;
import org.apache.camel.main.MainListenerSupport;
import org.apache.camel.main.MainSupport;
import org.apache.log4j.Logger;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import es.gateway.config.Config;

public class Main {

    final static Logger logger = Logger.getLogger(Main.class);

    private static boolean exit = false;

    private org.apache.camel.spring.Main camelMain;

    public static void main(String[] args) throws Exception {
        try {
            ApplicationContext springContext = new ClassPathXmlApplicationContext("app-context.xml");
            logger.info("Spring context initialized");

            Config config = (Config) springContext.getBean("config");

            Main main = new Main();
            main.boot();
        } catch (Exception e) {
            logger.error("Unknown error in main", e);
        }
    }

    public void boot() throws Exception {
        camelMain = new org.apache.camel.spring.Main();
        camelMain.addMainListener((MainListener) new Events());
        camelMain.setApplicationContextUri("camel-context.xml"); 
        logger.info("Starting Camel. Use ctrl + c to terminate the JVM.\n");
        camelMain.run();
    }

    public static class Events extends MainListenerSupport {

        @Override
        public void afterStart(MainSupport main) {
            logger.info("Camel is now started!");
        }

        @Override
        public void beforeStop(MainSupport main) {
            logger.info("Camel is now being stopped!");
        }
    }
}

当在我的主类中运行相同的路由和配置时，问题就会出现，由于RuntimeCamelException出现，几秒钟后问题就会消失。

异常跟踪是：

骆驼开始没问题

[30/11/2018 08:37:08][INFO ][es.gateway.main.Main] - Spring context initialized
[30/11/2018 08:37:08][INFO ][es.gateway.main.Main] - Starting Camel. Use ctrl + c to terminate the JVM. 

10秒后

[30/11/2018 08:37:21][ERROR][es.gateway.main.Main] - Unknown error in Camel
    org.apache.camel.RuntimeCamelException: java.util.concurrent.TimeoutException
        at org.apache.camel.util.ObjectHelper.wrapRuntimeCamelException(ObjectHelper.java:1830)
        at org.apache.camel.spring.SpringCamelContext.start(SpringCamelContext.java:136)
        at org.apache.camel.spring.CamelContextFactoryBean.start(CamelContextFactoryBean.java:369)
        at org.apache.camel.spring.CamelContextFactoryBean.onApplicationEvent(CamelContextFactoryBean.java:416)
        at org.apache.camel.spring.CamelContextFactoryBean.onApplicationEvent(CamelContextFactoryBean.java:94)
        at org.springframework.context.event.SimpleApplicationEventMulticaster.doInvokeListener(SimpleApplicationEventMulticaster.java:172)
        at org.springframework.context.event.SimpleApplicationEventMulticaster.invokeListener(SimpleApplicationEventMulticaster.java:165)
        at org.springframework.context.event.SimpleApplicationEventMulticaster.multicastEvent(SimpleApplicationEventMulticaster.java:139)
        at org.springframework.context.support.AbstractApplicationContext.publishEvent(AbstractApplicationContext.java:393)
        at org.springframework.context.support.AbstractApplicationContext.publishEvent(AbstractApplicationContext.java:347)
        at org.springframework.context.support.AbstractApplicationContext.finishRefresh(AbstractApplicationContext.java:883)
        at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:546)
        at org.springframework.context.support.ClassPathXmlApplicationContext.<init>(ClassPathXmlApplicationContext.java:139)
        at org.springframework.context.support.ClassPathXmlApplicationContext.<init>(ClassPathXmlApplicationContext.java:93)
        at org.apache.camel.spring.Main.createDefaultApplicationContext(Main.java:222)
        at org.apache.camel.spring.Main.doStart(Main.java:154)
        at org.apache.camel.support.ServiceSupport.start(ServiceSupport.java:61)
        at org.apache.camel.main.MainSupport.run(MainSupport.java:170)
        at es.gateway.main.Main.boot(Main.java:105)
        at es.gateway.main.Main.main(Main.java:77)
    Caused by: java.util.concurrent.TimeoutException
        at org.fusesource.mqtt.client.Promise.await(Promise.java:83)
        at org.apache.camel.component.mqtt.MQTTEndpoint.connect(MQTTEndpoint.java:348)
        at org.apache.camel.component.mqtt.MQTTConsumer.doStart(MQTTConsumer.java:38)
        at org.apache.camel.support.ServiceSupport.start(ServiceSupport.java:61)
        at org.apache.camel.impl.DefaultCamelContext.startService(DefaultCamelContext.java:3705)
        at org.apache.camel.impl.DefaultCamelContext.doStartOrResumeRouteConsumers(DefaultCamelContext.java:4023)
        at org.apache.camel.impl.DefaultCamelContext.doStartRouteConsumers(DefaultCamelContext.java:3958)
        at org.apache.camel.impl.DefaultCamelContext.safelyStartRouteServices(DefaultCamelContext.java:3878)
        at org.apache.camel.impl.DefaultCamelContext.doStartOrResumeRoutes(DefaultCamelContext.java:3642)
        at org.apache.camel.impl.DefaultCamelContext.doStartCamel(DefaultCamelContext.java:3494)
        at org.apache.camel.impl.DefaultCamelContext.access$000(DefaultCamelContext.java:209)
        at org.apache.camel.impl.DefaultCamelContext$2.call(DefaultCamelContext.java:3253)
        at org.apache.camel.impl.DefaultCamelContext$2.call(DefaultCamelContext.java:3249)
        at org.apache.camel.impl.DefaultCamelContext.doWithDefinedClassLoader(DefaultCamelContext.java:3272)
        at org.apache.camel.impl.DefaultCamelContext.doStart(DefaultCamelContext.java:3249)
        at org.apache.camel.support.ServiceSupport.start(ServiceSupport.java:61)
        at org.apache.camel.impl.DefaultCamelContext.start(DefaultCamelContext.java:3165)
        at org.apache.camel.spring.SpringCamelContext.start(SpringCamelContext.java:133)
        ... 18 more

看起来，当无法连接到MQTT代理时，路由会引发超时（没关系），但是这个异常不是在路由中捕获的，而是在context-main类中捕获的（没关系）。

我遵循了这本食谱：

有人经历过吗？当RuntimeExceptions在路由中出现时，是否有一种方法可以保持路由的工作和主程序的活动？我希望如此！

Thansk提前！

编辑：我发现有几个话题在谈论这个。解决方案似乎是激活监督控制器，但我找不到没有Spring Boot的方法（我使用普通Spring混合xml和注释配置）。有人能帮忙吗？

Answer 1

我建议使用camel-paho，因为Eclipse Paho比camel-mqtt使用的旧的FuseSource MQTT客户机库维护得更活跃。

也就是说，camel-mqtt组件的问题是，它在启动时需要一个工作连接。您可以配置它的重新连接选项来设置各种延迟等。

还有一种替代的路由启动机制，它允许使用后台线程启动路由，后台线程可以监视路由并处理重试等。它的supervisoringRouteController，请注意，围绕其JMX管理功能，还有一些事情需要实现，但除此之外应该没有问题。它缺乏更适当的文档。我们正在考虑将其作为Camel3中的一个更突出的或默认的内容。