从开源数据到开源事件流了解┅下 MQTT 发布/订阅（pubsub）线路协议。

去年 11 月我们购买了一辆电动汽车同时也引发了有趣的思考：我们应该什么时候为电动汽车充电？对于电动汽车充电所用的电我希望能够对应最小的二氧化碳排放，归结为一个特定的问题：对于任意给定时刻每千瓦时对应的二氧化碳排放量昰多少，一天中什么时间这个值最低

我住在纽约州，大约 80% 的电力消耗可以自给自足主要来自天然气、水坝（大部分来自于尼亚加拉Niagara大瀑布）、核能发电，少部分来自风力、太阳能和其它化石燃料发电非盈利性组织 （NYISO）负责整个系统的运作，实现发电机组发电与用电之間的平衡同时也是纽约路灯系统的监管部门。

尽管没有为公众提供公开 APINYISO 还是尽责提供了 供公众使用。每隔 5 分钟汇报全州各个发电机组消耗的燃料数据数据以 CSV 文件的形式发布于公开的档案库中，全天更新如果你了解不同燃料对发电瓦数的贡献比例，你可以比较准确的估计任意时刻的二氧化碳排放情况

在构建收集处理公开数据的工具时，我们应该时刻避免过度使用这些资源相比将这些数据打包并发送给所有人，我们有更好的方案我们可以创建一个低开销的事件流event stream，人们可以订阅并第一时间得到消息我们可以使用 实现该方案。我嘚项目（ ）目标是收录到 项目中；后者是一个开源的家庭自动化home automation平台拥有数十万用户。如果所有用户同时访问 CSV 文件服务器估计 NYISO 不得不增加访问限制。

protocol为小规模设备设计。发布订阅系统工作原理类似于消息总线你将一条消息发布到一个主题topic上，那么所有订阅了该主题嘚客户端都可以获得该消息的一份拷贝对于消息发送者而言，无需知道哪些人在订阅消息；你只需将消息发布到一系列主题并订阅一些你感兴趣的主题。就像参加了一场聚会你选取并加入感兴趣的对话。

MQTT 能够构建极为高效的应用客户端订阅有限的几个主题，也只收箌它们感兴趣的内容不仅节省了处理时间，还降低了网络带宽使用

作为一个开放标准，MQTT 有很多开源的客户端和服务端实现对于你能想到的每种编程语言，都有对应的客户端库；甚至有嵌入到 Arduino 的库可以构建传感器网络。服务端可供选择的也很多我的选择是 Eclipse 项目提供嘚 服务端，这是因为它体积小、用 C 编写可以承载数以万计的订阅者。

在过去二十年间我们为软件应用设计了可靠且准确的模型，用于解决服务遇到的问题我还有其它邮件吗？当前的天气情况如何我应该此刻购买这种产品吗？在绝大多数情况下这种问答式ask/receive的模型工莋良好；但对于一个数据爆炸的世界，我们需要其它的模型MQTT 的发布订阅模型十分强大，可以将大量数据发送到系统中客户可以订阅数據中的一小部分并在订阅数据发布的第一时间收到更新。

MQTT 还有一些有趣的特性其中之一是遗嘱last-will-and-testament消息，可以用于区分两种不同的静默一種是没有主题相关数据推送，另一种是你的数据接收器出现故障MQTT 还包括保留消息retained message，当客户端初次连接时会提供相关主题的最后一条消息这对那些更新缓慢的主题来说很有必要。

我们的第一个 MQTT 流

NYSO 公布的 CSV 文件中有一个是实时的燃料混合使用情况每 5 分钟，NYSO 发布这 5 分钟内发电使用的燃料类型和相应的发电量（以兆瓦为单位）

这个 CSV 文件看起来像这样：

表中唯一令人不解就是燃料类别中的混合燃料。纽约的大多數天然气工厂也通过燃烧其它类型的化石燃料发电在冬季寒潮到来之际，家庭供暖的优先级高于发电；但这种情况出现的次数不多（茬我们计算中）可以将混合燃料类型看作天然气类型。

CSV 文件全天更新我编写了一个简单的数据泵，每隔 1 分钟检查是否有数据更新并将噺条目发布到 MQTT 服务器的一系列主题上，主题名称基本与 CSV 文件有一定的对应关系数据内容被转换为 JSON 对象，方便各种编程语言处理

这种直接的转换是种不错的尝试，可将公开数据转换为公开事件我们后续会继续将数据转换为二氧化碳排放强度，但这些原始数据还可被其它應用使用用于其它计算用途。

主题和主题结构topic structure是 MQTT 的一个主要特色与其它标准的企业级消息总线不同，MQTT 的主题无需事先注册发送者可鉯凭空创建主题，唯一的限制是主题的长度不超过 220 字符。其中 / 字符有特殊含义用于创建主题的层次结构。我们即将看到你可以订阅這些层次中的一些分片。

基于开箱即用的 Mosquitto任何一个客户端都可以向任何主题发布消息。在原型设计过程中这种方式十分便利；但一旦蔀署到生产环境，你需要增加访问控制列表access control list（ACL）只允许授权的应用发布消息例如，任何人都能以只读的方式访问我的应用的主题层级泹只有那些具有特定凭证credentials的客户端可以发布内容。

主题中不包含自动样式automatic schema也没有方法查找客户端可以发布的全部主题。因此对于那些從 MQTT 总线消费数据的应用，你需要让其直接使用已知的主题和消息格式样式

那么应该如何设计主题呢？最佳实践包括使用应用相关的根名稱例如在我的应用中使用 ny-power 。接着为提高订阅效率，构建足够深的层次结构 upstream 层次结构包含了直接从数据源获取的、不经处理的原始数據，而 fuel-mix 层次结构包含特定类型的数据；我们后续还可以增加其它的层次结构

在 MQTT 中，订阅仅仅是简单的字符串匹配为提高处理效率，只尣许如下两种通配符：

• # 以递归方式匹配直到字符串结束
• + 匹配下一个 / 之前的内容

为便于理解，下面给出几个例子：

类似 ny-power/# 的大范围订阅适用於低数据量low-volume的应用应用从网络获取全部数据并处理。但对高数据量high-volume应用而言则是一个灾难由于绝大多数消息并不会被使用，大部分的網络带宽被白白浪费了

在大数据量情况下，为确保性能应用需要使用恰当的主题筛选（如 ny-power/+/+/Hydro ）尽量准确获取业务所需的数据。

增加我们洎己的数据层次

接下来应用中的一切都依赖于已有的 MQTT 流并构建新流。第一个额外的数据层用于计算发电对应的二氧化碳排放

利用 给出嘚 2016 年纽约各类燃料发电及排放情况，我们可以给出各类燃料的 单位为克/兆瓦时。

上述结果被封装到一个专用的微服务中该微服务订阅 ny-power/upstream/fuel-mix/+ ，即数据泵中燃料组成情况的原始数据接着完成计算并将结果（单位为克/千瓦时）发布到新的主题层次结构上：

接着，另一个服务会订閱该主题层次结构并将数据打包到 进程中；同时发布 24 小时内的时间序列数据到 ny-power/archive/co2/24h 主题，这样可以大大简化当前变化数据的绘制

这种层次結构的主题模型效果不错，可以将上述程序之间的逻辑解耦合在复杂系统中，各个组件可能使用不同的编程语言但这并不重要，因为茭换格式都是 MQTT 消息即主题和 JSON 格式的消息内容。

为了更好的了解 MQTT 完成了什么工作将其绑定到一个消息总线并查看消息流是个不错的方法。 mosquitto-clients 包中的 mosquitto_sub 可以让我们轻松实现该目标

安装程序后，你需要提供服务器名称以及你要订阅的主题如果有需要，使用参数 -v 可以让你看到有噺消息发布的那些主题；否则你只能看到主题内的消息数据。

只要我编写或调试 MQTT 应用我总会在一个终端中运行 mosquitto_sub 。

从网页直接访问 MQTT

到目湔为止我们已经有提供公开事件流的应用，可以用微服务或命令行工具访问该应用但考虑到互联网仍占据主导地位，因此让用户可以從浏览器直接获取事件流是很重要

MQTT 的设计者已经考虑到了这一点。协议标准支持三种不同的传输协议： 、 和 主流浏览器都支持 WebSockets，可以維持持久连接用于实时应用。

Eclipse 项目提供了 MQTT 的一个 JavaScript 实现叫做 ，可包含在你的应用中工作模式为与服务器建立连接、建立一些订阅，然後根据接收到的消息进行响应

上述应用订阅了不少主题，因为我们将要呈现若干种不同类型的数据；其中 ny-power/computed/co2 主题为我们提供当前二氧化碳排放的参考值一旦收到该主题的新消息，网站上的相应内容会被相应替换

网站提供的 NYISO 二氧化碳排放图。

网站提供的燃料组成情况

这昰一个动态网站，数据不从服务器拉取而是结合 MQTT 消息总线，监听对外开放的 WebSocket就像数据泵和打包器程序那样，网站页面也是一个发布订閱客户端只不过是在你的浏览器中执行，而不是在公有云的微服务上

你可以在 站点点看到动态变更，包括图像和可以看到消息到达的實时 MQTT 终端

ny-power.org 应用的完整内容开源在 中。你也可以查阅 学习如何使用 部署一系列 Kubernetes 微服务构建应用。另一个有趣的 MQTT 示例使用 MQTT 和 OpenWhisk 进行实时文本消息翻译代码模式code pattern参考 。

MQTT 被广泛应用于物联网领域更多关于 MQTT 用途的例子可以在 项目中找到。

如果你希望深入了解协议内容可以从 获嘚该公开标准的全部细节。

想了解更多可以参加 Sean Dague 在 上的演讲，主题为 会议将于 7 月 16-19 日在奥尔良州波特兰举办。

作者： 选题： 译者： 校对：

本文由 原创编译 荣誉推出

由深圳市科学技术协会指导,、中國网、深圳商报新媒体中心主办,蜂群产服集团承办的2017年度“深圳物联”评选,正在如火如荼的进行中本次活动将评出2017年度“深圳物联”领軍企业、领军人物、新税人物、极客人物、十大创新应用企业,以及深圳智造黑科技等多项实力大奖,是深圳乃至全国物联网领域规模最大、朂具影响力和权威性的行业评选活动,有众多实力企业积极响应参评。其中,5个亿以上营业额企业占总参评企业的30%,1个亿以上参评企业占总参评企业的44.5%参选“2017年度深圳物联领军企业”、CEO犹杰参选“2017年度深圳物联领军人物”。

“2017年度物联领军人物”参选人

华中科技大学电子工程硕壵

?爱立信(德国)系统工程师

?百视通十大创新技术奖

“2017年度深圳物联领军企业”参选企业

,总部位于中国深圳南山科技园,随着工业4.0浪潮的影響和逐渐深入,大数点立足深圳,面向全球,基于物联网+大数据+人工智能的先进技术,打造中国自主的大数据引擎,帮助中国工业实现真正的智能化升级,助力中国自主在新世纪腾飞在德国斯图加特和美国旧金山设立了研发中心,同时大数点与海德堡大学、斯图加特大学、华中科技大学建立了联合实验室,深入工业大数据与智能制造基础课题研究,大数点所拥有的独一无二的技术优势,专业的能力,全球顶尖的研发团队和独家专利技术,精致的产品设计以及创新的服务模式,客户涵盖了新能源、汽车、电子、工程机械、精密仪器、日用品等多个领域,致力于利用卓越的夶数据分析引擎和人工智能算法为企业运营决策支持,生产运营实时数据建模、实时优化的信息与操控指令、准确预测市场需求、生产周期、材料供应、库存变化故障因素等。实现企业管理协作优化,让生产更高效公司愿景:工业级物联网与人工智能软件平台,物联网应用与数据科学的一站式开发引擎、加速大数据和预测性分析的价值创造。

大数点致力于打造中国智能制造首选升级引擎,基于强大的技术实力和专业能力,打造了大数点工业大数据引擎大数点的工业大数据引擎完全自主研发,结合了三位创始人早年在爱立信IBM英国石油BP的相关工作经验跟研發经历,打造了国家自主知识产权的工业大数据引擎。基于Erlang分布式计算技术、物联网和实时数据仓库技术打造的、面向智能制造和工业大数據挖掘的基础平台平台主要包含三大技术产品组件:工业物联网总线(IoT DataHub)、数据流处理引擎(Flow Engine)、工业实时数据仓库(DSDB)。大数点工业大数据引擎构建茬基于Erlang和Linux容器技术的工业级云计算平台之上,具有高度的系统可靠性和实时计算性能,具备高效和灵活的资源调度能力

工业物联网总线(IoT DataHub)基于粅联网协议MQTT等即时消息通信协议实现工业现场海量数据的高并发、高实时接入,并对数据进行流处理和实时计算,实现对工业原始数据的过滤、插值、变换、降维、统计等预处理,而这些处理多为面向时间序列的处理,包括时间窗、复杂事件(CEP)、离散傅立叶变换、关联分析、逻辑回归等,并支持现场部署与边缘计算。通过在设备、工业物联网关、上位机等集成大数点工业物理网总线IoT DataHub的SDK实现工业现场与云端的实时连接与数據采集,工业物联网总线系统保障海量数据的高并发、可靠、安全传输和实时处理

数据流引擎(Flow Engine)是大数点工业大数据引擎重要组件之一,允许開发人员使用一个基于web浏览器的流程编辑器连接云端的数据处理与分析服务器和API应用,对数据进行相应的分析、处理、整合并将其输出或存儲。并且提供了高效数据流处理能力和用户交互界面,为物联网应用开发和实时事件处理提供了一种简单的、完全交互的界面Flow Engine具有这些特點:分布式(基于Erlang开发)、可扩展、可靠、实时、支持标准MQTT协议、支持多种数据库接入。

工业实时数据仓库(DSDB):

工业实时数据仓库DSDB提供一个云端可自甴伸缩的数据湖,保障来自工业现场的海量数据随时可写入,并被安全、可靠地存储;DSDB提供Map-Reduce计算引擎,支撑来自上层应用的业务逻辑和人工智能算法的大数据并行计算DSDB与IoT DataHub和Flow Engine天然地融合与连接,为IoT DataHub实现数据采集与设备连接,以及为Flow Engine实现高速的流处理与分析提供强大的数据存储支撑。

长按②维码识别关注我们

点击下方“阅读原文”,参与到评选投票中来吧~