mqtt 发送消息过多_MQTT系列Eclipse.Paho源码分析(二)消息的发送与接收

在 MQTT 协议中，消息的发送和接收是非常重要的。Eclipse Paho 是一个 MQTT 客户端库，支持多种编程语言，例如 Java、C++、Python 等。本文将重点介绍 Eclipse Paho 中消息的发送和接收。

一、消息的发送

1.1 发送消息的基本流程

在 Eclipse Paho 中，发送消息的基本流程如下：

1. 创建一个 MqttClient 对象。
2. 连接到 MQTT 服务端。
3. 创建一个 MqttMessage 对象，设置消息的内容和 QoS 等级。
4. 调用 MqttClient 的 publish() 方法发送消息。
5. 断开 MQTT 连接。

1.2 代码示例

下面是一个使用 Eclipse Paho 发送消息的 Java 代码示例：

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

public class MqttPublishSample {
    public static void main(String[] args) {
        String topic        = "MQTT Examples";
        String content      = "Message from MqttPublishSample";
        int qos             = 2;
        String broker       = "tcp://localhost:1883";
        String clientId     = "JavaSample";
        MemoryPersistence persistence = new MemoryPersistence();

        try {
            MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setCleanSession(true);
            sampleClient.connect(connOpts);
            MqttMessage message = new MqttMessage(content.getBytes());
            message.setQos(qos);
            sampleClient.publish(topic, message);
            sampleClient.disconnect();
            System.out.println("Message published");
        } catch(MqttException me) {
            System.out.println("Reason: " + me.getReasonCode());
            System.out.println("Message: " + me.getMessage());
            System.out.println("Loc: " + me.getLocalizedMessage());
            System.out.println("Cause: " + me.getCause());
            System.out.println("Excep: " + me);
            me.printStackTrace();
        }
    }
}

在这个代码示例中，我们创建了一个 MqttClient 对象，并连接到本地的 MQTT 服务器。然后，我们创建了一个 MqttMessage 对象，设置了消息的内容和 QoS 等级，并使用 publish() 方法将消息发送到指定的主题。

1.3 QoS 等级

在 MQTT 协议中，有三种不同的 QoS 等级：0、1 和 2。QoS 等级越高，消息的可靠性越高，但是网络负载也会相应增加。

- QoS 0：最多发送一次，没有确认，可能会出现消息丢失或重复。
- QoS 1：至少发送一次，确保消息至少被接收一次，但可能会出现重复消息。
- QoS 2：只发送一次，确保消息仅被接收一次，但网络负担较大。

二、消息的接收

2.1 接收消息的基本流程

在 Eclipse Paho 中，接收消息的基本流程如下：

1. 创建一个 MqttClient 对象。
2. 连接到 MQTT 服务端。
3. 创建一个 MqttCallback 对象，用于处理接收到的消息。
4. 订阅一个或多个主题。
5. 等待接收消息。
6. 断开 MQTT 连接。

2.2 代码示例

下面是一个使用 Eclipse Paho 接收消息的 Java 代码示例：

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

public class MqttSubscribeSample {
    public static void main(String[] args) {
        String topic        = "MQTT Examples";
        String broker       = "tcp://localhost:1883";
        String clientId     = "JavaSample";
        MemoryPersistence persistence = new MemoryPersistence();

        try {
            MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setCleanSession(true);
            sampleClient.connect(connOpts);

            sampleClient.setCallback(new MqttCallback() {
                public void connectionLost(Throwable cause) {
                    System.out.println("Connection lost");
                }

                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    System.out.println("Message received: " + new String(message.getPayload()));
                }

                public void deliveryComplete(IMqttDeliveryToken token) {
                    System.out.println("Delivery complete");
                }
            });

            sampleClient.subscribe(topic);

            // Wait for messages
            Thread.sleep(5000);

            sampleClient.disconnect();
            System.out.println("Disconnected");
        } catch(MqttException me) {
            System.out.println("Reason: " + me.getReasonCode());
            System.out.println("Message: " + me.getMessage());
            System.out.println("Loc: " + me.getLocalizedMessage());
            System.out.println("Cause: " + me.getCause());
            System.out.println("Excep: " + me);
            me.printStackTrace();
        } catch(InterruptedException e) {
            e.printStackTrace();
        }
    }
}

在这个代码示例中，我们创建了一个 MqttClient 对象，并连接到本地的 MQTT 服务器。然后，我们创建了一个 MqttCallback 对象，用于处理接收到的消息。在 MqttCallback 的 messageArrived() 方法中，我们可以处理接收到的消息。

最后，我们使用 subscribe() 方法订阅了一个主题，并等待 5 秒钟接收消息。在接收完消息之后，我们断开了 MQTT 连接。

总结

在本文中，我们介绍了使用 Eclipse Paho 发送和接收 MQTT 消息的基本流程，并提供了 Java 代码示例。同时，我们还介绍了 MQTT 的 QoS 等级。