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。

相关文章

golang实现动态路由的项目实践

golang实现动态路由的项目实践

《golang实现动态路由的项目实践》本文主要介绍了golang实现动态路由项目实践,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习... 目录一、动态路由1.结构体（数据库的定义）2.预加载preload3.添加关联的方法一、动态路由1
阅读更多...
Vue 2 项目中配置 Tailwind CSS 和 Font Awesome 的最佳实践举例

Vue 2 项目中配置 Tailwind CSS 和 Font Awesome 的最佳实践举例

《Vue2项目中配置TailwindCSS和FontAwesome的最佳实践举例》：本文主要介绍Vue2项目中配置TailwindCSS和FontAwesome的最... 目录vue 2 项目中配置 Tailwind css 和 Font Awesome 的最佳实践一、Tailwind CSS 配置1. 安
阅读更多...
MyBatis分页插件PageHelper深度解析与实践指南

MyBatis分页插件PageHelper深度解析与实践指南

《MyBatis分页插件PageHelper深度解析与实践指南》在数据库操作中,分页查询是最常见的需求之一,传统的分页方式通常有两种内存分页和SQL分页,MyBatis作为优秀的ORM框架,本身并未提... 目录1. 为什么需要分页插件？2. PageHelper简介3. PageHelper集成与配置3.
阅读更多...
Spring Boot集成SLF4j从基础到高级实践(最新推荐)

Spring Boot集成SLF4j从基础到高级实践(最新推荐)

《SpringBoot集成SLF4j从基础到高级实践(最新推荐)》SLF4j(SimpleLoggingFacadeforJava)是一个日志门面（Facade）,不是具体的日志实现,这篇文章主要介... 目录一、日志框架概述与SLF4j简介1.1 为什么需要日志框架1.2 主流日志框架对比1.3 SLF4
阅读更多...
Spring Boot 常用注解详解与使用最佳实践建议

Spring Boot 常用注解详解与使用最佳实践建议

《SpringBoot常用注解详解与使用最佳实践建议》：本文主要介绍SpringBoot常用注解详解与使用最佳实践建议,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要... 目录一、核心启动注解1. @SpringBootApplication2. @EnableAutoConfi
阅读更多...
Redis实现分布式锁全解析之从原理到实践过程

Redis实现分布式锁全解析之从原理到实践过程

《Redis实现分布式锁全解析之从原理到实践过程》：本文主要介绍Redis实现分布式锁全解析之从原理到实践过程,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、背景介绍二、解决方案（一）使用 SETNX 命令（二）设置锁的过期时间（三）解决锁的误删问题（四）Re
阅读更多...
在 Spring Boot 中实现异常处理最佳实践

在 Spring Boot 中实现异常处理最佳实践

《在SpringBoot中实现异常处理最佳实践》本文介绍如何在SpringBoot中实现异常处理,涵盖核心概念、实现方法、与先前查询的集成、性能分析、常见问题和最佳实践,感兴趣的朋友一起看看吧... 目录一、Spring Boot 异常处理的背景与核心概念1.1 为什么需要异常处理？1.2 Spring B
阅读更多...
Spring Boot 整合 SSE的高级实践(Server-Sent Events)

Spring Boot 整合 SSE的高级实践(Server-Sent Events)

《SpringBoot整合SSE的高级实践(Server-SentEvents)》SSE（Server-SentEvents）是一种基于HTTP协议的单向通信机制,允许服务器向浏览器持续发送实... 目录1、简述2、Spring Boot 中的SSE实现2.1 添加依赖2.2 实现后端接口2.3 配置超时时
阅读更多...
Python使用getopt处理命令行参数示例解析(最佳实践)

Python使用getopt处理命令行参数示例解析(最佳实践)

《Python使用getopt处理命令行参数示例解析(最佳实践)》getopt模块是Python标准库中一个简单但强大的命令行参数处理工具,它特别适合那些需要快速实现基本命令行参数解析的场景,或者需要... 目录为什么需要处理命令行参数？getopt模块基础实际应用示例与其他参数处理方式的比较常见问http
阅读更多...
Java Optional的使用技巧与最佳实践

Java Optional的使用技巧与最佳实践

《JavaOptional的使用技巧与最佳实践》在Java中,Optional是用于优雅处理null的容器类,其核心目标是显式提醒开发者处理空值场景,避免NullPointerExce... 目录一、Optional 的核心用途二、使用技巧与最佳实践三、常见误区与反模式四、替代方案与扩展五、总结在 Java
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2