☰

如何从 MQTT 流中提取某些数据

tag-icon • tag-icon text-processing streaming data
如何从 MQTT 流中提取某些数据

我mosquitto_sub -d -t +/#从 Ubuntu 终端输入来访问 MQTT 流。

不过，我只想查看 MQTT 流中的一组特定数据。我已经添加了| grep -v PING所有内容，除了带有 PING 的行之外，所有内容仍然会被打印，但这不起作用。

我已经尝试过| grep -A1 PUBLISH，所以我可以看到 MQTT 中带有 PUBLISH 的行，没有任何变化：我只是获取整个 MQTT 流，没有任何内容被过滤掉。

如何从 MQTT 流中看到我想看到的特定内容？或者;如何从 MQTT 流中过滤某些内容或者如何从 MQTT 流中提取数据？

我从 ubuntu 终端的输入如下$ mosquitto_sub -d -t +/#，我已经尝试过$ mosquitto_sub -d -t +/# | grep -v PING，$ mosquitto_sub -d -t +/# | grep -A1 PUBLISH但它只是流式传输所有内容，不会 grep 任何内容。输出如下：

ed@agharta:~$ mosquitto_sub -d -t +/#
Received CONNACK
Received SUBACK
Subscribed (mid: 1): 0
Sending PINGREQ
Received PINGRESP
Sending PINGREQ
Received PINGRESP
Sending PINGREQ
Received PINGRESP
Sending PINGREQ
Received PINGRESP
Sending PINGREQ
Received PINGRESP
Sending PINGREQ
Received PINGRESP
Sending PINGREQ
Received PINGRESP
Sending PINGREQ
Received PINGRESP
Sending PINGREQ
Received PINGRESP
Sending PINGREQ
Received PINGRESP
Sending PINGREQ
Received PINGRESP
Sending PINGREQ
Received PINGRESP
Sending PINGREQ
Received PINGRESP
Received PUBLISH (d0, q0, r0, m0, 'm/jsyd/TEST/001/d/status', ... (34 bytes))
Sending PINGREQ
Received PINGRESP
Received PUBLISH (d0, q0, r0, m0, 'm/jsyd/TEST/001/d/SVlts', ... (28 bytes))
Sending PINGREQ
Received PINGRESP

例如我怎样才能仅有的从直播中看到 PUBLISH，而不是 PING？

答案1

我正在写另一个相关问题的以下答案这里当它被删除时。我不知道您是否能够点击该链接，但是（缩写）下面提到的示例数据是这样的：

Sending PINGREQ
Received PINGRESP
Sending PINGREQ
Received PINGRESP
Received PUBLISH (d0, q0, r0, m0, 'm/gf/TMX6BP/075/d/SVlts', ... (28 bytes))
86,1224830,27610 27869 17565
Received PUBLISH (d0, q0, r0, m0, 'm/gf/TMX6BP/075/d/status', ... (39 bytes))
86,1243000,164573,-33.836465,151.051189
Sending PINGREQ
Received PINGRESP

sed所以我写了以下内容以使其像这样工作：

mosquitto ... 2>&1 | 
sed -u "/.*PUBLISH[^']*\([^,]*,\).*/!d 
       s//\1/;N;s/[\%]/&&/g;s/'/&\\\\&&/g
       s/.*/printf '&'/;s/\n/'&date '+[%s,]/"|  
sh -s

现在，我没有你的mosquitto程序，但是如果我将你的整个示例内容复制到我的剪贴板，然后执行以下操作：

xsel -bo | sed ... | sh ...

它打印：

'm/gf/TMX6BP/075/d/SVlts',[1425002404,]86,1224830,27610 27869 17565
'm/gf/TMX6BP/075/d/status',[1425002404,]86,1243000,164573,-33.836465,151.051189
'm/NWRL/TMX/098/d/SVlts',[1425002404,]806,3040421,7549 7750 3904
'm/NWRL/TMX/098/d/status',[1425002404,]806,3069000,59666,-33.836465,151.051189
'm/NWRL/TMX/098/d/SVlts',[1425002404,]810,5440995,6143 7807 4076
'm/NWRL/TMX/098/d/status',[1425002404,]810,5489000,59897,-33.836465,151.051189

...我想这就是你想要的。它的工作原理是确保 shell 和datefor的格式字符串安全地引用所有输入printf，然后从每个行对构造 shell 命令。

对于每个行对，printf首先打印第一行的内容，不带尾随的\newline，然后date打印[timestamp]自纪元以来的秒数，然后打印第二行的内容和尾随的\newline。sed的输出看起来像这样：

printf ''\''m/gf/TMX6BP/075/d/SVlts'\'','
date '+[%s,]86,1224830,27610 27869 17565'

sh当然，在 处打印的任何命令都不会被执行，直到sed在 处打印它们为止。因此，如果sed在读取mosquitto的输出中，有一行晚于前一行，那么该date行将反映这一点。

例如：

(set -f; IFS='
';  for l in $(xsel -bo)
    do  printf %s\\n "$l"
        sleep 1
    done) |
sed -u "/.*PUBLISH[^']*\([^,]*,\).*/!d 
       s//\1/;N;s/[\%]/&&/g;s/'/&\\\\&&/g
       s/.*/printf '&'/;s/\n/'&date '+[%s,]/"|  
sh -s

...在打印每个输出行之前等待一秒钟，sed以便更忠实地模拟实时流。它利用 GNU sed -unbuffered 开关来sed尽可能频繁地刷新 的输出，但如果您无法做到这一点，请告诉我，我将向您展示如何正确阻塞管道以dd避免缓冲问题。

不管怎样，上面的打印结果是：

'm/gf/TMX6BP/075/d/SVlts',[1425002863,]86,1224830,27610 27869 17565
'm/gf/TMX6BP/075/d/status',[1425002865,]86,1243000,164573,-33.836465,151.051189
'm/NWRL/TMX/098/d/SVlts',[1425002869,]806,3040421,7549 7750 3904
'm/NWRL/TMX/098/d/status',[1425002871,]806,3069000,59666,-33.836465,151.051189
'm/NWRL/TMX/098/d/SVlts',[1425002881,]810,5440995,6143 7807 4076
'm/NWRL/TMX/098/d/status',[1425002885,]810,5489000,59897,-33.836465,151.051189

相关内容