MQTT 连接到 ActiveMQ TOPIC 在 WSO2 CEP 的事件接收器中花费大量时间

【问题标题】：MQTT connect to ActiveMQ TOPIC take lots of time in Event Receiver of WSO2 CEPMQTT 连接到 ActiveMQ TOPIC 在 WSO2 CEP 的事件接收器中花费大量时间
【发布时间】：2016-08-17 06:45:08
【问题描述】：

更新

在 carbon-analytics-common 项目的 MQTTAdapterListener 的源代码中为 link

run() 将使线程休眠，我相信这就是 MQTT 连接需要这么长时间的原因。

@Override
public void run() {
    while (!connectionSucceeded) {
        try {
            MQTTEventAdapterConstants.initialReconnectDuration = MQTTEventAdapterConstants.initialReconnectDuration
                    * MQTTEventAdapterConstants.reconnectionProgressionFactor;
            Thread.sleep(MQTTEventAdapterConstants.initialReconnectDuration);
            startListener();
            connectionSucceeded = true;
            log.info("MQTT Connection successful");
        } catch (InterruptedException e) {
            log.error("Interruption occurred while waiting for reconnection", e);
        } catch (MqttException e) {
            log.error("MQTT Exception occurred when starting listener", e);

        }

    }
}

initialReconnectDuration 和 reconnectionProgressionFactor 如下MQTTEventAdapterConstants

public static int initialReconnectDuration = 10000;
public static final int reconnectionProgressionFactor = 2;

如果我有 12 个使用 MQTT 的接收器，第 12 个接收器将休眠 40960 秒。好像没有办法修改这两个常量？有没有什么办法可以解决这个问题？为什么 MQTT 连接会以这种方式设置为休眠线程？

我们使用的是 WSO2 CEP 4.0.0 版和 ActiveMQ 5.13.0 版。部署的事件接收器如下

<?xml version="1.0" encoding="UTF-8"?>
<eventReceiver name="5718a6b1851cb3474c6f03c2_PROBE_DATA_RECEIVER"
    statistics="disable" trace="disable" xmlns="http://wso2.org/carbon/eventreceiver">
    <from eventAdapterType="mqtt">
        <property name="clientId">5718a6b1851cb3474c6f03c2</property>
        <property name="topic">PROBE_DATA_571844f5851cb3474c6f0391_57184581851cb3474c6f0394_Topic</property>
        <property name="cleanSession">false</property>
        <property name="url">tcp://127.0.0.1:1883</property>
    </from>
    <mapping customMapping="enable" type="json">
        <property>
            <from jsonPath="$.event.payloadData.dyna.Speed3"/>
            <to name="speed" type="double"/>
        </property>
    </mapping>
    <to streamName="5718a6b1851cb3474c6f03c2_PROBE_DATA_IS" version="1.0.0"/>
</eventReceiver>

clientId是我们数据库内部的这个事件执行计划的随机对象Id。该主题存在于 ActiveMQ 中，我确信该消息已排入其中。 clean session 属性已设置为 false 以实现持久订阅。

但是，无论是在我们重新启动 WSO2 CEP 还是部署新的接收器之后，接收器总是需要花费大量时间来连接到 ActiveMQ TOPIC。

[2016-04-24 01:07:59,018]  INFO {org.wso2.carbon.event.processor.manager.core.internal.CarbonEventManagementService} -  Starting polling event receivers
[2016-04-24 01:07:59,019]  INFO {org.wso2.carbon.event.input.adapter.core.internal.InputAdapterRuntime} -  Connecting receiver 56f66bcc7a22442328cd229a_PROBE_DATA_RECEIVER
[2016-04-24 01:07:59,020]  INFO {org.wso2.carbon.event.input.adapter.core.internal.InputAdapterRuntime} -  Connecting receiver 5718a6b1851cb3474c6f03c2_PROBE_DATA_RECEIVER
[2016-04-24 01:07:59,021]  INFO {org.wso2.carbon.event.input.adapter.core.internal.InputAdapterRuntime} -  Connecting receiver 57144fa0851cb33590672418_X_DATA_RECEIVER
[2016-04-24 01:07:59,022]  INFO {org.wso2.carbon.event.input.adapter.core.internal.InputAdapterRuntime} -  Connecting receiver 5714b5f4851cb3338c26a1cc_PROBE_DATA_RECEIVER
[2016-04-24 01:07:59,022]  INFO {org.wso2.carbon.event.input.adapter.core.internal.InputAdapterRuntime} -  Connecting receiver 571b9ef4851cb356e04ec90b_PROBE_DATA_RECEIVER
[2016-04-24 01:07:59,023]  INFO {org.wso2.carbon.event.input.adapter.core.internal.InputAdapterRuntime} -  Connecting receiver 5714a305851cb3338c26a1c4_DELIVERY_ORDER_RECEI
VER
[2016-04-24 01:07:59,023]  INFO {org.wso2.carbon.event.input.adapter.core.internal.InputAdapterRuntime} -  Connecting receiver 56f6b03e7a22440c30855015_PROBE_DATA_RECEIVER
[2016-04-24 01:07:59,024]  INFO {org.wso2.carbon.event.input.adapter.core.internal.InputAdapterRuntime} -  Connecting receiver 56f665f27a22442328cd2299_PROBE_DATA_RECEIVER
[2016-04-24 01:07:59,024]  INFO {org.wso2.carbon.event.input.adapter.core.internal.InputAdapterRuntime} -  Connecting receiver 5718a5d7851cb3474c6f03bf_PROBE_DATA_RECEIVER
[2016-04-24 01:07:59,025]  INFO {org.wso2.carbon.event.input.adapter.core.internal.InputAdapterRuntime} -  Connecting receiver 56ef98137a22442e28360c0b_DELIVERY_ORDER_RECEI
VER
[2016-04-24 01:07:59,026]  INFO {org.wso2.carbon.event.input.adapter.core.internal.InputAdapterRuntime} -  Connecting receiver 571747f2851cb3338c26a1da_PROBE_DATA_RECEIVER
[2016-04-24 01:07:59,026]  INFO {org.wso2.carbon.event.input.adapter.core.internal.InputAdapterRuntime} -  Connecting receiver 56f66c657a22442328cd229c_PROBE_DATA_RECEIVER
[2016-04-24 01:08:19,044]  INFO {org.wso2.carbon.event.input.adapter.mqtt.internal.util.MQTTAdapterListener} -  MQTT Connection successful
[2016-04-24 01:08:39,041]  INFO {org.wso2.carbon.event.input.adapter.mqtt.internal.util.MQTTAdapterListener} -  MQTT Connection successful
[2016-04-24 01:09:19,034]  INFO {org.wso2.carbon.event.input.adapter.mqtt.internal.util.MQTTAdapterListener} -  MQTT Connection successful
[2016-04-24 01:10:39,037]  INFO {org.wso2.carbon.event.input.adapter.mqtt.internal.util.MQTTAdapterListener} -  MQTT Connection successful
[2016-04-24 01:13:19,046]  INFO {org.wso2.carbon.event.input.adapter.mqtt.internal.util.MQTTAdapterListener} -  MQTT Connection successful
[2016-04-24 01:18:39,035]  INFO {org.wso2.carbon.event.input.adapter.mqtt.internal.util.MQTTAdapterListener} -  MQTT Connection successful
[2016-04-24 01:29:19,038]  INFO {org.wso2.carbon.event.input.adapter.mqtt.internal.util.MQTTAdapterListener} -  MQTT Connection successful
[2016-04-24 01:50:39,036]  INFO {org.wso2.carbon.event.input.adapter.mqtt.internal.util.MQTTAdapterListener} -  MQTT Connection successful
[2016-04-24 02:33:19,039]  INFO {org.wso2.carbon.event.input.adapter.mqtt.internal.util.MQTTAdapterListener} -  MQTT Connection successful
[2016-04-24 03:58:39,043]  INFO {org.wso2.carbon.event.input.adapter.mqtt.internal.util.MQTTAdapterListener} -  MQTT Connection successful
[2016-04-24 06:49:19,038]  INFO {org.wso2.carbon.event.input.adapter.mqtt.internal.util.MQTTAdapterListener} -  MQTT Connection successful
[2016-04-24 12:30:39,040]  INFO {org.wso2.carbon.event.input.adapter.mqtt.internal.util.MQTTAdapterListener} -  MQTT Connection successful

有 12 个接收器，WSO2 于 01:07:59 开始轮询。大约 10 分钟后，12 个接收器中只有 6 个连接成功，其他接收器大约需要 10 个小时或更长时间。

有人知道为什么 WSO2 CEP 接收器的 MQTT 连接这么慢吗？

【问题讨论】：

    标签： wso2 activemq mqtt jms-topic wso2cep


    【解决方案1】：

    没有提供开箱即用的方法来配置重新连接持续时间和渐进因子值。但是由于您已经弄清楚了代码，您可以修补该组件（从配置文件中读取这些值或通过适配器 ui 或硬编码值对其进行配置）并将修补程序应用为 described here 以克服此限制.

    【讨论】：

    • 您好 Rajeev，我非常感谢您的精彩想法和回答。我检查了我使用的 WSO2CEP 版本是 4.0.0，并且它的存储库\组件\插件中的 MQTT 是 org.wso2.carbon.event.input.adapter.mqtt_5.0.3.jar。于是我查看了 carbon-analytics-common 项目的 tag v5.0.3 的源代码，修改了代码并 maven 构建了 org.wso2.carbon.event.input.adapter.mqtt 模块。终于得到了一个org.wso2.carbon.event.input.adapter.mqtt-5.0.3.jar！
    • 但是，我仍然不知道如何修补 WSO2CEP。 WSO2 的补丁只是一个简单的 .jar 文件并将其放入 /repository/components/patches/ 中？我已经这样做了并重新启动了 WSO2CEP 服务器，但无论是否使用 -DapplyPatches 参数，它都不起作用。
    • 顺便说一句，由于我构建的文件名是 org.wso2.carbon.event.input.adapter.mqtt-5.0.3.jar 而 WSO2CEP 插件中的文件名是 org.wso2。 carbon.event.input.adapter.mqtt_5.0.3.jar。因此，我将构建的 jar 重命名为 org.wso2.carbon.event.input.adapter.mqtt_5.0.3.jar 并将其放入 patch 文件夹中。
    • 您需要重命名 jar 文件，类似于 plugins 文件夹中的 jar。还在 /repository/component/patches 中创建一个类似于 patch0999 的文件夹，并将您的 jar 放在该文件夹中。然后重启服务器，它会自动被应用
    • 如果您想检查它是否正确应用，请在重新启动服务器后检查 patchs/patch0999 中的 jar 与插件中的 jar 的 md5sum。
    猜你喜欢
    • 1970-01-01
    • 2016-08-26
    • 2011-05-03
    • 2013-09-09
    • 2014-04-16
    • 2017-03-27
    • 2014-10-10
    • 1970-01-01
    • 1970-01-01
    相关资源
    最近更新 更多
    热门标签
    Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode