消息队列 MQTT 版使用客户端发送消息

在 MQTT 的发布/订阅模式中，发布 (Publish) 是消息通信的起点。它是指客户端将消息发送到 Broker 的特定 Topic 的过程。Broker 负责接收发布的消息并将其转发给所有订阅了匹配 Topic 的客户端。发布功能是设备或应用向上汇报数据、发送指令、触发事件的核心机制。
发布原理
1. 建立连接：客户端首先与 MQTT 代理建立 TCP/IP 连接，并发送 CONNECT 报文进行认证握手，成功后建立会话。
2. 准备消息：客户端构建 PUBLISH 报文，即准备要发送的消息，包括 Topic（消息目的地）、Payload（消息数据） 和一些控制属性（QoS 等级、保留标志等）。
3. 发送消息并确认：客户端向代理发送一个 PUBLISH 报文，其中包含了上述所有消息信息。Broker 根据报文中的 QoS 等级，按规则向发布客户端回复确认报文（如 PUBACK/PUBREC），完成发送方的可靠性保证。
4. 路由消息： Broker 收到消息后，查找订阅匹配该 Topic 的客户端，并将消息转发给它们。对于每个订阅者，将依据其订阅时指定的 QoS 等级再次进行消息传递。
服务质量 (QoS)
QoS（Quality of Service）指代消息传输的服务质量，每条消息都可以在发送时单独设置 QoS。它包括以下级别：
QoS 等级
工作流程
优点
缺点
适用场景
QoS = 0
（最多分发一次）
发送者将消息发送出去，不要求接收方进行确认
传输速度最快，开销最小
消息可能会丢失，例如网络故障、接收方不在线情况
非关键性的数据，允许偶尔丢失，例如周期性的传感器数据（温度、湿度），丢失下一个数据点很快会补上。
QoS = 1
（至少到达一次）
1. 发送者发送消息并保留一个副本。
2. 接收者收到消息后，必须回复一个 PUBACK (发布确认) 报文。
3. 发送者收到 PUBACK 后，才会丢弃消息副本。
4. 如果发送者在合理时间内未收到 PUBACK，它会重新发送该消息。
保证了消息不会丢失
可能导致消息重复
需要保证消息必达，但可以接受偶尔重复的场景，例如控制指令（“开关灯”），重复执行一次可能也没问题。
QoS = 2
（仅到达一次）
1. PUBLISH：发送者发送消息，并保留副本。
2. PUBREC：接收者收到后回复一个“已收到”确认。如果发送者没收到 PUBREC，会重发 PUBLISH。
3. PUBREL：发送者收到 PUBREC 后，发送一个“发布释放”报文，并可以丢弃消息副本。它现在只需要等待最终确认。
4. PUBCOMP：接收者收到 PUBREL 后，回复一个“发布完成”确认。之后才将消息交付给应用。发送者收到 PUBCOMP，流程结束。如果任何一方没收到应答，都会重发上一条报文。
既保证了消息不丢失，又确保了不会重复
速度最慢，开销最大
对可靠性和准确性要求极高的关键业务，例如计费系统、金融交易、关键状态同步，任何重复或丢失都会造成严重后果。
保留消息 (Retained Message)
发布一条消息时，可以设置retained参数等于true，Broker 会为这个 Topic 保存最新的这条保留消息。
当一个新的客户端订阅匹配的 Topic 时，Broker 会立即将这条最新的保留消息发送给它，而无需等待其下一次发布，这对于获取设备的最新状态非常有用。
例如，一个温度传感器每隔小时发布一次当前温度并设置为保留消息。任何新上线的客户端在订阅 sensor/temperature 后，能立刻收到当前的最新温度，而不必等待下一个小时。
遗嘱消息 (Last Will and Testament - LWT)
LWT 不是直接的发布动作，而是在客户端连接时预先定义好的一条消息。如果客户端意外断开连接（网络故障、心跳超时），代理将自动代表该客户端发布这条遗言消息到指定的主题。常用于通知其他客户端某设备已异常离线。
发送消息示例代码
﻿
package com.tencent.tdmq.mqtt.example.paho.v5;
﻿
import java.nio.charset.StandardCharsets;
import java.util.List;
import org.eclipse.paho.mqttv5.client.IMqttToken;
import org.eclipse.paho.mqttv5.client.MqttCallback;
import org.eclipse.paho.mqttv5.client.MqttClient;
import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
import org.eclipse.paho.mqttv5.client.MqttDisconnectResponse;
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.eclipse.paho.mqttv5.common.packet.MqttProperties;
import org.eclipse.paho.mqttv5.common.packet.UserProperty;
﻿
public class PublisherQuickStart {
﻿
    public static void main(String[] args) throws MqttException, InterruptedException {
    
        // 从MQTT控制台获取接入点:
        // 通过Private Link实现VPC网络打通的用户, 使用内网接入点;
        // 通过公网访问的用户, 确保公网安全策略允许, 程序运行机器有公网接入;
        String serverUri = "tcp://mqtt-xxx.mqtt.tencenttdmq.com:1883";
﻿
        // 合法的Client Identifier包含 数字0-9, 小写英文字母a-z, 以及大写英文字母A-Z, 总长度为1-23个字符
        // 参考 https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901059
        String clientId = "PublisherQuickStart";
﻿
        // 在控制台 --> 认证 Tab页创建账户, 复制用户名和密码
        String username = "user0";
        String password = "secret0";
﻿
        // MQTT topic
        String topicName = "home/test";
﻿
        MqttClient client = new MqttClient(serverUri, clientId, new MemoryPersistence());
        client.setTimeToWait(3000);
        MqttConnectionOptions options = new MqttConnectionOptions();
        options.setUserName(username);
        options.setPassword(password.getBytes(StandardCharsets.UTF_8));
        options.setCleanStart(true);
        options.setAutomaticReconnect(true);
﻿
        client.setCallback(new MqttCallback() {
            @Override
            public void disconnected(MqttDisconnectResponse response) {
                System.out.println("Disconnected: " + response.getReasonString());
            }
﻿
            @Override
            public void mqttErrorOccurred(MqttException e) {
                e.printStackTrace();
            }
﻿
            @Override
            public void messageArrived(String topic, MqttMessage message) {
            }
﻿
            @Override
            public void deliveryComplete(IMqttToken token) {
                System.out.println("Delivery completed: packet-id=" + token.getMessageId() +
                    ", reason-code=" + token.getResponse().getReasonCodes()[0]);
﻿
                List<UserProperty> userProperties = token.getResponseProperties().getUserProperties();
                printUserProperties(userProperties);
            }
﻿
            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                System.out.println(reconnect ? "Reconnected" : "Connected" + " to " + serverURI);
            }
﻿
            @Override
            public void authPacketArrived(int i, MqttProperties properties) {
                System.out.println("Received auth packet with id: " + i);
            }
        });
﻿
        client.connect(options);
﻿
        int total = 16;
        for (int i = 0; i < total; i++) {
            String msg = "Hello MQTT " + i;
            MqttMessage message = new MqttMessage(msg.getBytes(StandardCharsets.UTF_8));
            message.setQos(1);
            System.out.printf("Prepare to publish message %d%n", i);
            client.publish(topicName, message);
            System.out.printf("Published message %d%n", i);
        }
﻿
        client.disconnect();
﻿
        client.close();
    }
﻿
    static void printUserProperties(List<UserProperty> userProperties) {
        if (null != userProperties) {
            for (UserProperty userProperty : userProperties) {
                System.out.printf("User property: %s = %s%n", userProperty.getKey(), userProperty.getValue());
            }
        }
    }
}

QoS 等级	工作流程	优点	缺点	适用场景
QoS = 0 （最多分发一次）	发送者将消息发送出去，不要求接收方进行确认	传输速度最快，开销最小	消息可能会丢失，例如网络故障、接收方不在线情况	非关键性的数据，允许偶尔丢失，例如周期性的传感器数据（温度、湿度），丢失下一个数据点很快会补上。
QoS = 1 （至少到达一次）	1. 发送者发送消息并保留一个副本。 2. 接收者收到消息后，必须回复一个 PUBACK (发布确认) 报文。 3. 发送者收到 PUBACK 后，才会丢弃消息副本。 4. 如果发送者在合理时间内未收到 PUBACK，它会重新发送该消息。	保证了消息不会丢失	可能导致消息重复	需要保证消息必达，但可以接受偶尔重复的场景，例如控制指令（“开关灯”），重复执行一次可能也没问题。
QoS = 2 （仅到达一次）	1. PUBLISH：发送者发送消息，并保留副本。 2. PUBREC：接收者收到后回复一个“已收到”确认。如果发送者没收到 PUBREC，会重发 PUBLISH。 3. PUBREL：发送者收到 PUBREC 后，发送一个“发布释放”报文，并可以丢弃消息副本。它现在只需要等待最终确认。 4. PUBCOMP：接收者收到 PUBREL 后，回复一个“发布完成”确认。之后才将消息交付给应用。发送者收到 PUBCOMP，流程结束。如果任何一方没收到应答，都会重发上一条报文。	既保证了消息不丢失，又确保了不会重复	速度最慢，开销最大	对可靠性和准确性要求极高的关键业务，例如计费系统、金融交易、关键状态同步，任何重复或丢失都会造成严重后果。

使用客户端发送消息

本页目录：

发布原理

服务质量 (QoS)

保留消息 (Retained Message)

遗嘱消息 (Last Will and Testament - LWT)

发送消息示例代码