EMQX 实践

2024-03-18 14:36
文章标签 实践 emqx

本文主要是介绍EMQX 实践，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

MQTT 核心概念

发布订阅

MQTT 基于发布订阅模式，它解耦了消息的发送方（发布者）和接收方（订阅者），引入了一个中间代理的角色来完成消息的路由和分发。发布者和订阅者不需要知道彼此的存在，他们之间唯一的联系就是对消息的一致约定，例如消息将使用什么主题、消息将包含哪些字段等等。这让 MQTT 的通信更加灵活，因为我们可以随时动态地增加或减少订阅者和发布者。通过发布订阅，我们可以轻易地实现消息的广播、组播和单播。

服务端

在发布消息的客户端和订阅的客户端之间充当中介，将所有接收到的消息转发到匹配的订阅客户端。所以有时我们也会直接将服务端称为 Broker。

客户端

使用 MQTT 协议连接到 MQTT 服务端的设备或应用程序。它既可以是发布者，也可以是订阅者，也可以具备这两种身份。

主题

主题被用来标识和区分不同的消息，它是 MQTT 消息路由的基础。发布者可以在发布时指定消息的主题，订阅者则可以选择订阅自己感兴趣的主题来接收相关的消息。

通配符

订阅者可以在订阅的主题中使用通配符来达到一次订阅多个主题的目的。MQTT 提供了单层通配符和多层通配符两种主题通配符，以满足不同的订阅需要。

QoS

MQTT 定义了三种 QoS 等级，来分别提供不同的消息可靠性保证。每条消息都可以在发布时独立设置自己的 QoS。QoS 0 最多交付一次，消息可能丢失；QoS 1 至少交付一次，消息可以保证到达，但是可能重复；QoS 2 只交付一次，消息保证到达，并且不会重复。QoS 越大，消息的传输复杂程度也越高，我们需要根据实际场景来选择合适的 QoS。

会话

QoS 只是设计了消息可靠到达的理论机制，而会话则确保了 QoS 1、2 的协议流程得以真正实现。会话是客户端与服务端之间的有状态交互，它可以仅持续和网络连接一样长的时间，也可以跨越多个网络连接存在，我们通常将后者称为持久会话。我们可以选择让连接从已存在的会话中恢复，也可以选择从一个全新的会话开始。

保留消息

与普通消息不同，保留消息可以保留在 MQTT 服务器中。任何新的订阅者订阅与该保留消息中的主题匹配的主题时，都会立即接收到该消息，即使这个消息是在它们订阅主题之前发布的。这使订阅者在上线后可以立即获得数据更新，而不必等待发布者再次发布消息。在某种程度上，我们可以把保留消息当作是一个消息 “云盘” 来使用：随时上传消息到 “云盘”，然后在任意时刻从 “云盘” 获取消息。当然，这个 “云盘” 还有一个主题下只能存储一条最新的保留消息的限制。

遗嘱消息

发布订阅模式的特性决定了，除了服务器以外没有客户端能够感知到某个客户端从通信网络中离开。而遗嘱消息则为连接意外断开的客户端提供了向其他客户端发出通知的能力。客户端可以在连接时向服务器设置自己的遗嘱消息，服务器将在客户端异常断开后立即或延迟一段时间后发布这个遗嘱消息。而订阅了对应遗嘱主题的客户端，将收到这个遗嘱消息，并且采取相应的措施，例如更新该客户端的在线状态等等。

共享订阅

默认情况下，消息会被转发给所有匹配的订阅者。但有时，我们可能希望多个客户端协同处理接收到的消息，以便以水平扩展的方式来提高负载能力。又或者，我们希望为客户端增加一个备份客户端，当主客户端离线时，能够无缝切换到备份客户端继续接收消息，以确保高可用性。而 MQTT 的共享订阅特性，则提供了这一能力。我们可以将客户端划分为多个订阅组，消息仍然会被转发给所有订阅组，但每个订阅组内每次只会有一个客户端收到消息。

MQTT选型

MQTT BROKER 技术选型

EMQX安装

本地开发环境

可以选择安装Windows版本
Windows安装EMQ X
官方Windows部署
安装成功后，直接访问http://localhost:18083/
账号/密码：admin/public
image.png

生产环境

EMQX 本身支持分布式集群架构，能够在保证高可用性、容错性和可扩展性的同时，处理大量的客户端和消息。通过使用 EMQX 集群，您可以在一个或多个节点发生故障时仍然保持集群运行，从而享受到容错和高可用性的好处。
相比与之前版本，EMQX 5.0 集群采用了新的 Mria 集群架构，单节点能支持 500 万 MQTT 设备连接，集群可扩展至 1 亿并发 MQTT 连接。官方集群部署
image.png

安全指南

网络与 TLS
介绍了 EMQX 如何支持端对端加密通信，包括如何启用 SSL/TLS 连接和获取 SSL/TLS 证书。
认证
身份认证是物联网应用的重要组成部分，可以帮助有效阻止非法客户端的连接。为了提供更好的安全保障，EMQX 支持多种认证机制，如 X.509 证书认证、密码认证、JWT 认证、基于 MQTT 5.0 协议的增强认证以及 PSK 认证。本节介绍了这些认证机制的工作方式和配置方法。
授权
在 EMQX 中，授权是指对 MQTT 客户端的发布和订阅操作进行权限控制。本节将介绍如何通过内置数据库、文件、或通过集成 MySQL、PostgreSQL、MongoDB 和 Redis 进行授权相关操作。
黑名单
EMQX 为用户提供了黑名单功能，用户可以通过 Dashboard 和 HTTP API 将指定客户端加入黑名单以拒绝该客户端访问，除了客户端标识符以外，还支持直接封禁用户名甚至 IP 地址。
连接抖动检测
EMQX 支持自动封禁那些被检测到短时间内频繁登录的客户端，并且在一段时间内拒绝这些客户端的登录，以避免此类客户端过多占用服务器资源而影响其他客户端的正常使用。

认证

EMQX Dashboard 提供了开箱即用的认证与权限管理功能，用户仅通过用户界面，就可以快速实现客户端认证授权机制的配置，无需编写代码或手动编辑配置文件，即可对接各类数据源与认证服务，实现各个级别与各类场景下的安全配置，以更高的开发效率获得更安全的保障。

创建认证

在认证页面下的右上角，点击 创建 按钮，即可进入到创建认证的页面。创建一个认证需要选择一种认证方式，选择完成后需要选择一个存储或获取认证信息的数据源（JWT 认证方式除外），认证数据可以从这些数据源包括数据库或 HTTP 服务中获取，最后再配置连接到该数据源的连接信息即可。
认证方式：Password-Based，使用客户端 ID 或用户名加密码的认证方式；
image.png
数据源选择：redis
image.png
选择加密方式及加盐方式：加密方式md5 ，加盐方式prefix
image.png
初始化数据到redis:
HMSET “mqtt_user:username” “password_hash” “66ace8890090c2a50e729318d45fe53b” “salt” “abc”

验证

image.png

http签名配置

创建API秘钥

image.png
image.png

记录秘钥

appId: *************
appSecret: *************

MQTT通用组件开发

源码地址

目录

├─component-mqtt-client
└─component-mqtt-client-starter

component-mqtt-client

mqtt上下文

image.png

建立连接
public MqttClientApp connect() {countDownLatch = new CountDownLatch(1);Vertx.vertx().deployVerticle(this);return this;
}
接收消息
 @Overridepublic void start() {if (Objects.isNull(this.mqttClient)) {this.mqttClient = MqttClient.create(vertx, createMqttClientOptions());}//接收服务端消息处理handlermqttClient.publishHandler(pub -> {Buffer buffer = pub.payload();String topicName = pub.topicName();String[] split = topicName.split("/");String string = buffer.toString(StandardCharsets.UTF_8);UpMessage upRawMessage = new UpMessage();HashMap<String, Object> headers = Maps.newHashMap();headers.put("topic",topicName);headers.put("qos",pub.qosLevel().value());upRawMessage.setHeaders(headers);upRawMessage.setMessageContent(string);upRawMessage.setProductKey(split[0]);upRawMessage.setDeviceId(split[1]);mqttListenerList.forEach(f -> {String topic = f.getTopic();String[] listenerTopic = topic.split("/");boolean flag = true;for (int i = 0; i < split.length; i++) {if (allWildcard.equals(listenerTopic[i])) {break;}if (singleWildcard.equals(listenerTopic[i])) {continue;}if (!split[i].equals(listenerTopic[i])) {flag = false;break;}}if (flag){f.onMessage(upRawMessage);}});});mqttClient.closeHandler(unused -> getVertx().setTimer(RECONNECT_INTERVAL, h -> start()));mqttClient.connect(mqttConfig.getListenerInfos().getPort(), mqttConfig.getListenerInfos().getHost(),s -> {if (s.succeeded()) {log.info("MqttClient connect success.");subscribe();countDownLatch.countDown();} else {log.error("MqttClient connect fail: ", s.cause());if (s.cause() != null) {vertx.setTimer(RECONNECT_INTERVAL, handler -> this.start());}}});}
长连接推送消息
public MqttResp publish(MqttReq request) {
MqttResp response = new MqttResp();
Buffer payload = Buffer.buffer(request.getMessageContent());
mqttClient.publish(request.getTopic(), payload, MqttQoS.valueOf(request.getQos()), false, false, s -> {if (s.succeeded()) {log.info("===>MqttClient publish success[{}]", s.result());} else {log.error("===>MqttClient publish fail.", s.cause());}
});
response.setCode(200);
return response;
}
http推送消息
public Map<String, ?> callHttp(MqttReq params) {
String path = "";
String url = config().getAddress() + path;
log.debug("http url[{}] requestBodyStr[{}]", url, params.getMessageContent());Dict dict = Dict.create();
dict.set("topic", params.getTopic());              //订阅主题
dict.set("payload", params.getMessageContent());   //内容
dict.set("qos", 0);                                //质量
dict.set("retain",false);                          //是否保存
String requestBodyStr = JSON.toJSONString(dict);RequestBody requestBody = RequestBody.create(HTTP_MEDIA_TYPE_JSON_UTF8, requestBodyStr);
Request request = new Request.Builder()
.url(url)
.post(requestBody)
.header("Content-Type", "application/json")
.header("Authorization", Credentials.basic(config().getAppId(), config().getAppSecret()))
.build();try (Response response = getHttpClientInstance().newCall(request).execute()) {log.debug("Call http success. url[{}] response[{}]", url, response);if (response.code() == 404) {return ImmutableMap.of("code", 404, "Message", "404 Not Found");} else if (!response.isSuccessful()) {return ImmutableMap.of("code", response.code(), "Message", "Server Error");}// 输出响应内容assert response.body() != null;String string = response.body().string();return JSON.parseObject(string);
} catch (IOException e) {log.warn("Call http failed, {}. url[{}] requestBodyStr[{}]", e.getMessage(), url, requestBodyStr);
}return Collections.emptyMap();
}
Mqtt配置信息

image.png

public class MqttConfig {private String appId;private String appSecret;private String address;private String username;private String password;private ListenerInfo listenerInfos;@Data@NoArgsConstructor@AllArgsConstructorpublic static class ListenerInfo {private String host;private int port;private boolean ssl;//订阅的topicprivate List<String> subscribeTopics;}}
监听信息接口

image.png

public interface MqttListener {void setTopic(String topic);String getTopic();void onMessage(Message message);
}

component-mqtt-client-starter

MqttClientAutoConfiguration

image.png

META-INF

image.png

com.gitee.xmhzzz.component.mqtt.client.MqttClientAutoConfiguration

Mqtt服务实战demo

通过component-mqtt-client-starter快速构建mqtt-service服务
image.png

发生消息

public class MqttController {@Autowiredprivate IMqttApi IMqttApi;@PostMapping("/pub/tcp")public void pubTcp(){MqttReq mqttReq = new MqttReq();mqttReq.setTopic("topicA/001/in");Map<String, Object> map = Maps.newHashMap();map.put("1","o");mqttReq.setData(map);IMqttApi.tcpPub(mqttReq);}
}

监听消息

@Slf4j
@Component
public class AMqttListener implements MqttListener {private String topic;public AMqttListener() {this.topic = "topicA/+/msg";}@Overridepublic void setTopic(String topic) {}@Overridepublic String getTopic() {return this.topic;}@Overridepublic void onMessage(Message message) {log.info("a message[{}]", JSONObject.toJSONString(message));}
}

这篇关于EMQX 实践的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/822643。

相关文章

基于MySQL Binlog的Elasticsearch数据同步实践

基于MySQL Binlog的Elasticsearch数据同步实践

一、为什么要做 随着马蜂窝的逐渐发展，我们的业务数据越来越多，单纯使用 MySQL 已经不能满足我们的数据查询需求，例如对于商品、订单等数据的多维度检索。 使用 Elasticsearch 存储业务数据可以很好的解决我们业务中的搜索需求。而数据进行异构存储后，随之而来的就是数据同步的问题。 二、现有方法及问题 对于数据同步，我们目前的解决方案是建立数据中间表。把需要检索的业务数据，统一放到一张M
阅读更多...
系统架构师考试学习笔记第三篇——架构设计高级知识（20）通信系统架构设计理论与实践

系统架构师考试学习笔记第三篇——架构设计高级知识（20）通信系统架构设计理论与实践

本章知识考点：         第20课时主要学习通信系统架构设计的理论和工作中的实践。根据新版考试大纲,本课时知识点会涉及案例分析题(25分),而在历年考试中,案例题对该部分内容的考查并不多,虽在综合知识选择题目中经常考查,但分值也不高。本课时内容侧重于对知识点的记忆和理解,按照以往的出题规律,通信系统架构设计基础知识点多来源于教材内的基础网络设备、网络架构和教材外最新时事热点技术。本课时知识
阅读更多...
Prometheus与Grafana在DevOps中的应用与最佳实践

Prometheus与Grafana在DevOps中的应用与最佳实践

Prometheus 与 Grafana 在 DevOps 中的应用与最佳实践 随着 DevOps 文化和实践的普及，监控和可视化工具已成为 DevOps 工具链中不可或缺的部分。Prometheus 和 Grafana 是其中最受欢迎的开源监控解决方案之一，它们的结合能够为系统和应用程序提供全面的监控、告警和可视化展示。本篇文章将详细探讨 Prometheus 和 Grafana 在 DevO
阅读更多...
springboot整合swagger2之最佳实践

springboot整合swagger2之最佳实践

来源:https://blog.lqdev.cn/2018/07/21/springboot/chapter-ten/ Swagger是一款RESTful接口的文档在线自动生成、功能测试功能框架。 一个规范和完整的框架，用于生成、描述、调用和可视化RESTful风格的Web服务，加上swagger-ui，可以有很好的呈现。 SpringBoot集成 pom <!--swagge
阅读更多...
vue2实践：el-table实现由用户自己控制行数的动态表格

vue2实践：el-table实现由用户自己控制行数的动态表格

需求 项目中需要提供一个动态表单，如图： 当我点击添加时，便添加一行；点击右边的删除时，便删除这一行。 至少要有一行数据，但是没有上限。 思路 这种每一行的数据固定，但是不定行数的，很容易想到使用el-table来实现，它可以循环读取:data所绑定的数组，来生成行数据，不同的是： 1、table里面的每一个cell，需要放置一个input来支持用户编辑。 2、最后一列放置两个b
阅读更多...
【HarmonyOS】-TaskPool和Worker的对比实践

【HarmonyOS】-TaskPool和Worker的对比实践

ArkTS提供了TaskPool与Worker两种多线程并发方案，下面我们将从其工作原理、使用效果对比两种方案的差异，进而选择适用于ArkTS图片编辑场景的并发方案。 TaskPool与Worker工作原理 TaskPool与Worker两种多线程并发能力均是基于 Actor并发模型实现的。Worker主、子线程通过收发消息进行通信；TaskPool基于Worker做了更多场景化的功能封装，例
阅读更多...
vue2实践：第一个非正规的自定义组件-动态表单对话框

vue2实践：第一个非正规的自定义组件-动态表单对话框

前言 vue一个很重要的概念就是组件，作为一个没有经历过前几代前端开发的我来说，不太能理解它所带来的“进步”，但是，将它与后端c++、java类比，我感觉，组件就像是这些语言中的类和对象的概念，通过封装好的组件（类），可以通过挂载的方式，非常方便的调用其提供的功能，而不必重新写一遍实现逻辑。 我们常用的element UI就是由饿了么所提供的组件库，但是在项目开发中，我们可能还需要额外地定义一
阅读更多...
《C++中的移动构造函数与移动赋值运算符：解锁高效编程的最佳实践》

《C++中的移动构造函数与移动赋值运算符：解锁高效编程的最佳实践》

在 C++的编程世界中，移动构造函数和移动赋值运算符是提升程序性能和效率的重要工具。理解并正确运用它们，可以让我们的代码更加高效、简洁和优雅。 一、引言 随着现代软件系统的日益复杂和对性能要求的不断提高，C++程序员需要不断探索新的技术和方法来优化代码。移动构造函数和移动赋值运算符的出现，为解决资源管理和性能优化问题提供了有力的手段。它们允许我们在不进行不必要的复制操作的情况下，高效地转移资源
阅读更多...
Vue3+elementplus实现图片上传下载（最强实践）

Vue3+elementplus实现图片上传下载（最强实践）

图片上传子组件： 实现照片的上传，预览，以及转成以逗号隔开的图片地址，即时监听，并发送消息到父组件。 <!-- ImageUploader.vue --> <template><div><el-upload class="avatar-uploader" :http-request="customUpload" :before-upload="beforeUpload":show-fil
阅读更多...
【知识分享】MQTT实战-使用mosquitto客户端连接emqx服务器

【知识分享】MQTT实战-使用mosquitto客户端连接emqx服务器

一、简介     MQTT（Message Queuing Telemetry Transport）是一种轻量级的、基于发布/订阅模式的通信协议，旨在实现物联网设备之间的低带宽、高延迟的通信。MQTT协议设计简洁，使用TCP/IP协议进行通信，适用于各种网络环境，尤其适合在有限的网络带宽和不稳定的网络连接条件下进行通信。     MQTT的工作原理是基于发布/订阅模式的消息传递，它包括两个主要
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2