- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
各位开发者,
我正在查看 Paho MQTT 客户端(用于 Python),我想我了解发布者的 QoS 设置,例如传感器或任何数据源确实有意义 - 毕竟您希望能够确保(由代理/服务器)收到消息。
我也认为从订阅者的角度来看,请求例如QoS“2”,以确保 MQTT 服务器确实将每条消息发送给订阅者,但我正在努力解决这个问题:订阅者似乎没有办法表示成功(或不)处理收到的消息消息,换句话说,似乎缺少某种明确确认的方式?
我想订阅一个主题并成功处理每个数据点,例如通过存储到数据库。因此,我需要适应订户“飞行中”失败的情况,即在收到来自代理的消息之后但在成功处理(存储到数据库)之前。
现在,如果订阅者(固定的 client_id)在处理数据时失败,它将重新启动,然后重新连接到 MQTT 代理,代理通过 id 识别这个特定的客户端并再次开始推送消息 - 从订阅者断开连接后的下一条消息开始- 就代理而言,它确实成功传递了最后一条消息(不知道,订阅者崩溃了)。
如果上述假设成立,那么我最好不要使用固定的 client_id,而是使用“clean_session”和随机的 client_id;这样,代理开始传递所有消息,这些消息是为特定主题保留的。这当然会将跟踪成功处理的消息的责任放在订阅者身上。
需要这样做吗?或者是否有一种方法可以明确确认订户成功处理消息,以便代理可以重新传输应该需要 - 我对 Paho Python 库特别感兴趣。
提前致谢!
相关代码:
def _handle_on_message(self, message):
matched = False
with self._callback_mutex:
try:
topic = message.topic
except UnicodeDecodeError:
topic = None
if topic is not None:
for callback in self._on_message_filtered.iter_match(message.topic):
with self._in_callback:
callback(self, self._userdata, message)
matched = True
if matched == False and self.on_message:
with self._in_callback:
self.on_message(self, self._userdata, message)
来源:https://github.com/eclipse/paho.mqtt.python/blob/v1.3.1/src/paho/mqtt/client.py#L2631
@hardlib 确实是正确的——当回调函数内部失败时,客户端代码不会向代理确认
一些用于说明的代码
on_message
订阅者回调def on_message(client, userdata, msg):
print(msg.topic+" "+str(msg.payload)+ " mid:" + str(msg.mid))
num = ''.join(x for x in str(msg.payload) if x.isdigit())
if int(num) % 3 == 0:
print("Going to show myself out now (sys.exit(1))")
time.sleep(1)
sys.exit(1)
while True:
count += 1
logging.debug("At: " + str(count))
msg = "message: {counter}".format(counter=count)
mqttc.publish("paho/stacko", msg, qos=2, retain=False)
root@14f00c2576b2:/usr/src/app# python publisher.py
DEBUG:root:Sending CONNECT (u0, p0, wr0, wq0, wf0, c1, k60) client_id=b''
DEBUG:root:At: 1
DEBUG:root:Sending PUBLISH (d0, q2, r0, m1), 'b'paho/stacko'', ... (10 bytes)
DEBUG:root:Received CONNACK (0, 0)
DEBUG:root:Received PUBREC (Mid: 1)
DEBUG:root:Sending PUBREL (Mid: 1)
DEBUG:root:Received PUBCOMP (Mid: 1)
DEBUG:root:At: 2
DEBUG:root:Sending PUBLISH (d0, q2, r0, m2), 'b'paho/stacko'', ... (10 bytes)
DEBUG:root:Received PUBREC (Mid: 2)
DEBUG:root:Sending PUBREL (Mid: 2)
DEBUG:root:Received PUBCOMP (Mid: 2)
DEBUG:root:At: 3
DEBUG:root:Sending PUBLISH (d0, q2, r0, m3), 'b'paho/stacko'', ... (10 bytes)
DEBUG:root:Received PUBREC (Mid: 3)
DEBUG:root:Sending PUBREL (Mid: 3)
DEBUG:root:Received PUBCOMP (Mid: 3)
DEBUG:root:At: 4
DEBUG:root:Sending PUBLISH (d0, q2, r0, m4), 'b'paho/stacko'', ... (10 bytes)
DEBUG:root:Received PUBREC (Mid: 4)
DEBUG:root:Sending PUBREL (Mid: 4)
DEBUG:root:Received PUBCOMP (Mid: 4)
DEBUG:root:At: 5
DEBUG:root:Sending PUBLISH (d0, q2, r0, m5), 'b'paho/stacko'', ... (10 bytes)
DEBUG:root:Received PUBREC (Mid: 5)
DEBUG:root:Sending PUBREL (Mid: 5)
DEBUG:root:Received PUBCOMP (Mid: 5)
DEBUG:root:At: 6
DEBUG:root:Sending PUBLISH (d0, q2, r0, m6), 'b'paho/stacko'', ... (10 bytes)
DEBUG:root:Received PUBREC (Mid: 6)
DEBUG:root:Sending PUBREL (Mid: 6)
DEBUG:root:Received PUBCOMP (Mid: 6)
DEBUG:root:At: 7
DEBUG:root:Sending PUBLISH (d0, q2, r0, m7), 'b'paho/stacko'', ... (10 bytes)
DEBUG:root:Received PUBREC (Mid: 7)
DEBUG:root:Sending PUBREL (Mid: 7)
DEBUG:root:Received PUBCOMP (Mid: 7)
DEBUG:root:At: 8
DEBUG:root:Sending PUBLISH (d0, q2, r0, m8), 'b'paho/stacko'', ... (10 bytes)
DEBUG:root:Received PUBREC (Mid: 8)
DEBUG:root:Sending PUBREL (Mid: 8)
DEBUG:root:Received PUBCOMP (Mid: 8)
DEBUG:root:At: 9
DEBUG:root:Sending PUBLISH (d0, q2, r0, m9), 'b'paho/stacko'', ... (10 bytes)
DEBUG:root:Received PUBREC (Mid: 9)
DEBUG:root:Sending PUBREL (Mid: 9)
DEBUG:root:Received PUBCOMP (Mid: 9)
DEBUG:root:At: 10
DEBUG:root:Sending PUBLISH (d0, q2, r0, m10), 'b'paho/stacko'', ... (11 bytes)
DEBUG:root:Received PUBREC (Mid: 10)
DEBUG:root:Sending PUBREL (Mid: 10)
DEBUG:root:Received PUBCOMP (Mid: 10)
DEBUG:root:At: 11
DEBUG:root:Sending PUBLISH (d0, q2, r0, m11), 'b'paho/stacko'', ... (11 bytes)
DEBUG:root:Received PUBREC (Mid: 11)
DEBUG:root:Sending PUBREL (Mid: 11)
DEBUG:root:Received PUBCOMP (Mid: 11)
root@ca7dcaaed68f:/usr/src/app# python subscriber.py
DEBUG:root:Sending CONNECT (u0, p0, wr0, wq0, wf0, c0, k60) client_id=b'client_02'
DEBUG:root:Received CONNACK (0, 0)
DEBUG:root:Connected
Connected with result code 0
DEBUG:root:Sending SUBSCRIBE (d0) [(b'#', 2)]
DEBUG:root:Received SUBACK
DEBUG:root:Received PUBLISH (d0, q2, r0, m1), 'paho/stacko', ... (10 bytes)
DEBUG:root:Sending PUBREC (Mid: 1)
DEBUG:root:Received PUBREL (Mid: 1)
DEBUG:root:Message!!!! b'message: 2'
paho/stacko b'message: 2' mid:1
Num: 2
DEBUG:root:Sending PUBCOMP (Mid: 1)
DEBUG:root:Received PUBLISH (d0, q2, r0, m2), 'paho/stacko', ... (10 bytes)
DEBUG:root:Sending PUBREC (Mid: 2)
DEBUG:root:Received PUBREL (Mid: 2)
DEBUG:root:Message!!!! b'message: 3'
paho/stacko b'message: 3' mid:2
Num: 3
Going to show myself out now (sys.exit(1))
root@ca7dcaaed68f:/usr/src/app# python subscriber.py
DEBUG:root:Sending CONNECT (u0, p0, wr0, wq0, wf0, c0, k60) client_id=b'client_02'
DEBUG:root:Received CONNACK (1, 0)
DEBUG:root:Connected
Connected with result code 0
DEBUG:root:Sending SUBSCRIBE (d0) [(b'#', 2)]
DEBUG:root:Received PUBREL (Mid: 2)
DEBUG:root:Received PUBLISH (d0, q2, r0, m3), 'paho/stacko', ... (10 bytes)
DEBUG:root:Sending PUBREC (Mid: 3)
DEBUG:root:Received PUBLISH (d0, q2, r0, m4), 'paho/stacko', ... (10 bytes)
DEBUG:root:Sending PUBREC (Mid: 4)
DEBUG:root:Received SUBACK
DEBUG:root:Received PUBREL (Mid: 3)
DEBUG:root:Message!!!! b'message: 4'
paho/stacko b'message: 4' mid:3
Num: 4
DEBUG:root:Sending PUBCOMP (Mid: 3)
DEBUG:root:Received PUBREL (Mid: 4)
DEBUG:root:Message!!!! b'message: 5'
paho/stacko b'message: 5' mid:4
Num: 5
DEBUG:root:Sending PUBCOMP (Mid: 4)
DEBUG:root:Received PUBLISH (d0, q2, r0, m5), 'paho/stacko', ... (10 bytes)
DEBUG:root:Sending PUBREC (Mid: 5)
DEBUG:root:Received PUBREL (Mid: 5)
DEBUG:root:Message!!!! b'message: 6'
paho/stacko b'message: 6' mid:5
Num: 6
Going to show myself out now (sys.exit(1))
root@ca7dcaaed68f:/usr/src/app# python subscriber.py
DEBUG:root:Sending CONNECT (u0, p0, wr0, wq0, wf0, c0, k60) client_id=b'client_02'
DEBUG:root:Received CONNACK (1, 0)
DEBUG:root:Connected
Connected with result code 0
DEBUG:root:Sending SUBSCRIBE (d0) [(b'#', 2)]
DEBUG:root:Received PUBREL (Mid: 2)
DEBUG:root:Received PUBREL (Mid: 5)
DEBUG:root:Received SUBACK
DEBUG:root:Received PUBLISH (d0, q2, r0, m6), 'paho/stacko', ... (10 bytes)
DEBUG:root:Sending PUBREC (Mid: 6)
DEBUG:root:Received PUBREL (Mid: 6)
DEBUG:root:Message!!!! b'message: 7'
paho/stacko b'message: 7' mid:6
Num: 7
DEBUG:root:Sending PUBCOMP (Mid: 6)
DEBUG:root:Received PUBREL (Mid: 2)
DEBUG:root:Received PUBLISH (d0, q2, r0, m7), 'paho/stacko', ... (10 bytes)
DEBUG:root:Sending PUBREC (Mid: 7)
DEBUG:root:Received PUBREL (Mid: 7)
DEBUG:root:Message!!!! b'message: 8'
paho/stacko b'message: 8' mid:7
Num: 8
DEBUG:root:Sending PUBCOMP (Mid: 7)
DEBUG:root:Received PUBLISH (d0, q2, r0, m8), 'paho/stacko', ... (10 bytes)
DEBUG:root:Sending PUBREC (Mid: 8)
DEBUG:root:Received PUBREL (Mid: 8)
DEBUG:root:Message!!!! b'message: 9'
paho/stacko b'message: 9' mid:8
Num: 9
Going to show myself out now (sys.exit(1))
root@ca7dcaaed68f:/usr/src/app# python subscriber.py
DEBUG:root:Sending CONNECT (u0, p0, wr0, wq0, wf0, c0, k60) client_id=b'client_02'
DEBUG:root:Received CONNACK (1, 0)
DEBUG:root:Connected
Connected with result code 0
DEBUG:root:Sending SUBSCRIBE (d0) [(b'#', 2)]
DEBUG:root:Received PUBREL (Mid: 2)
DEBUG:root:Received PUBREL (Mid: 5)
DEBUG:root:Received PUBREL (Mid: 8)
DEBUG:root:Received SUBACK
DEBUG:root:Received PUBLISH (d0, q2, r0, m9), 'paho/stacko', ... (11 bytes)
DEBUG:root:Sending PUBREC (Mid: 9)
DEBUG:root:Received PUBREL (Mid: 9)
DEBUG:root:Message!!!! b'message: 10'
paho/stacko b'message: 10' mid:9
Num: 10
DEBUG:root:Sending PUBCOMP (Mid: 9)
DEBUG:root:Received PUBREL (Mid: 5)
DEBUG:root:Received PUBLISH (d0, q2, r0, m10), 'paho/stacko', ... (11 bytes)
DEBUG:root:Sending PUBREC (Mid: 10)
DEBUG:root:Received PUBREL (Mid: 10)
DEBUG:root:Message!!!! b'message: 11'
paho/stacko b'message: 11' mid:10
Num: 11
DEBUG:root:Sending PUBCOMP (Mid: 10)
DEBUG:root:Received PUBREL (Mid: 2)
MQTT(至少是 Mosquitto)在持久性方面确实按预期工作:如果客户端重新连接,它可以“追赶”自上次连接以来错过的那些消息。然而,即使为订阅者和发布者都设置了 qos=2
,崩溃前的最后一条消息也不会被重新处理
最佳答案
如果您的客户端在处理消息时有可能失败,您可以在开始处理之前将消息存储在某个地方(可能在磁盘或数据库中)。
然后您可以在客户端重新启动时检查此存储并尝试再次处理它。如果您在重新连接(使用固定的 client_id)之前执行此操作,那么您不必担心在尝试处理失败的消息时传递新消息。
编辑:
还要更详细地查看代码,对于 QOS2,传递确认的最后一段似乎仅在 on_message
完成后发送,因此如果您在处理消息时崩溃 on_message
然后代理应该重新传递该消息。
关于python - PAHO MQTT Python 客户端 - 缺少确认,保证为订户交付,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/46998496/
我想知道如果我只连接到 MQTT 服务器(不发送或接收任何消息),一个月内(大约)会消耗多少数据。 我需要计算它来衡量我应该为物联网应用程序中使用的 sim 卡充值什么数据计划。 谢谢 最佳答案 编辑
对于单个消息,在 mqtt 中向主题(代理或接收器)发送(发布)数据是否有任何限制? 提前致谢.. 最佳答案 IBM 写了一个 article关于 MQTT。 A payload up to 256
我一直在阅读有关 MQTT 的资料,我知道它使用 TCP 进行网络传输。因此,如果我有一个将发送订阅请求的移动应用程序,我假设这将是一个全双工连接,以便客户端可以收到传入推送数据的通知。 那么如何提高
最近看了一篇物联网文章 http://www.eclipse.org/community/eclipse_newsletter/2014/february/article2.php 在这里他说,通过
我有一个项目涉及与众多客户进行后端通信。我正在寻找要使用的最佳协议(protocol)。 MQTT 适合我的项目吗? 最佳答案 Mqtt 最适合涉及大量用户的项目。目标是提供一个围绕发布/订阅模型优化
是否允许在 MQTT 主题中使用多个 #? 例如: sensors/data/#/something/else/# 最佳答案 没有 MQTT 规范(v3 和 v5)规定主题只能在主题中使用一次“#”,
如果 MQTT 已经是一个轻量级协议(protocol),而且它使用的功率和带宽很少,那么为什么我们有 MQTT-SN?什么时候适合使用 MQTT,什么时候适合使用 MQTT-SN? 最佳答案 与 M
我正在使用 Paho 和 Mosqitto 进行 MQTT 通信。我们必须支持两种序列化模型——xml 和 json。所以我正在寻找如何识别 MQTT 中的内容类型或有效负载类型。 MQTT 中是否有
我想知道消息发送顺序是否被保留。也就是说,当发布者发送消息序列时,是否保证每个订阅者收到与发布者发送的序列相同的序列?对于干净和持久的 session ? 最佳答案 MQTT 3.1.1 中消息排序功
我正在使用 Paho 和 Mosqitto 进行 MQTT 通信。我们必须支持两种序列化模型——xml 和 json。所以我正在寻找如何识别 MQTT 中的内容类型或有效负载类型。 MQTT 中是否有
我已经在我的本地机器上安装了 MQTT 的 mosquitto 客户端。 我有以下配置: listener 1883 protocol mqtt listener 9001 protocol webs
MQTT 客户端每分钟轮询服务器多少次?是不是大数据流量?我知道数据包的大小可以很小,但是客户端 ping 代理多少次才能使自己在代理中“在线”。 如果我不清楚,请评论这个问题,我会尝试更好地解释我的
我们在不同的场所安装了许多本地网关。这些网关将托管一个 MQTT 代理,用于连接所有本地服务。然后需要本地 MQTT 代理将此数据发送到两个远程 MQTT 代理(每个托管于不同的一方)。我已阅读有关
无论我做什么,我都无法在我的 angular 应用程序中通过 websocket 连接到 mqtt 代理(在 chrome 和 firefox 中尝试)。 为简单起见,我使用 HiveMQ broke
我有一个 MQTT 客户端想要订阅以下主题 node1/数据/更新 节点1/数据/增量 节点1/数据 当然我有多个节点,所以我必须使用通配符,如下所述:https://www.hivemq.com/b
我正在使用 mosquitto (http://mosquitto.org/) 作为 MQTT 代理,并正在寻找有关负载平衡订阅者的建议(针对同一主题)。这是如何实现的?我读到的关于协议(protoc
如题,我可以区分新的和保留的MQTT消息吗?以及如何知道客户端何时发布消息?所有这些没有在主题或消息中指定时间戳字符串。 最佳答案 客户端使用保留标志向代理指示代理应将消息保留为要发布到的主题的“最后
有什么方法可以从另一个客户端找到所有连接的客户端详细信息(IP 和名称)?我知道有一个主题“$SYS/broker/clients/active”给出了当前连接的客户端的数量,但是如果我想了解更多关于
当处于 QOS 1 和 2 时,它会重播所有过去的消息。标准实现中是否有一种方法可以在再次生效时接收整个过去的队列(作为数组)? (当然仅限订阅的) 最佳答案 当客户端以 QOS 1 或 2 订阅主题
我想连接一个客户端,该客户端将监视代理的所有主题,以便在我不知道主题名称是什么时响应事件。 最佳答案 订阅 # 可以让您订阅除以 $ 开头的主题之外的所有内容(无论如何,这些主题通常都是控制主题)。
我是一名优秀的程序员,十分优秀!