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。

相关文章

Spring WebFlux 与 WebClient 使用指南及最佳实践

Spring WebFlux 与 WebClient 使用指南及最佳实践

《SpringWebFlux与WebClient使用指南及最佳实践》WebClient是SpringWebFlux模块提供的非阻塞、响应式HTTP客户端,基于ProjectReactor实现,... 目录Spring WebFlux 与 WebClient 使用指南1. WebClient 概述2. 核心依
阅读更多...
MyBatis-Plus 中 nested() 与 and() 方法详解(最佳实践场景)

MyBatis-Plus 中 nested() 与 and() 方法详解(最佳实践场景)

《MyBatis-Plus中nested()与and()方法详解(最佳实践场景)》在MyBatis-Plus的条件构造器中,nested()和and()都是用于构建复杂查询条件的关键方法,但... 目录MyBATis-Plus 中nested()与and()方法详解一、核心区别对比二、方法详解1.and()
阅读更多...
Spring Boot @RestControllerAdvice全局异常处理最佳实践

Spring Boot @RestControllerAdvice全局异常处理最佳实践

《SpringBoot@RestControllerAdvice全局异常处理最佳实践》本文详解SpringBoot中通过@RestControllerAdvice实现全局异常处理,强调代码复用、统... 目录前言一、为什么要使用全局异常处理？二、核心注解解析1. @RestControllerAdvice2
阅读更多...
Spring事务传播机制最佳实践

Spring事务传播机制最佳实践

《Spring事务传播机制最佳实践》Spring的事务传播机制为我们提供了优雅的解决方案,本文将带您深入理解这一机制,掌握不同场景下的最佳实践,感兴趣的朋友一起看看吧... 目录1. 什么是事务传播行为2. Spring支持的七种事务传播行为2.1 REQUIRED（默认）2.2 SUPPORTS2
阅读更多...
Java中的雪花算法Snowflake解析与实践技巧

Java中的雪花算法Snowflake解析与实践技巧

《Java中的雪花算法Snowflake解析与实践技巧》本文解析了雪花算法的原理、Java实现及生产实践,涵盖ID结构、位运算技巧、时钟回拨处理、WorkerId分配等关键点,并探讨了百度UidGen... 目录一、雪花算法核心原理1.1 算法起源1.2 ID结构详解1.3 核心特性二、Java实现解析2.
阅读更多...
MySQL 中 ROW_NUMBER() 函数最佳实践

MySQL 中 ROW_NUMBER() 函数最佳实践

《MySQL中ROW_NUMBER()函数最佳实践》MySQL中ROW_NUMBER()函数,作为窗口函数为每行分配唯一连续序号,区别于RANK()和DENSE_RANK(),特别适合分页、去重... 目录mysql 中 ROW_NUMBER() 函数详解一、基础语法二、核心特点三、典型应用场景1. 数据分
阅读更多...
深度解析Spring AOP @Aspect 原理、实战与最佳实践教程

深度解析Spring AOP @Aspect 原理、实战与最佳实践教程

《深度解析SpringAOP@Aspect原理、实战与最佳实践教程》文章系统讲解了SpringAOP核心概念、实现方式及原理,涵盖横切关注点分离、代理机制（JDK/CGLIB）、切入点类型、性能... 目录1. @ASPect 核心概念1.1 AOP 编程范式1.2 @Aspect 关键特性2. 完整代码实
阅读更多...
MySQL 用户创建与授权最佳实践

MySQL 用户创建与授权最佳实践

《MySQL用户创建与授权最佳实践》在MySQL中,用户管理和权限控制是数据库安全的重要组成部分,下面详细介绍如何在MySQL中创建用户并授予适当的权限,感兴趣的朋友跟随小编一起看看吧... 目录mysql 用户创建与授权详解一、MySQL用户管理基础1. 用户账户组成2. 查看现有用户二、创建用户1. 基
阅读更多...
Spring Boot 实现 IP 限流的原理、实践与利弊解析

Spring Boot 实现 IP 限流的原理、实践与利弊解析

《SpringBoot实现IP限流的原理、实践与利弊解析》在SpringBoot中实现IP限流是一种简单而有效的方式来保障系统的稳定性和可用性,本文给大家介绍SpringBoot实现IP限... 目录一、引言二、IP 限流原理2.1 令牌桶算法2.2 漏桶算法三、使用场景3.1 防止恶意攻击3.2 控制资源
阅读更多...
springboot项目中整合高德地图的实践

springboot项目中整合高德地图的实践

《springboot项目中整合高德地图的实践》：本文主要介绍springboot项目中整合高德地图的实践,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一：高德开放平台的使用二：创建数据库（我是用的是mysql）三：Springboot所需的依赖（根据你的需求再
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2