1. 首页
  2. 文章中心
  3. python

mqtt mqtt_限制MQTT数据

54 阅读 0 评论
我是靠谱客的博主 孝顺大神，最近开发中收集的这篇文章主要介绍mqtt mqtt_限制MQTT数据，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

mqtt mqtt

介绍

当前市场上可用的大多数MQTT代理都对WebSocket提供了本机支持，从而使任何MQTT JavaScript库都可以通过将MQTT消息封装到WebSocket框架中来建立通信：这称为基于WebSocket的MQTT。

这种方法的最大好处是允许所有现代浏览器（包括在智能手机上运行的浏览器）发送和接收MQTT消息。 这有伸出的协议到网上，从而为M2H它越来越有吸引力（机器对人的）场景太的影响。

但是，互联网在数据包丢失和可用带宽方面的不可预测性（尤其是在移动网络上）使这种“哑管道”方法变得非常不可靠。 例如，由于存在网络，浏览器和用户超载的风险，因此将IoT传感器生成的所有大实时数据发送到基于浏览器的应用程序和移动应用程序既不可行也无用。动态节流是关键。 应该限制​​通过WebSockets流过的数据，以通过重新采样来适应可用带宽。 这样，即使通过很小的带宽连接，在不同网络上订阅了相同高频MQTT主题的不同客户端都将看到最新数据。

MQTT.Cool超越WebSockets

作为位于Web客户端和代理之间的优化网关，MQTT.Cool能够使实时数据流更加有效。 此外，如果您的基础结构不支持WebSocket（例如，由于透明代理或严格的公司防火墙阻止了它们），则MQTT.Cool会自动为每个客户端选择最佳的传输方式，可能会退回到HTTP Streaming和HTTP Long Polling： MQTT.Cool，WebSocket不再是从Web连接到MQTT代理的唯一方法。

通过特殊的技术，如排队 ， 重采样 ，并运用混为一谈数据节流，MQTT.Cool动态优化信息流， 正在使用无论什么交通工具。

数据可以通过两种不同的方式进行调节：

  1. 自适应节流， MQTT.Cool通过自动调整对数据进行实时采样，同时应用合并以适应任何网络拥塞。
  2. 客户端控制的节流，每个客户端可以通过该节流显式配置其下游通道的最大带宽，以及每个扇出订阅的最大更新频率。

您可以在《 入门指南》中找到有关扇出订阅的更多信息； 但出于完整性考虑：

  • 共享连接是从MQTT.Cool到MQTT代理的单个代理连接，在该代理连接的顶部是不同的MQTT.Cool客户端通过隧道传输和多路复用。
  • 同样，扇出订阅是由MQTT.Cool在代理上完成的单个MQTT订阅，用于管理在该主题上发布并由多个远程MQTT.Cool客户端订阅的所有消息。

共享连接和扇出订阅是使任何MQTT代理都能在网络上实现大规模扇出的两种主要机制。

MQTT.Cool具有强大的灵活性和高水平的优化功能，因为它基于Lightstreamer ，这是一种高性能的服务器，世界各地的众多顶级客户都在使用该服务器，以高效，可靠地在Web上传递实时数据。

MQTT节流的展示

为了更好地说明MQTT.Cool如何管理节流，我们创建了MQTT节流演示，该节着重于客户端控制的节流。 该演示是与来自Gambit Communications的朋友合作构建的，该朋友通过功能强大的MIMIC MQTT Simulator帮助了我们，稍后将对此进行介绍。

该演示展示了构建管理实时遥测的Web客户端非常容易，并且非常重要的是，如何根据带宽和频率进一步控制消息的传入流。

MQTT数据

客户端可视化来自十个不同IoT传感器的实时更新，这些传感器不断检测并发布其与移动对象之间的距离并向MQTT经纪人发布。

为了使节流数据与非节流数据之间的差异清晰可见，每个传感器都显示两个图形： 红色点表示节流数据， 橙色线表示非节流数据。

此外，客户端使您可以与UI进行交互，以准确地全局和单独控制节流数据流的影响，如下所示。

控制频率

对于每一帧，您都可以使用滑块来动态更改与传感器相关的传入消息可以到达的最大更新速率。 如下面的动画所示，这会影响在框架上显示红点的频率，在该动画中，频率选择器已从“无限”移动到“ 1更新/秒”，然后又移动到“ 3更新/秒”秒”。

MQTT数据

控制带宽

同样，在页面顶部，选择器使您可以更改所有订阅使用的带宽，从而影响所有帧中呈现的所有红点的整体频率。 在下面的动画中，带宽选择器已从“无限kpbs”更改为
”设置为“ 5 kpbs”，然后设置为“ 15 kbps”：

MQTT数据

与UI交互可帮助您了解带宽和频率限制如何设置MQTT动态管理的上限，在不同级别上很酷：带宽限制全局应用于连接，而频率限制则分别应用于每个MQTT订阅。

请注意，实时更新既不被缓冲也不被延迟，而是被重新采样和合并。 换句话说，当订阅有机会更新（基于循环算法）时，它将收到最新的可用消息，而不是旧消息。

演示架构

下图显示了用于构建演示的整体架构：

MQTT数据

订户

通过使用Web Client API ，JavaScript客户端为每个传感器/主题提交MQTT订阅，以接收实时数据并将其显示在相对图表上。

必须使用Web客户端API，因为MQTT.Cool不会与Web客户端“说”纯MQTT：它通过Lightstreamer协议透明地重新映射MQTT协议，以利用Lightstreamer的能力来移动大量数据。网络。 这意味着您具有从一侧直接连接到MQTT代理的本机MQTT客户端和从另一侧通过Web客户端API冷却的Web客户端连接到MQTT。

发行人

如预期的那样，我们利用了MIMIC MQTT Simulator提供的灵活性和功能，通过它可以轻松生成无限范围的模拟方案，因为它能够生成任意，可定制，可伸缩和可预测的遥测。

甘比特人在其MQTT实验室上部署了模拟的IoT传感器，以发布以正弦波形式生成的距离变化，每个正弦波具有不同的频率以显示不同的流量模式。

经纪人

MQTT代理托管在我们的云基础架构上，位于tcp://broker.mqtt.cool:1883 。 说到这一点，作为可公开访问的经纪人，可以随时将其用于任何测试目的！

门户

作为订户和代理之间的中介，MQTT.Cool服务器扮演着真正的Web网关的角色。 该演示的实时版本连接到https://cloud.mqtt.cool ，这是我们的在线MQTT.Cool实例的地址。 如进一步详细介绍，您可以在演示的本地副本中将其替换为自己的实例。

挖掘代码

该演示基于jQuery，并且如前所述，它是使用MQTT.Cool Web Client API开发的 ，并支持另外两个小型库：

  1. 用于图形渲染的Flotr2。
  2. rangelider.js用于滑块的可视化和操作。

让我们开始研究一些有趣的代码片段，这些片段取自定义了所有应用程序代码的js/app.js文件。

主机地址

在文件顶部，可以找到MQTT.Cool服务器和MQTT代理的地址： 

$(function() {
  // Define urls for MQTT.Cool and the external MQTT broker.
  const MQTT_COOL_URL = 'http://localhost:8080';
  const BROKER_URL = 'tcp://broker.mqtt.cool:1883';
  ...
})

如果要在便携式计算机上克隆项目并定位到其他MQTT .Cool服务器，请根据您的环境替换MQTT_COOL_URL常量。

请注意，这不适用于MQTT代理，因为MIMIC Simulator当前正在将数据发布到我们的云代理。

感测器

定义其他变量后，将通过Sensor构造函数对模拟的远程IoT传感器进行抽象，该函数由传感器ID和框架ID标识 ：

  • 传感器ID用于形成主题，MIMIC Simulator将针对该模拟传感器向其发布遥测数据。 该主题用于提交MQTT订阅。
  • 框架ID用于在HTML页面上查找专用于呈现动画图形的区域。 
function Sensor (sensorId, frameId) {
  this.sensorId = sensorId;
  this.frameId= frameId;
  this.topic = '/gambit/' + this.sensorId + '/telemetry';
  ...
}

以下阵列列出了Gambitt提供的所有传感器： 

const SENSORS = [
    new Sensor('20:19:AB:F4:0D:0D', 'sensor1'),
    new Sensor('20:19:AB:F4:0D:0E', 'sensor2'),
    new Sensor('20:19:AB:F4:0D:0F', 'sensor3'),
    new Sensor('20:19:AB:F4:0D:10', 'sensor4'),
    new Sensor('20:19:AB:F4:0D:11', 'sensor5'),
    new Sensor('20:19:AB:F4:0D:12', 'sensor6'),
    new Sensor('20:19:AB:F4:0D:13', 'sensor7'),
    new Sensor('20:19:AB:F4:0D:14', 'sensor8'),
    new Sensor('20:19:AB:F4:0D:15', 'sensor9'),
    new Sensor('20:19:AB:F4:0D:16', 'sensor10')
  ];

讯息处理

MessageHandler处理从MQTT订阅接收的消息，并将其转发到正确的本地Sensor对象： 

const MessageHandler = function (sensorType) {
    return function (message) {
      ...
    };
  }

一旦实例化，此构造函数将返回将在收到Message实例（ message参数）时调用的回调。

为了识别哪个模拟的远程传感器已发出消息，我们简单地遍历所有SENSORS项，直到唯一可能的主题（如先前定义）与接收到的消息的Message.destinationName不匹配： 

const sourceSensor = SENSORS.find(function (s) {
  return s.topic == message.destinationName
});

由于MIMIC Simulator已配置为将有效载荷发布为JSON字符串，因此我们解析Message.payloadString字段以获取JSON对象： 

const payload = JSON.parse(message.payloadString);

最后，触发Sensor对象以更新sensorType （根据与MessageHandler关联的客户端的“ sensorType ”或“ raw”）查找的图表，并传递有效负载所携带的值： 

sourceSensor.update(sensorType, payload);

这是MessageHanlder整体的外观： 

const MessageHandler = function (sensorType) {
  return function (message) {
    const sourceSensor = SENSORS.find(function (s) {
      return s.topic == message.destinationName;
    });
    const payload = JSON.parse(message.payloadString);
    sourceSensor.update(sensorType, payload);
  };
}

现在，让我们逐步完成连接和订阅。

连接和订阅

首先，我们必须针对网关打开一个新会话： 

mqttcool.openSession(MQTT_COOL_URL, 'demouser', '', {

 onConnectionSuccess: function (mqttCoolSession) {
   ...
 },
 onLsClient: function (lsClient) {
   ...
 }
}

在这里， demouser是我们的在线服务器为所有已部署的现场演示所需的用户名（在这种情况下，必须提供一个空密码）。

最后一个参数是MQTTCoolListener接口的实现，该接口将通知与会话创建有关的事件。

与MQTT.Cool的连接成功后，将触发强制的onConnectionSuccess回调。 通常在这里发生MQTT连接：提供的mqttCoolSession参数确实提供了创建MqttClient实例的入口，该实例配置为在指定地址连接到代理： 

throttledClient = mqttCoolSession.createClient(BROKER_URL);

然后，将客户端与MessageHandler实例相关联，以调度可以限制的消息。 为此，我们设置MqttClient.onMessageArrived回调： 

throttledClient.onMessageArrived = new MessageHandler('throttled');

最后，我们可以连接到代理，并在连接后订阅每个传感器的主题： 

throttledClient.connect({
  onSuccess: function () {
    for (var i = 0; i < SENSORS.length; i++) {
      throttledClient.subscribe(SENSORS[i].topic);
    }
  }
});

以下是onConnectionSuccess的完整版本： 

onConnectionSuccess: function (mqttCoolSession) {
  throttledClient = mqttCoolSession.createClient(BROKER_URL);

  throttledClient.onMessageArrived = new MessageHandler('throttled');

  throttledClient.connect({
    onSuccess: function () {
      for (var i = 0; i < SENSORS.length; i++) {
        throttledClient.subscribe(SENSORS[i].topic);
      }
    }
}

有关使用MQTT.Cool建立连接的深入讨论，请参阅《 入门指南 》的“ MQTT.Cool连接模式”部分。

底层API

因为我们还想操纵全局带宽，所以我们必须利用LighststreamerClient对象，该对象是Lightstreamer Web Client API的入口点，在此对象的基础上构建了MQTT.Cool Web Client API。

该对象管理客户端与MQTT.Cool服务器之间的所有通信，对于每个MQTT.Cool会话，存在一个对应的LightstreamerClient对象。

通过此对象，我们可以在方便时显式更改带宽。 但是我们首先必须缓存该引用以供以后使用。 这是onLsClient来帮助我们的地方： 

onLsClient: function(lsClient) {
  lightstreamerClientReference = lsClient
}

频率管理

现在，让我们研究一下该应用程序如何允许您修改给定IoT传感器的最大更新频率。

我们之前见过的Sensor构造函数还定义了一个内部函数（ initFrequencySelector ），该函数设置关联选择器的初始值。 更重要的是，该函数接受每次移动滑块以选择新值时都要调用的回调： 

function Sensor(sensorId, frameId) {
  ...
  // Initialize the Frequency Selector of this IoT Sensor
  function initFrequencySelector(callback) {
    ...
  }

  // Trigger the Frequency Selector initialization passing the callback
  initFrequencySelector(function(subOptions) {
    throttledClient.subscribe(self.topic, subOptions);
  });
}

回调仅将订阅重新提交给同一主题，但提供不同的SubscribeOptions对象（ subOpts参数
），其中包含选择器提供的更新后的maxFrequency值。 

onSlideEnd: function (position, value) {
  ...
  const subOptions = {};
  if (value !== 'Unlimited') {
    subOptions['maxFrequency'] = value;
  }
  onSlideEndCallback(subOptions);
}

重新订阅后，消息将以指定的速率开始流。

带宽管理

更新带宽的逻辑是相似的，但是这次您必须像上面预期的那样使用缓存的LightstramerClient对象。

initBandwidhtSelector函数（在外部作用域中定义）用于初始化全局带宽选择器的状态，并需要在处理滑块时触发的回调： 

initBandwidthSelector(function (value) {
  lightstreamerClientRef.connectionOptions.setMaxBandwidth(value);
});

通过调用LightstreamerClient.connectOptions.setMaxBandiwidth方法，可LightstreamerClient.connectOptions.setMaxBandiwidth所有订阅可用的最大允许带宽。
; 不久之后，您将看到这如何影响所有图形，其中基于通过选择器选择的频段，以可变间隔开始渲染正弦波。

入门指南的附录“访问Lightstreamer客户端API”深入研究了两个客户端API之间的关系，并向您展示了如何从中获得其他好处。

非限制数据

频率和带宽的操作仅对呈现为红点的传入消息有影响，因此与从受限制的客户端请求的MQTT订阅有关。

但是，直观地将限制数据与原始数据（即非限制数据）进行比较有助于我们更好地了解带宽和频率如何真正影响消息流。

这就是为什么我们在演示中使用了另一种类型的MQTT.Cool连接： 专用连接 （ Dedicated Connection） ，其MQTT订阅将接收来自MQTT代理的数据，而无需任何进一步的操作，这可能是由客户端或MQTT.Cool服务器触发的。 流经此连接的消息在框架中显示为连续的橙色线。

以下片段显示了为原始客户端进行的连接和后续订阅： 

mqttcool.openSession(MQTT_COOL_URL, 'demouser', '', {

  onConnectionSuccess: function (mqttCoolSession) {
    const clientId = 'client-' + new Date().getTime().toString();
    rawClient = mqttCoolSession.createClient(BROKER_URL, clientId);

    rawClient.onMessageArrived = new MessageHandler('raw');

    rawClient.connect({
      onSuccess: function () {
        for (var i = 0; i < SENSORS.length; i++) {
          rawClient.subscribe(SENSORS[i].topic);
        }
      }
    });
});

使用的模式是通常的模式。 实际上，代码段与节流情况类似，但是有两个突出的重要区别：

  • 第4-5行：客户机标识符（尽管并不假装在所有可能的客户机中唯一），会激发会话返回MQTT客户机，该客户机将触发服务器端的专用代理连接。
  • 第7行：MessageHanlder 现在使用 raw 作为传感器类型进行 实例化 ，从而解决了对应的图表类型，即橙色线。

结论

该演示展示了可以用来改善MQTT流管理的基本技术，可以很好地控制消息流的频率和带宽。 底层的Lightstreamer技术使之成为可能，该技术允许为每个订阅分配所需的频率，并为整个连接分配总带宽，还保证不超过要求的值。

在不久的将来，通过以下改进，新的API扩展（不会破坏类似于Paho的设计）将使客户端控制的节流更加容易：

  • 临时添加到SubscribeOptions以控制频率，从而不再需要重新订阅。
  • 通过MQTTCoolSession公开的显式操作直接控制带宽。

通过MQTT享受MQTT消息传递。酷，敬请期待！

翻译自: https://www.javacodegeeks.com/2019/07/throttling-mqtt-data.html

mqtt mqtt

最后

以上就是孝顺大神为你收集整理的mqtt mqtt_限制MQTT数据的全部内容，希望文章能够帮你解决mqtt mqtt_限制MQTT数据所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。

本图文内容来源于网友提供，作为学习参考使用，或来自网络收集整理，版权属于原作者所有。
点赞(36)

相关文章

MQTT协议及安全浅析
MQTT协议及安全浅析
消息通讯——MQTT安全认证和测试EMQX安全认证
消息通讯——MQTT安全认证和测试EMQX安全认证
MQTT学习之树莓派安装mosquito 服务器MQTT学习之树莓派安装mosquito 服务器
MQTT学习之树莓派安装mosquito 服务器MQTT学习之树莓派安装mosquito 服务器
MQTT使用TLS加密
MQTT使用TLS加密
mqtt mqtt_限制MQTT数据
mqtt mqtt_限制MQTT数据
MQTT1. 什么是 MQTT？2. MQTT 协议消息格式3. MQTT 协议中的消息4. MQTT 协议核心特性MQTT通配符使用
MQTT1. 什么是 MQTT？2. MQTT 协议消息格式3. MQTT 协议中的消息4. MQTT 协议核心特性MQTT通配符使用
Mqtt之安全
Mqtt之安全
Mqtt精髓系列之安全
Mqtt精髓系列之安全

评论列表共有 0 条评论

立即
投稿
返回
顶部