EMQ

杭州映云科技有限公司 (简称 EMQ) 是一家开源软件公司，专注 5G 和物联网领域的消息与流处理软件，面向全球运营。EMQ 总部位于杭州，在北京、上海、南京、重庆、昆明、硅谷、法兰克福设有分支机构，核心产品包括面向云和边缘计算的分布式 IoT 消息中间件和轻量级流式处理软件，累计服务全球 5000+ 企业，并与数十家世界 500 强企业达成战略合作，典型用户包括华为、中兴、HPE、VMware、Ericsson、中国移动、中国电信、国家电网、上汽大众、中国银联、长虹等。

EMQ 当前产品均为自主研发，拥有完全国内知识产权，同时 EMQ 主导着物联网领域具有全球影响力的开源项目。EMQ 致力于同上下游科技企业携手合作，融合全球智慧，共同推进 5G 时代开源软件与物联网产业的持续发展。

ActorCloud

ActorCloud 是 EMQ 推出的一个开源物联网平台，集成了物联网平台所需的设备多协议接入、基于 SQL 的轻量级流式物联网数据处理、可扩展的数据持久化等功能，并且向上提供了基本的设备管理 API 和参考 UI 实现，第三方可以基于此平台快速开发出满足定制需求的物联网平台。ActorCloud 借助 Pulsar 实现对消息的存储和流式处理，通过使用 Pulsar Functions 实现对数据的快速处理，并且通过 ActorCloud 对上层暴露的物联网 SQL 处理引擎来实现对物联网数据的分析与处理。

ActorCloud 面临的问题

ActorCloud 作为一个通用的物联网平台，需要实现数据接入、设备管理、数据存储和数据分析等功能，并且向上层提供编程接口以便开发者能够利用该平台快速开发出满足符合特定垂直行业的物联网应用。由于物联网本身具有接入设备（传感器）数目多，数据量大等特点，因此还要求 ActorCloud 能够支持水平扩展以支撑业务扩展等需求。

为什么选择 Pulsar

为了解决 ActorCloud 平台开发遇到的问题，我们希望找到一款高可用、分布式、能够灵活扩展的消息处理平台。经过深入研究，我们发现 Apache Pulsar 提供的如下特性完全能够满足我们的这些需求。

• 高可用、高扩展性、部署简单、易运维。
• 高吞吐：单个分区高达 1.8 M 消息/秒，这一特点完全符合我们数据量大的需求。
• Pulsar Functions 是一个轻量化的计算平台，能从一个或多个 Pulsar 主题中消费消息，把用户提供的处理逻辑应用于每个消息，把计算结果发布到另一个主题。Pulsar Functions 支持 Thread、Process、Kubernetes 等运行时，这为我们编写、运行和部署 Functions 提供了很好的灵活性，所以我们只用关心计算逻辑，无需处理复杂的配置或管理，更便捷地构建基于消息触发的流平台。
• 存储计算分离，IO 隔离，能够灵活处理数据，处理和存储可以独立扩展。

Apache Pulsar 高可用、灵活扩展的性能，及其 Pulsar Functions 和 IO 功能，能够帮助我们快速开发 ActorCloud 物联网平台。所以，我们最终采用 Apache Pulsar 来开发 ActorCloud 物联网平台。

Apache Pulsar 在 ActorCloud 上的应用

ActorCloud 把基于 SQL 的业务规则通过 API 的方式传入到数据处理规则管理引擎中，并将这些业务规则翻译为 Pulsar 中对应的 Source、Functions 和 Sink。Pulsar 的 Source 通过共享订阅的方式对接入 EMQ X Broker 设备数据进行消费，Pulsar 将这些数据进行持久化 ，并通过扩展 Pulsar 的 Functions 来对消息进行实时处理，处理完后通过 Sinks 将数据发送到相关的外部系统中。

规则样例

  
    { 
        "id": "mailTest",
        "sql": "SELECT temp FROM sensor WHERE temp > 0",
        "enabled": true,
        "actions": [{
            "mail": {
                "title": "temperature warning",
                "content": "temperature is ${temp} degrees, please take action promptly",
                "emails": [ "alert@emqx.io" ]
            }
        }]
    }
    

Pulsar Functions 的应用

Pulsar Functions 作为 Pulsar 的计算组件，是消息和计算平台在 Serverless 方向的一个融合和创新。Pulsar Functions 支持 Thread、Process、Kubernetes 等运行时，这为我们编写、运行和部署 Functions 提供了很好的灵活性，所以在 ActorCloud 物联网平台上，我们只需关注计算逻辑，无需处理复杂的配置或管理，就能更加便捷地构建基于消息触发的流平台。

• 为了更好地处理流和批的场景，ActorCloud 物联网平台利用了 Pulsar Functions 的 Window 特性，更加方便地实现物联网平台所需的时间窗口功能。目前，Pulsar Functions 支持 Count Window 和 Time Window 这两种形式。
• 在部署和管理方面，ActorCloud 利用 Pulsar Functions 的 API 和 pulsar-admin functions 命令（create，delete，update，restart，stop 和 get）进行增删改查等相应的操作，大大简化了 ActorCloud 部署与管理的复杂度。
• 为了实现灵活的订阅模式，ActorCloud 采用了 Pulsar Functions 提供的 Shared 订阅类型，来无限扩展消费能力。除了 Shared 订阅模式，Pulsar Functions 还支持 Failover 订阅模式。
• 在多语义保证方面，ActorCloud 采用了 Pulsar Functions 提供 At Least Once 的语义保证，确保每条发送出来的数据至少被处理一次。
• ActorCloud 利用 Pulsar 存储与计算分离的特性，在存储数据时通过配置消息保存（retention）策略，灵活选择物联网平台保存数据的时间。同时，Pulsar 集成了 Presto SQL 的相关功能，允许用户直接使用 Presto SQL 来查询存储在 BookKeeper 中的数据。ActorCloud 充分利用这一特性，使用 Presto SQL 查询实时数据和历史数据，来处理离线分析相关的业务。

Pulsar IO 的应用

Pulsar IO 分为输入（Input）和输出（Output）两个模块，输入代表数据从哪里来，通过 Source 实现数据输入。输出代表数据要往哪里去，通过 Sink 实现数据输出。ActorCloud 项目充分利用 Pulsar IO 的灵活性，实现了多个 Source 和 Sink，来完成不同的工作。

1. EMQ source

   从 EMQ 中拉取数据，写到 Pulsar Topic 中，负责将外部 EMQ 中的数据同步到 Pulsar 中来。

2. Mail sink

   从 Pulsar 的 Topic 中接收数据，发送邮件。

3. Publish sink

   从 Pulsar Topic 接收数据，利用初始化好的 HttpClient 将数据投递到外部系统中。

4. DB sink

   从 Pulsar Topic 中接收数据，然后将数据发送到外部系统中，该 Sink 封装了 JDBC，当前支持将 Pulsar 中的数据插入到 SQLite、MySQL 和 PostgreSQL 中。

5. EMQ sink

   从 Pulsar Topic 中接收数据，将数据发布到 EMQ X 的主题中。

Pulsar 测试结果

通过测试，ActorCloud 物联网平台可以在双机 Pulsar 部署， 8 核 CPU、16GB 内存的配置下，实现后端每秒 6 万物联网消息处理的吞吐量，通过增加节点的方式来实现水平扩展。

总结

通过结合使用 EMQ X 和 Apache Pulsar，ActorCloud 实现了物联网设备数据接入、设备管理、数据存储和数据分析等功能，并且提供了灵活的编程接口以便快速开发出满足符合特定垂直行业的物联网应用。通过 EMQ X 和 Apache Pulsar 提供的水平扩展能力，ActorCloud 实现了设备接入和数据处理的水平扩展。