MQTT 订阅选项的使用

简介: 在本文中,我们将重点介绍在 MQTT 中哪些订阅选项可供我们使用,以及它们的使用方法。

MQTT 发布/订阅模式介绍这篇博客中,我们已经了解到,我们需要先向服务端发起订阅,才能从服务端接收对应的消息。如果说订阅时指定的主题过滤器决定了服务端将向我们转发哪些主题下的消息,那么订阅选项则是允许我们进一步定制服务端的转发行为。

在本文中,我们将重点介绍在 MQTT 中哪些订阅选项可供我们使用,以及它们的使用方法。

订阅选项

在 MQTT 中,一个订阅由一个主题过滤器和对应的订阅选项组成。所以理论上,我们可以为每个订阅都设置不同的订阅选项。

MQTT 5.0 提供了 4 个订阅选项,分别是 QoS、No Local、Retain As Published、Retain Handling,而 MQTT 3.1.1 则仅提供了 QoS 这一个订阅选项。不过这些 MQTT 5.0 新增的订阅选项的默认行为,仍与 MQTT 3.1.1 保持一致,如果你正准备从 MQTT 3.1.1 升级到 MQTT 5.0,这会非常地友好。

现在,让我们一起看看这些订阅选项的作用吧。

QoS

QoS 是最常用的一个订阅选项,它表示服务端在向订阅端发送消息时可以使用的最大 QoS 等级。

客户端可能会在订阅时指定一个小于 2 的 QoS,因为它的实现不支持 QoS 1 或者 QoS 2。而如果服务端支持的最大 QoS 小于客户端订阅时请求的最大 QoS,那么显然服务端将无法满足客户端的要求,这时服务端就会通过订阅的响应报文(SUBACK)告知订阅端最终授予的最大 QoS 等级,订阅端可以自行评估是否接受并继续通信。

image.png

一个简单的计算公式:

服务端最终授予的最大 QoS = min ( 服务端支持的最大 QoS, 客户端请求的最大 QoS )

但是,我们在订阅时请求的最大 QoS,并不能限制发布端发布消息时使用的 QoS。当我们订阅时请求的最大 QoS,小于消息发布时的 QoS 时,为了尽可能地投递消息,服务端不会忽略这些消息,而是会在转发时对这些消息的 QoS 进行降级处理。

image.png

同样,我们也有一个简单的计算公式:

消息被转发时的 QoS = min ( 消息原始的 QoS, 服务端最终授予的最大 QoS )

No Local

No Local 只有 0 和 1 两个可取值,为 1 表示服务端不能将消息转发给发布这个消息的客户端,为 0 则相反。

这个选项通常被用在桥接场景中。桥接本质上是两个 MQTT Server 建立了一个 MQTT 连接,然后相互订阅一些主题,Server 将客户端的消息转发给另一个 Server,而另一个 Server 则可以将消息继续转发给它的客户端。

image.png

那么最简单的一个例子,我们假设两个 MQTT Server 分别是 Server A 和 Server B,它们分别向对方订阅了 # 主题。现在,Server A 将一些来自客户端的消息转发给了 Server B,而当 Server B 查找匹配的订阅时,Server A 也会位于其中。如果 Server B 将消息转发给了 Server A,那么同样 Server A 在收到消息后又会把它们再次转发给 Server B,这样就陷入了无休止的转发风暴。

而如果 Server A 和 Server B 在订阅 # 主题的同时,将 No Local 选项设置为 1,就可以完美地避免这个问题。

Retain As Published

Retain As Published 同样只有 0 和 1 两个可取值,为 1 表示服务端在向此订阅转发应用消息时需要保持消息中的 Retain 标识不变,为 0 则表示必须清除。

Retain As Published 与 No Local 一样,同样也是主要适用于桥接场景。我们知道当服务端收到一条保留消息时,除了将它存储起来,还会将它像普通消息一样转发给当前已经存在的订阅者,并且在转发时会清除消息的 Retain 标识。

这在桥接场景下带来了一些问题。我们继续沿用前面的设定,当 Server A 将保留消息转发给 Server B 时,由于消息中的 Retain 标识已经被清除,Server B 将不会知道这原本是一条保留消息,自然不会再存储它。这就导致了保留消息无法跨桥接使用。

那么在 MQTT 5.0 中,我们可以让桥接的服务端在订阅时将 Retain As Published 选项设置为 1,来解决这个问题。

image.png

Retain Handling

Retain Handling 这个订阅选项被用来向服务端指示当订阅建立时,是否需要发送保留消息。

我们知道默认情况下,只要订阅建立,那么服务端中与订阅匹配的保留消息就会下发。

但某些时候,客户端可能并不想接收保留消息,比如客户端在连接时复用了会话,但是客户端无法确认上一次连接中是否成功创建了订阅,所以它可能会再次发起订阅。如果订阅已经存在,那么可能保留消息已经被消费过了,也可能服务端已经在会话中缓存了一些离线期间到达的消息,这时客户端可能并不希望服务端发布保留消息。

另外,客户端也可能在任何时刻都不想收到保留消息,即使是第一次订阅。比如我们将开关状态作为保留消息发送,但对某个订阅端来说,开关事件将触发一些操作,那么在这种情况下不发送保留消息是很有用的。

这三种不同的行为,我们可以通过 Retain Handling 来选择。

  • 将 Retain Handling 设置为 0,表示只要订阅建立,就发送保留消息;

  • 将 Retain Handling 设置为 1,表示只有建立全新的订阅而不是重复订阅时,才发送保留消息;

  • 将 Retain Handling 设置为 2,表示订阅建立时不要发送保留消息。

演示

订阅选项 QoS 的演示

  1. 在 Web 浏览器上访问 MQTTX Web

  2. 创建一个使用 WebSocket 的 MQTT 连接,并且连接免费的 公共 MQTT 服务器:

    MQTTX

  3. 连接成功后,我们订阅主题 mqttx_4299c767/demo,并指定 QoS 为 0。由于公共服务器可能同时被很多人使用,为了避免主题与别人重复,我们可以将 Client ID 作为主题前缀:

    Subscribe to the topic "mqttx_4299c767/demo"

  4. 订阅成功后,我们向主题 mqttx_4299c767/demo 发布一条 QoS 1 消息,这时我们将看到,我们发出的是 QoS 1 消息,但收到的却是 QoS 0 消息,这说明发生了 QoS 降级:

    Publish a QoS 1 message

订阅选项 No Local 的演示

  1. 在 Web 浏览器上访问 MQTTX Web。

  2. 创建一个使用 WebSocket 的 MQTT 连接,并且连接免费的公共 MQTT 服务器。

  3. 连接成功后,我们订阅主题 mqttx_4299c767/demo,并且将 No Local 设置为 true:

    Subscribe to the topic "mqttx_4299c767/demo"

  4. 订阅成功后,与前面 QoS 的演示一样,我们还是由订阅端自己来发布消息,但这一次我们会发现订阅端将无法收到消息:

    Publish MQTT Message

订阅选项 Retain As Published 的演示

  1. 在 Web 浏览器上访问 MQTTX Web。

  2. 创建一个使用 WebSocket 的 MQTT 连接,并且连接免费的公共 MQTT 服务器。

  3. 连接成功后,我们先订阅主题 mqttx_4299c767/rap0,并且将 Retain As Published 设置为 false,然后订阅主题 mqttx_4299c767/rap1,并且将 Retain As Published 设置 true:

    Subscribe to the topic "mqttx_4299c767/rap0"

    Subscribe to the topic "mqttx_4299c767/rap1"

  4. 订阅成功后,我们分别向主题 mqttx_4299c767/rap0mqttx_4299c767/rap1 发布一条保留消息,我们将看到前者收到的消息中 Retain 标识被清除,而后者收到的消息中 Retain 标识被保留:

    Receive messages

订阅选项 Retain Handling 的演示

  1. 在 Web 浏览器上访问 MQTTX Web。

  2. 创建一个使用 WebSocket 的 MQTT 连接,并且连接免费的公共 MQTT 服务器。

  3. 连接成功后,我们先向主题 mqttx_4299c767/rh 发布一条保留消息。然后订阅主题 mqttx_4299c767/rh,并且将 Retain Handling 设置为 0:

    Publish a retained message to the topic "mqttx_4299c767/rh"

  4. 订阅成功后,我们将收到服务端发送的保留消息:

    Receive the retained message

  5. 取消当前订阅,重新订阅主题 mqttx_4299c767/rh,并且将 Retain Handling 设置为 2。不过这一次订阅成功后,我们将不会收到服务端发送的保留消息:

    Retain Handling set to 2

在 MQTTX 中,我们没有办法演示 Retain Handling 设置为 1 时的效果。不过你可以在 这里 获取订阅选项的 Python 示例代码。

版权声明: 本文为 EMQ 原创,转载请注明出处。
原文链接:https://www.emqx.com/zh/blog/an-introduction-to-subscription-options-in-mqtt

相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
消息中间件 Java 物联网
一文搞懂MQTT,如何在SpringBoot中使用MQTT实现消息的订阅和发布
之前介绍了RabbitMQ以及如何在SpringBoot项目中整合使用RabbitMQ,看过的朋友都说写的比较详细,希望再总结一下目前比较流行的MQTT。所以接下来,就来介绍什么MQTT?它在IoT中有着怎样的作用?如何在项目中使用MQTT?
13464 5
一文搞懂MQTT,如何在SpringBoot中使用MQTT实现消息的订阅和发布
|
3天前
|
消息中间件 测试技术
通过轻量消息队列(原MNS)主题HTTP订阅+ARMS实现自定义数据多渠道告警
轻量消息队列(原MNS)以其简单队列模型、轻量化协议及按量后付费模式,成为阿里云产品间消息传输首选。本文通过创建主题、订阅、配置告警集成等步骤,展示了该产品在实际应用中的部分功能,确保消息的可靠传输。
14 2
|
监控 物联网 API
【.NET+MQTT】.NET6 环境下实现MQTT通信,以及服务端、客户端的双边消息订阅与发布的代码演示
MQTT广泛应用于工业物联网、智能家居、各类智能制造或各类自动化场景等。MQTT是一个基于客户端-服务器的消息发布/订阅传输协议,在很多受限的环境下,比如说机器与机器通信、机器与物联网通信等。好了,科普的废话不多说,下面直接通过.NET环境来实现一套MQTT通信demo,实现服务端与客户端的双边消息发布与订阅的功能和演示。
1274 0
【.NET+MQTT】.NET6 环境下实现MQTT通信,以及服务端、客户端的双边消息订阅与发布的代码演示
|
消息中间件 存储 负载均衡
两个实验让我彻底弄懂了「订阅关系一致」
这篇文章,笔者想聊聊 RocketMQ 最佳实践之一:**保证订阅关系一致**。 订阅关系一致指的是同一个消费者 Group ID 下所有 Consumer 实例所订阅的 Topic 、Tag 必须完全一致。 如果订阅关系不一致,消息消费的逻辑就会混乱,甚至导致消息丢失。
两个实验让我彻底弄懂了「订阅关系一致」
|
传感器 负载均衡 物联网
MQTT v5共享订阅是怎么回事?如何使用共享订阅提高消息订阅的灵活性和可伸缩性?
MQTT v5共享订阅是怎么回事?如何使用共享订阅提高消息订阅的灵活性和可伸缩性?
477 1
|
6月前
|
消息中间件 Java RocketMQ
MQ产品使用合集之在同一个 Java 进程内建立三个消费对象并设置三个消费者组订阅同一主题和标签的情况下,是否会发生其中一个消费者组无法接收到消息的现象
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
118 1
|
消息中间件 安全 Go
动态订阅时 rocketmq-client-go 代码有map并发bug
动态订阅时 rocketmq-client-go 代码有map并发bug
65 2
|
6月前
|
传感器 监控 网络协议
MQTT 发布、订阅模式介绍
【2月更文挑战第17天】
423 6
MQTT 发布、订阅模式介绍
|
6月前
|
存储 负载均衡 安全
MQTT常见问题之MQTT使用共享订阅失败如何解决
MQTT(Message Queuing Telemetry Transport)是一个轻量级的、基于发布/订阅模式的消息协议,广泛用于物联网(IoT)中设备间的通信。以下是MQTT使用过程中可能遇到的一些常见问题及其答案的汇总:
|
6月前
|
消息中间件 Java
RabbitMQ中的消息发布-订阅模式是什么?如何实现?
RabbitMQ中的消息发布-订阅模式是什么?如何实现?
167 0