kafka接收mqtt数据

去年11月，我们买了一辆电动汽车，这引起了一个有趣的问题：我们什么时候应该充电？ 我担心汽车充电所用的排放量最低，所以这是一个具体的问题：在任何给定时间，每千瓦时的二氧化碳排放率是多少？白天是什么时候最低？

查找数据

我住在纽约州。 我们约80％的电力来自州内发电，主要来自天然气，水坝（其中大部分来自尼亚加拉大瀑布），核能以及一些风能，太阳能和其他化石燃料。 整个系统由纽约独立系统运营商 （NYISO）管理， 纽约独立系统运营商是一个非营利性机构，其成立目的是平衡发电机，消费者和监管机构的需求，以使纽约保持正常运转。

在构建用于收集和处理开放数据的工具时，我们应该始终保持友善，以免使这些系统过载。 与其将所有人一直发送到他们的存档服务来下载文件，我们可以做得更好。 我们可以创建一个开销很小的事件流，人们可以订阅并随时获取更新。 我们可以使用MQTT做到这一点。 我的项目（ ny-power.org ）的目标是包含在Home Assistant项目中，该项目是一个拥有数十万用户的开源家庭自动化平台。 如果所有这些用户一直都在访问此CSV服务器，则NYISO可能需要限制对其的访问。

什么是MQTT？

MQTT是一种发布/订阅（pubsub）有线协议，其设计考虑了小型设备。 Pubsub系统的工作就像消息总线一样。 您将消息发送到主题，任何订阅该主题的软件都将获得消息的副本。 作为发送者，您永远不会真正知道谁在听。 您只需将信息提供给一组主题，然后聆听您可能关心的其他任何主题。 就像参加派对并听有趣的对话一样。

这可以使应用程序效率极高。 客户订阅的主题很窄，只接收他们正在寻找的信息。 这样既节省了处理时间，又节省了网络带宽。

作为一种开放标准，MQTT具有许多客户端和服务器的开源实现。 您可以想象到每种语言的客户端库，甚至可以嵌入到Arduino中的用于建立传感器网络的库。 有许多服务器可供选择。 我要去的是Eclipse的Mosquitto服务器，它很小，是用C编写的，可以处理成千上万的订户而不会费力。

为什么我喜欢MQTT

在过去的二十年中，我们为软件应用程序提出了经过验证的真实模型，以提出服务问题。 我还有电子邮件吗？ 当前天气如何？ 我现在应该买这东西吗？ 这种“询问/接收”模式在很多时候都有效。 但是，在充满数据的世界中，我们还需要其他模式。 MQTT pubsub模型功能强大，可以将大量数据发布到系统中。 客户端可以订阅少量数据，并在收到数据后立即接收更新。

MQTT还具有其他有趣的功能，例如“最后遗嘱”消息，这使得可以区分由于没有相关数据而导致的静默和由于数据收集器崩溃而导致的静默。 MQTT还保留了消息，这些消息在客户端首次连接时将有关主题的最后一条消息提供给客户端。 这对于主题更新缓慢非常有用。

在与Home Assistant项目一起工作时，我发现此消息总线模型对于异构系统非常有效。 如果您进入物联网领域，那么您将很快在任何地方都遇到MQTT。

我们的第一个MQTT流

NYSO的CSV文件之一是实时燃料混合。 每五分钟更新一次该时间段内的燃料来源和发电量（兆瓦）。

CSV文件如下所示：

该表中唯一奇怪的是双燃料类别。 纽约的大多数天然气厂还可以燃烧其他化石燃料来发电。 在冬季寒冷期间，天然气供应受到限制，其用于家庭供暖的优先级高于发电。 这种情况发生的频率足够低，我们可以将双重燃料视为天然气（用于我们的计算）。

该文件全天更新。 我创建了一个简单的数据泵，该泵每分钟轮询一次文件并查找更新。 它将所有新条目发布到MQTT服务器上，并散布到很大程度上镜像此CSV文件的一组主题中。 有效负载被转换成一个JSON对象，几乎可以从任何一种编程语言中对其进行解析。

   

    

     ny-power/upstream/fuel-mix/Hydro {"units": "MW", "value": 3229, "ts": "05/09/2018 00:05:00"}
     

ny-power/upstream/fuel-mix/Dual Fuel {"units": "MW", "value": 1400, "ts": "05/09/2018 00:05:00"}
     

ny-power/upstream/fuel-mix/Natural Gas {"units": "MW", "value": 2144, "ts": "05/09/2018 00:05:00"}
     

ny-power/upstream/fuel-mix/Other Fossil Fuels {"units": "MW", "value": 4, "ts": "05/09/2018 00:05:00"}
     

ny-power/upstream/fuel-mix/Wind {"units": "MW", "value": 41, "ts": "05/09/2018 00:05:00"}
     

ny-power/upstream/fuel-mix/Other Renewables {"units": "MW", "value": 226, "ts": "05/09/2018 00:05:00"}
     

ny-power/upstream/fuel-mix/Nuclear {"units": "MW", "value": 4114, "ts": "05/09/2018 00:05:00"}
    
   

这种直接反映是将开放数据转换为开放事件的良好第一步。 我们会将其转换为CO2强度，但其他应用程序可能希望这些原始饲料对其进行其他计算。

MQTT主题

主题和主题结构是MQTT的主要设计要点之一。 与更多的“企业”消息总线不同，MQTT中的主题未预先注册。 发件人可以即时创建主题，唯一的限制是主题少于220个字符。 /字符是特殊的； 它用于创建主题层次结构。 正如我们将很快看到的，您可以在这些层次结构中订阅数据片。

Mosquitto开箱即用，每个客户都可以发布到任何主题。 虽然非常适合制作原型，但在投入生产之前，您需要添加访问控制列表（ACL）以限制对授权应用程序的写入。 例如，每个人都可以以只读格式访问我的应用程序树，但是只有具有特定凭据的客户端才能向其发布。

没有围绕主题的自动架构，也没有发现客户将发布到的所有可能主题的方法。 您必须将这种理解直接编码到使用MQTT总线的任何应用程序中。

那么，您应该如何设计主题？ 最佳实践是从特定于应用程序的根名称开始，在本例中为ny-power 。 之后，构建有效订阅所需的深度层次结构。 upstream树将包含直接来自上游源的数据，而无需任何处理。 我们的fuel-mix类别是一种特定的数据类型。 我们可能会在以后添加其他人。

订阅主题

MQTT中的订阅是简单的字符串匹配。 为了提高处理效率，只允许使用两个通配符：

• #递归匹配所有内容
• +仅匹配到下一个/字符

用一些示例最容易解释这一点：

   

    

     ny-power/#  - match everything published by the ny-power app
     

ny-power/upstream/#  - match all raw data
     

ny-power/upstream/fuel-mix/+  - match all fuel types
     

ny-power/+/+/Hydro - match everything about Hydro power that's
     

       nested 2 deep (even if it's not in the upstream tree)
    
   

ny-power/#这样的广泛订阅在小批量应用中很常见。 只需通过网络获取所有内容，然后在您自己的应用程序中进行处理即可。 这对于大容量应用程序效果不佳，因为当您将大多数消息放在地板上时，大部分网络带宽将被浪费。

为了保持较高的性能，应用程序将执行一些巧妙的主题幻灯片，例如ny-power/+/+/Hydro以准确获取所需数据的横截面。

添加我们的下一层数据

从现在开始，应用程序中的所有内容都将在现有的MQTT流上运行。 数据的第一附加层是计算电源的CO2强度。

使用2016年美国能源信息管理局（Energy Information Administration）在纽约的总排放量和按燃料类型划分的总功率，我们可以得出每兆瓦时功率的平均排放率 。

这被封装在专用微服务中。 这具有ny-power/upstream/fuel-mix/+上的订阅，该订阅与来自数据泵的所有上游燃料混合项匹配。 然后执行计算并发布到新的主题树：

 ny-power/computed/co2 {"units": "g / kWh", "value": 152.9486, "ts": "05/09/2018 00:05:00"} 

反过来，还有另一个过程可以订阅该主题树，并将该数据存档到InfluxDB实例中。 然后，它将24小时制的时间序列发布到ny-power/archive/co2/24h ，这使得绘制最近的变化变得很容易。

该层模型运行良好，因为每个程序的逻辑都可以彼此不同。 在更复杂的系统中，它们甚至可能不在同一编程语言中。 我们不在乎，因为交换格式是MQTT消息，带有著名的主题和JSON负载。

从命令行使用

为了对MQTT有所了解，将其附加到总线并查看消息流很有用。 该mosquitto_sub列入计划mosquitto-clients软件包是一个简单的方法来做到这一点。

安装后，您需要提供服务器主机名和您想听的主题。 如果要查看发布到的主题， -v标志很重要。 否则，您将仅看到有效载荷。

 mosquitto_sub -h mqtt.ny-power.org -t ny-power/# -v 

每当我编写或调试MQTT应用程序时，总是有一个运行mosquitto_sub的终端。

直接从Web访问MQTT

现在，我们有了一个提供开放事件流的应用程序。 我们可以使用微服务连接到它，并使用一些命令行工具将其连接到Internet上以供所有人查看。 但是网络仍然是王者，因此直接将其带入用户的浏览器很重要。

MQTT的人们对此进行了思考。 协议规范旨在处理三种传输协议： TCP ， UDP和WebSockets 。 所有主流浏览器都支持WebSocket，这是为实时应用程序保留持久连接的一种方式。

Eclipse项目具有名为Paho的MQTTJavaScript实现，可以将其包含在您的应用程序中。 模式是连接到主机，设置一些订阅，然后在收到消息时对消息做出React。

   

    
 
     // ny-power web console application 
     


     var client 
     = 
     new Paho. 
     MQTT . 
     Client 
     ( mqttHost 
     , 
     Number 
     ( 
     "80" 
     ) 
     , 
     "client-" 
     + 
     Math . 
     random 
     ( 
     ) 
     ) 
     ; 
     


     


     // set callback handlers 
     

client. 
     onMessageArrived 
     = onMessageArrived 
     ; 
     


     


     // connect the client 
     

client. 
     reconnect 
     = 
     true 
     ; 
     

client. 
     connect 
     ( 
     { onSuccess 
     : onConnect 
     } 
     ) 
     ; 
     


     


     // called when the client connects 
     


     function onConnect 
     ( 
     ) 
     { 
     

    
     // Once a connection has been made, make a subscription and send a message. 
     

    console. 
     log 
     ( 
     "onConnect" 
     ) 
     ; 
     

    client. 
     subscribe 
     ( 
     "ny-power/computed/co2" 
     ) 
     ; 
     

    client. 
     subscribe 
     ( 
     "ny-power/archive/co2/24h" 
     ) 
     ; 
     

    client. 
     subscribe 
     ( 
     "ny-power/upstream/fuel-mix/#" 
     ) 
     ; 
     


     } 
     


     


     // called when a message arrives 
     


     function onMessageArrived 
     ( message 
     ) 
     { 
     

    console. 
     log 
     ( 
     "onMessageArrived:" 
     + message. 
     destinationName 
     + message. 
     payloadString 
     ) 
     ; 
     

    
     if 
     ( message. 
     destinationName 
     == 
     "ny-power/computed/co2" 
     ) 
     { 
     

        
     var data 
     = JSON. 
     parse 
     ( message. 
     payloadString 
     ) 
     ; 
     

        $ 
     ( 
     "#co2-per-kwh" 
     ) . 
     html 
     ( 
     Math . 
     round 
     ( data. 
     value 
     ) 
     ) 
     ; 
     

        $ 
     ( 
     "#co2-units" 
     ) . 
     html 
     ( data. 
     units 
     ) 
     ; 
     

        $ 
     ( 
     "#co2-updated" 
     ) . 
     html 
     ( data. 
     ts 
     ) 
     ; 
     

    
     } 
     

    
     if 
     ( message. 
     destinationName . 
     startsWith 
     ( 
     "ny-power/upstream/fuel-mix" 
     ) 
     ) 
     { 
     

        fuel_mix_graph 
     ( message 
     ) 
     ; 
     

    
     } 
     

    
     if 
     ( message. 
     destinationName 
     == 
     "ny-power/archive/co2/24h" 
     ) 
     { 
     

        
     var data 
     = JSON. 
     parse 
     ( message. 
     payloadString 
     ) 
     ; 
     

        
     var plot 
     = 
     [ 
     

            
     { 
     

                x 
     : data. 
     ts 
     , 
     

                y 
     : data. 
     values 
     , 
     

                type 
     : 
     'scatter' 
     

            
     } 
     

        
     ] 
     ; 
     

        
     var layout 
     = 
     { 
     

            yaxis 
     : 
     { 
     

                title 
     : 
     "g CO2 / kWh" 
     , 
     

            
     } 
     

        
     } 
     ; 
     

        Plotly. 
     newPlot 
     ( 
     'co2_graph' 
     , plot 
     , layout 
     ) 
     ; 
     

    
     } 
    
   

该应用程序订阅了许多主题，因为我们将显示几种不同类型的数据。 ny-power/computed/co2主题为我们提供了当前强度的最重要数字。 每当我们收到该主题时，我们都会替换网站上的相关内容。

^{ny-power.org的 NY ISO Grid CO ₂强度图。}

ny-power/archive/co2/24h主题提供了可以加载到Plotly折线图中的时间序列。 ny-power/upstream/fuel-mix可提供所需数据，以提供当前燃料混合的漂亮条形图。

^{NYISO电网ny-power.org上的燃料混合。}

这是一个不轮询服务器的动态网站。 它连接到MQTT总线，并在其打开的WebSocket上侦听。 该网页是一个发布/订阅客户端，就像数据泵和存档器一样。 这只是在您的浏览器中执行，而不是在云中执行微服务。

您可以在http://ny-power.org上看到正在运行的页面。 其中包括图形和实时MQTT控制台，以查看传入的消息。

潜水更深

整个ny-power.org应用程序都可以在GitHub上以开源形式获得 。 您还可以查看此架构概述，以了解如何将其构建为与Helm一起部署的一组Kubernetes微服务。 您可以看到另一个有趣的MQTT应用程序示例，其中包含使用MQTT和OpenWhisk实时翻译文本消息的代码模式 。

MQTT在物联网领域得到了广泛的使用，在Home Assistant项目中可以找到更多的MQTT使用示例。

如果您想深入研究该协议，则mqtt.org包含此开放标准的所有详细信息。

要了解更多信息，请参加7月16日至19日在俄勒冈州波特兰举行的OSCON上的Sean Dague的演讲， 将MQTT添加到您的工具箱中 。

翻译自: https://opensource.com/article/18/6/mqtt

kafka接收mqtt数据