使用 Paho Java 库处理 MQTT 消息的最佳实践答案

【问题标题】：Best practices in handling MQTT messages with Paho Java library使用 Paho Java 库处理 MQTT 消息的最佳实践
【发布时间】：2017-07-25 22:20:12
【问题描述】：

我正在开发一个 REST API 接口，在某些部分，它必须通过 Paho 客户端库与 MQTT 交互。按照设计，Paho 客户端对于收到的每条消息只能进行一次回调：

mqttClient = new MqttClient(MQTT_ADDRESS, MQTT_CLIENT_ID);
mqttClient.setCallback(new MqttCallbackImpl());
...
private static class MqttCallbackImpl implements MqttCallback {

    @Override
    public void connectionLost(Throwable cause) { }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        switch(topic) {
            // Endless list of cases...
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) { }
}

我正在努力找出处理接收到的消息并做出相应反应的“正确”方式——如何通过避免在负载或主题的某些部分使用巨大的 switch() 来编写回调？

【问题讨论】：

有什么用例让您订阅了足够多的主题，以至于 if/switch 语句有问题？您必须在某处进行过滤
不要介意这个话题：在检查有效载荷的一些信息时，我的疑虑仍然存在

标签： java mqtt paho

【解决方案1】：

您可以通过使用调度 Map 来避免 if/switch 语句。

定义一个简单的接口（这也是一个functional or Single Abstract Method interface）来处理你的负载

interface MqttMessageProcessor {
    void processMessage(String topic, MqttMessage message) throws Exception;
}

然后根据您的要求实现不同的具体类，并将每个主题映射到适当的实例。当消息到达时，它会被分派给正确的处理程序。

Map<String,MqttMessageProcessor> dispatchMap = new HashMap<>();
dispatchMap.put("topic1", new Payload1MessageProcessor());
dispatchMap.put("topic2", new Payload2MessageProcessor());


@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
    dispatchMap.get(topic).processMessage(topic, message);
}

如果您使用的是Java8，您可以使用Map.getOrDefault方法轻松处理当您想要通用处理程序的情况。

@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
    dispatchMap.getOrDefault(topic, generalMessageProcessor).processMessage(topic, message);
}

当添加新主题或有效负载格式时，这更容易维护，因为您只需添加一行代码，而不是挖掘巨大的 if/else 瀑布。

这同样适用于您需要基于某些有效负载属性发送消息的情况。您在 messageArrived 回调中解析有效负载，然后使用调度映射。

【讨论】：

【解决方案2】：

简短回答：大 if/switch 语句是解决方案。

如果您必须根据有效负载做出决定，则无论如何都必须解析有效负载。如果您可以根据主题过滤有效负载类型，那么您可以通过将有效负载解析交给单独的方法来简化事情。

【讨论】：

好吧，看来我的问题并没有那么愚蠢……我松了一口气:)

【解决方案3】：

调度图对我来说似乎是个好主意，直到我发现您可以实际上每个订阅都有单独的侦听器，例如：

client.subscribe(topic, new IMqttMessageListener() {

        @Override
        public void messageArrived(String topic, MqttMessage message) throws Exception {
            // do something
        }

});

这应该可以帮助您在没有丑陋的 if 语句的情况下分离消息处理程序。

【讨论】：