云服务器特惠 99/年起
  1. 腾讯云优惠网首页
  2. 腾讯云产品

消息队列 MQTT 版 订阅消息

• 腾讯云产品

MQTT (Message Queuing Telemetry Transport) 是一种轻量级的、基于发布/订阅（Pub / Sub）模式的物联网消息传输协议。在 MQTT 通信模型中，订阅 (Subscription) 是客户端（订阅者）向服务器（Broker）表达其希望接收特定主题或一类主题消息的核心机制。

相关产品
消息队列 MQTT 版 产品页，含版本规格、价格与购买入口。

一、发布订阅模型

发布者将消息发送到特定的 Topic，Broker 负责将消息转发给所有订阅了该 Topic 的订阅者。

例如下场景中，温度传感器将温度读数发布到 “ temperature ” 主题，手机和后端业务系统订阅该主题后，都将获得温度传感器发布的全量数据。

二、订阅原理

1. 连接建立：客户端首先与 MQTT Broker 建立 TCP 连接，并发送 CONNECT 报文进行认证。

2. 发起订阅：连接成功后，客户端向 Broker 发送一个 SUBSCRIBE 报文。这个报文中包含了一个或多个它想要订阅的主题过滤器 (Topic Filter) 以及对应的服务质量等级 (QoS)。

3. 确认订阅：Broker 收到 SUBSCRIBE 报文后，会回复一个 SUBACK 订阅确认报文。该报文包含了为每个请求订阅的 Topic Filter 所授予的 QoS 等级（可能是客户端请求的，也可能是 Broker 根据自身策略授予的）。

4. 消息路由：此后，一旦有消息发布到匹配该 Topic Filter 的 Topic 上，Broker 就会立即将消息转发给该客户端。

5. 取消订阅：当客户端不再希望接收某些 Topic 的消息时，它会发送一个 UNSUBSCRIBE 报文，Broker 随后停止转发相关消息。

三、主题过滤器与通配符

客户端在订阅时使用的不是完全确定的 Topic 名，而是可以使用通配符的 Topic Filter，这使得它可以灵活地订阅多个 Topic。

通配符 说明 示例
+ 单层通配符，代表匹配一个且仅一个任意主题层级 订阅 sensor/+/temperature匹配 sensor/bedroom/temperature匹配 sensor/living_room/temperature不匹配 sensor/bedroom/upstairs/temperature (因为+只能匹配一层)不匹配 sensor/temperature (因为 sensor 和 temperature 之间缺少一层)
# 多层通配符，代表匹配零个或多个任意主题层级，必须是主题过滤器的最后一个字符 订阅 sensor/bedroom/#匹配 sensor/bedroom/temperature匹配 sensor/bedroom/humidity匹配 sensor/bedroom/light/intensity匹配 sensor/bedroom (# 可以匹配零层)不匹配 sensor/living_room/temperature (因为 bedroom 是固定的)

四、服务质量 (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 Messages)

发布一条消息时，可以设置retained参数等于true，Broker 会为这个 Topic 保存最新的这条保留消息。

当一个新的客户端订阅匹配的 Topic 时，Broker 会立即将这条最新的保留消息发送给它，而无需等待其下一次发布，这对于获取设备的最新状态非常有用。

例如，一个温度传感器每隔小时发布一次当前温度并设置为保留消息。任何新上线的客户端在订阅 sensor/temperature 后，能立刻收到当前的最新温度，而不必等待下一个小时。

六、使用示例

package com.tencent.tdmq.mqtt.example.paho.v5;import java.nio.ByteBuffer;import java.nio.charset.StandardCharsets;import java.util.List;import java.util.concurrent.TimeUnit;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 SubscriberQuickStart { 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 = "SubscriberQuickStart"; // 在控制台 –> 认证 Tab页创建账户, 复制用户名和密码 String username = "user0"; String password = "secret0"; // MQTT topic filters String[] topicFilters = new String[]{"home/test", "home/#", "home/+"}; int[] qos = new int[]{1, 1, 1}; 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) { byte[] payload = message.getPayload(); String content; if (4 == payload.length) { ByteBuffer buf = ByteBuffer.wrap(payload); content = String.valueOf(buf.getInt()); } else { content = new String(payload, StandardCharsets.UTF_8); } System.out.printf("Message arrived, topic=%s, QoS=%d content=[%s]%n", topic, message.getQos(), content); List<UserProperty> userProperties = message.getProperties().getUserProperties(); printUserProperties(userProperties); } @Override public void deliveryComplete(IMqttToken token) { System.out.println("Delivery complete for packet-id: " + token.getMessageId()); } @Override public void connectComplete(boolean reconnect, String serverURI) { System.out.println(reconnect ? "Reconnected" : "Connected" + " to " + serverURI); try { // Subscribe IMqttToken token = client.subscribe(topicFilters, qos); int[] reasonCodes = token.getReasonCodes(); for (int i = 0; i < reasonCodes.length; i++) { System.out.printf("Subscribed to topic %s with QoS=%d, Granted-QoS: %d%n", topicFilters[i], qos[i], reasonCodes[i]); } if (token.isComplete()) { List<UserProperty> userProperties = token.getResponseProperties().getUserProperties(); printUserProperties(userProperties); } } catch (MqttException e) { e.printStackTrace(); } } @Override public void authPacketArrived(int i, MqttProperties properties) { System.out.println("Received auth packet with id: " + i); } }); client.connect(options); TimeUnit.MINUTES.sleep(5); 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()); } } }}

相关内容：

  1. 消息队列 MQTT 版 发送消息 在 MQTT 的发布/订阅模式中，发布 (Publish) 是消息通信的起点。它是指客户端将消息发送到 Broker 的特定 Topic 的过程。Broker 负责接收发布的消息并将其转发给所有订阅了匹配 Topic 的客户端。发布功能是设备或应用向上汇报数据、发送指令、触发事件的核心机制。...
  2. 消息队列 MQTT 版 配置 SQL 过滤 MQTT 标准规范定义了 Topic Filter 的概念，允许订阅者基于 MQTT Topic Name 的层级结构和通配符来选择需要接收的消息。虽然 Topic 和通配符提供了较强的过滤能力，但在灰度发布、A/B 测试、系统升级等场景中，仅依赖 Topic 过滤仍无法满足更灵活的业务需求。...
  3. 消息队列 MQTT 版 Topic 与通配符订阅 在 MQTT 构建的物联网世界中，发布-订阅（Publish/Subscribe）模式实现了设备与应用之间的完全解耦，而支撑这一机制运转的核心，正是主题（Topic）。...
  4. 消息队列 MQTT 版 配置共享订阅 共享订阅是一种负载均衡的消费方式，属于同一个共享订阅组的多个客户端以轮询方式分配并接收订阅到的消息。因此，通常也称为消费者负载均衡。例如下图中，发布者发布了4条消息：M1、M2、M3、M4。客户端 Client-0 和 Client-1 以共享订阅的方式消费订阅的主题消息。这与默认的 MQTT Pub/Sub 不同，Client-0 仅消费到了 M1， M3； Client-1 仅消费到了 M2， M4。...
  5. 消息队列 MQTT 版 配置点对点订阅 除了 MQTT 标准协议定义的发布/订阅（Pub/Sub）消息模型外，消息队列 MQTT 版还支持点对点（Point to Point，简称 P2P）模式。...
  6. 消息队列 MQTT 版 MQTT 客户端开发注意事项 无论是通过公网还是通过内网接入 MQTT 集群, 传输层连接断开都是常态: 移动设备在基站间切换、 网络抖动、服务端版本发布等。因此 MQTT 客户端均需要设置连接断开后自动重连， 并配置合理的退避策略。...
  7. 消息队列 MQTT 版 用户属性（User Property） 用户属性（User Properties）是 MQTT 5.0 新增的特性,允许你在 MQTT 消息中添加自定义的键值对元数据。通过用户属性,可以以在不修改消息载荷的情况下传递额外信息,实现消息路由、数据格式标识、设备信息上报等功能。...
  8. 消息队列 MQTT 版 MQTT over QUIC 这篇说说腾讯云TDMQ MQTT版对MQTT over QUIC的支持，QUIC协议提供更低延迟、多路复用、连接迁移和弱网支持等特性，文档说明了使用限制、接入点端口、SDK支持及C/Java示例代码。 使用限制：仅支持IETF RFC9000 QUIC和单流模式，需1.4.0及以上集群实例。 接入点：QUIC公网端口14567（ALPN为mqtt），VPC接入地址在控制台查看。 SDK与示例：提供C、C++、Java。...
赞 (0)
上一篇 4天前
下一篇 4天前