Emqtt

emqx broker装置Emq x百万级开源 MQTT 音讯服务器 是基于 Erlang/OTP 语言平台开发 一款齐全开源，高可用低时延的百万级分布式物联网 MQTT 5.0 音讯服务器 官网地址: https://www.emqx.com/zh Centos7 装置#下载Centos7 amd64位版本wget https://www.emqx.com/zh/downloads/broker/4.3.9/emqx-centos7-4.3.9-amd64.zip#解压unzip emqx-centos7-4.3.9-amd64.zip#进入目录cd emqx./bin/emqx start 启动./bin/emqx_ctl status 查看状态./bin/emqx stop 进行./bin/emqx restart 重启拜访Dashboardhttp://localhost:18083/默认用户名/明码:admin/public Docker 装置#下载镜像docker pull emqx/emqx:4.3.9#启动容器docker run -d --name emqx --privileged=true -p 1883:1883 -p 8081:8081 -p 8083:8083 -p 8084:8084 -p 8883:8883 -p 18083:18083 emqx/emqx:4.3.9拜访Dashboardhttp://localhost:18083/默认用户名/明码:admin/public

本文形容如何通过云原生利用治理平台 Rainbond 一键装置高可用 EMQX 集群。这种形式适宜不太理解 Kubernetes、容器化等简单技术的用户应用，升高了在 Kubernetes 中部署 EMQX 的门槛。 一、背景信息1.1 Rainbond 与 EMQX 的联合Rainbond 是一款易于应用的开源云原生利用治理平台。借助于它，用户能够在图形化界面中实现微服务的部署与运维。借助 Kubernetes 和容器化技术的能力，将故障自愈、弹性伸缩等自动化运维能力赋能给用户的业务。 这类用户不用再关怀如何部署 EMQX 集群，通过开源利用商店一键装置 EMQX 高可用集群，这种装置形式极大的升高了用户应用 EMQX 集群的部署累赘，目前反对最新版本 4.4.3。 1.2 对于利用模版利用模版是面向 Rainbond 云原生利用治理平台的安装包，用户能够基于它一键装置业务零碎到本人的 Rainbond 中去。无论这个业务零碎如许简单，利用模版都会将其形象成为一个利用，裹挟着利用内所有组件的镜像、配置信息以及所有组件之间的关联关系一并装置起来。 二、前提条件部署好的 Rainbond 云原生利用治理平台：例如 疾速体验版本，能够在集体 PC 环境中以启动一个容器的代价运行。能够连贯到互联网。三、疾速开始3.1 拜访内置的开源利用商店抉择左侧的 利用市场 标签页，在页面中切换到 开源利用商店 标签页，搜寻关键词 EMQX 即可找到 EMQX 利用。 3.2 一键装置点击 EMQX 右侧的 装置 能够进入装置页面，填写简略的信息之后，点击 确定 即可开始装置，页面主动跳转到拓扑视图。 参数阐明： 选择项阐明团队名称用户自建的工作空间，以命名空间隔离集群名称抉择 EMQX 被部署到哪一个 K8s 集群抉择利用抉择 EMQX 被部署到哪一个利用，利用中蕴含有若干有关联的组件利用版本抉择 EMQX 的版本，目前可选版本为 4.4.3期待几分钟后，EMQX 集群就会装置实现，并运行起来。 ...

MongoDB 介绍非关系数据库（NoSQL） 用于超大规模数据的存储，例如谷歌或 Facebook 每天为他们的用户收集万亿比特的数据。这些类型的数据存储不需要固定的模式，无需多余操作就可以横向扩展。MongoDB 是一个介于关系数据库和非关系数据库之间的产品，是非关系数据库当中功能最丰富，最像关系数据库的。MongoDB 由 C++ 语言编写的，是一个基于分布式文件存储的开源数据库系统，MongoDB 旨在为数据存储提供可扩展的高性能数据存储解决方案，在高负载的情况下，可以轻松添加更多的节点保证服务性能。 MongoDB 将数据存储为一个文档，数据结构由键值 (key=>value) 对组成。MongoDB 文档类似于 JSON 对象。字段值可以包含其他文档，数组及文档数组。 MongoDB 下载地址：https://www.mongodb.com/download-center/community 场景介绍该场景需要将 EMQ X 指定主题下满足某条件的消息存储到 MongoDB 数据库。为了便于后续分析检索，消息内容需要进行拆分存储。 该场景下设备端上报信息如下： 上报主题：cmd/state/:id，主题中 id 代表车辆客户端识别码消息体： { "id": "NXP-058659730253-963945118132721-22", // 客户端识别码 "speed": 32.12, // 车辆速度 "direction": 198.33212, // 行驶方向 "tachometer": 3211, // 发动机转速，数值大于 8000 时才需存储 "dynamical": 8.93, // 瞬时油耗 "location": { // GPS 经纬度数据 "lng": 116.296011, "lat": 40.005091 }, "ts": 1563268202 // 上报时间}当上报数据发动机转速数值大于 8000 时，存储当前信息以便后续分析用户车辆使用情况。 ...

物联网设备终端种类繁杂，各厂商使用的编码格式各异，所以在接入物联网平台的时候就产生了统一数据格式的需求，以便平台之上的应用进行设备管理。 EMQ X 企业版 3.4.0 提供了 Schema Registry 功能，提供编解码能力。Schema Registry 管理编解码使用的 Schema、处理编码或解码请求并返回结果。Schema Registry 配合规则引擎，可适配各种场景的设备接入和规则设计。 数据格式下图展示了 Schema Registry 的一个应用案例。多个设备上报不同格式的数据，经过 Schema Registry 解码之后，变为统一的内部格式，然后转发给后台应用。 [图1: 使用 Schema Registry 对设备数据进行编解码] 二进制格式支持EMQ X 3.4.0 内置的 Schema Registry 数据格式包括 Avro 和 Protobuf。Avro 和 Protobuf 是依赖 Schema 的数据格式，编码后的数据为二进制，使用 Schema Registry 解码后的内部数据格式(Map，稍后讲解) 可直接被规则引擎和其他插件使用。此外 Schema Registry 支持用户自定义的 (3rd-party) 编解码服务，通过 HTTP 或 TCP 回调的方式，进行更加贴近业务需求的编解码。 架构设计Schema Registry 为 Avro 和 Protobuf 等内置编码格式维护 Schema 文本，但对于自定义编解码 (3rd-party) 格式，如需要 Schema，Schema 文本需由编解码服务自己维护。Schema Registry 为每个 Schema 创建一个 Schema ID，Schema API 提供了通过 Schema ID 的添加、查询和删除操作。 ...

又回到erlang了，使用了一段时间的golang再回到erlang有点那么的亲切感。在项目中也准备用mqtt来做消息上报，顺道就想看下他的代码。erlang中application都是通过supervisor来管理的，在emq中emqttd_sup是一个最大的supervisor，他下面面又连接了很多的supervisor或者worker。ekka:start()emqttd_sup|——–>emqttd_ctl 负责从emqttd_ctl命令过来的rpc handler|——–>emqttd_hooks(hook 函数的处理)|——–>emqttd_router(各node之间的消息路由)|——–>emqttd_pubsub_sup(管理pubsub相关的supervisor)|——–>emqttd_pool_sup(emqttd_pubsub的supervisor)gproc_pool |—–>emqttd_pubsub_1(worker) |—–>emqttd_pubsub_2(worker)|——–>emqttd_pool_sup(emqttd_server的supervisor)gproc_pool |—–>emqttd_server_1(worker) |—–>emqttd_server_2(worker)|———>emqttd_stats(stats topic相关的统计)|———>emqttd_stats(metrics topic相关的统计)|———>emqttd_pool_sup(pooler没看到哪里用到了这快)gproc_pool |——->pooler_1(worker) |——->pooler_2(worker)|———>emqttd_sm_sup( session management supervisor)gproc_pool |——->emqttd_sm_1(worker) |——->emqttd_sm_2(worker)|———>emqttd_ws_client_sup(websocket client supervisor)gproc_pool |——->emqttd_ws_client_1(worker) |——->emqttd_ws_client_2(worker)|———>emqttd_broker(broker统计相关handler)|———>emqttd_alarm(系统alerm相关的handler)|———>emqttd_mod_sup(管理外部mod的supervisor)|———>emqttd_bridge_sup_sup(bridge supervisor)|———>emqttd_access_control(auth/acl相关管理模块)|———>emqttd_sysmon_sup(system monitor supervisor) |——–>emqttd_sysmon(vm system monitor)register_acl_mod()start_listener()之后就开始socket监听了，等待新的连接到来。erlang的优势在于他又一套完善的process 监控系统。具体可以参考这里, 子进程退出后supervisor会给你自动重启。