SpringBoot如何实现EMQ设备的上下线告警

互联网
前言 上下线通知我遇到了一个难题，即在使用 EMQ X 4 4 10 的开源版本时，我需要实现设备的上下线状态监控，但该 4 4 10开源版本 并未内置设备上下

前言

上下线通知

我遇到了一个难题，即在使用 EMQ X 4.4.10 的开源版本时，我需要实现设备的上下线状态监控，但该 4.4.10开源版本 并未内置设备上下线提醒模块，只有企业版才内置了该模块。这为我带来了一些技术上的难题，迫使我必须另辟蹊径，通过其他方法来监听设备的上下线状态。为了解决这个问题，我采取了以下步骤：

首先，我修改了 EMQ X 的 acl.config 文件，添加了以下规则：

{allow, all, subscribe, ["$SYS/brokers/+/clients/#"]}.

图示：

这个规则允许订阅 $SYS/brokers/+/clients/# 主题的所有客户端。

接下来，我使用 Spring Boot 创建了一个应用程序，其中我设置了与 EMQ X 代理的连接。在这个应用程序中，我创建了一个监听器，用于订阅 $SYS/brokers/+/clients/# 主题，以感知设备的上下线信息。

通过这个解决方案，我能够实时监控设备的连接和断开事件，而不受 EMQ X 开源版本的功能限制。这使我能够根据设备状态采取适当的措施，比如发送通知或执行其他自定义操作。这个方法允许我灵活地定制解决方案，以满足我的特定需求，而无需依赖 EMQ X 的功能。

EMQ简介

EMQ （ Erlang MQTT Broker ）是一种基于 Erlang 编程语言开发的开源消息传递代理（ MQTT broker ），专门用于支持 MQTT （ Message Queuing Telemetry Transport ）协议。MQTT是一种轻量级、高效的消息传递协议，最初设计用于连接受限的设备，如嵌入式系统和物联网设备。 EMQ 作为 MQTT broker ，提供了可靠的消息传递机制，使设备能够相互通信，同时也可与后端应用程序集成，使其成为物联网生态系统中的重要组成部分。

环境

  • EMQX 安装方式： Docker
  • EMQX 版本： 4.4.10开源版本
  • 操作系统： CentOS 7

EMQX4.4版本官方文档

下载

下载 EMQX

准备工作

安装EMQX

修改EMQX的ACL规则内容

EMQX的Docker容器配置文件所在目录

# 进入EMQX容器
docker exec -it emqx /bin/sh
# 进入配置文件目录
cd /opt/emqx/etc
# 查看acl配置文件
cat acl.conf
# 编辑acl配置文件
vi acl.conf

非Docker容器配置文件所在目录

# 进入配置文件目录
cd /etc/emqx
# 查看acl配置文件
cat acl.conf
# 编辑acl配置文件
vi acl.conf

acl的默认文件内容

%%--------------------------------------------------------------------
%% [ACL](https://docs.emqx.io/broker/v3/en/config.html)
%%
%% -type(who() :: all | binary() |
%%                {ipaddr, esockd_access:cidr()} |
%%                {ipaddrs, [esockd_access:cidr()]} |
%%                {client, binary()} |
%%                {user, binary()}).
%%
%% -type(access() :: subscribe | publish | pubsub).
%%
%% -type(topic() :: binary()).
%%
%% -type(rule() :: {allow, all} |
%%                 {allow, who(), access(), list(topic())} |
%%                 {deny, all} |
%%                 {deny, who(), access(), list(topic())}).
%%--------------------------------------------------------------------
{allow, {user, "dashboard"}, subscribe, ["$SYS/#"]}.
{allow, {ipaddr, "127.0.0.1"}, pubsub, ["$SYS/#", "#"]}.
{deny, all, subscribe, ["$SYS/#", {eq, "#"}]}.
{allow, all}.

新增一条ACL规则

allow, all, subscribe, ["$SYS/brokers/+/clients/#"]}.
  • allow : 表示这是一个允许（ allow ）访问的规则。这意味着与此规则匹配的操作将被允许。
  • all : 表示这个规则适用于所有的客户端。
  • subscribe : 表示这个规则定义了对主题的订阅权限。
  • $SYS/brokers/+/clients/# : 这是一个主题过滤器，它指定了匹配的主题模式。在这里，$SYS/brokers/+/clients/# 表示以 $SYS/brokers/ 开头，然后是一个通配符 +，它可以匹配单个层级的任何名称，接着是 clients/ ，最后又有一个 # 通配符，它可以匹配零个或多个层级的名称。因此，这个主题过滤器匹配了所有以 $SYS/brokers/ 开头，然后紧跟着 clients/ 的主题。

综合起来，这个规则允许所有的客户端订阅以 $SYS/brokers/ 开头，然后跟着 clients/ 的所有主题。通常，这种规则用于允许所有客户端订阅系统级别的信息或监控数据，如经纪人（ Broker ）的客户端连接状态等。这可以用于监视和诊断 MQTT Broker 的性能和状态。

注意：修改完毕之后使用 emqx_ctl reload_acl 重新加载acl规则或者直接重启emqx服务

搭建一个能够监听EMQX主题的Spring Boot应用程序

MQTT相关依赖

<!-- MQTT相关依赖 -->
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-stream</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-mqtt</artifactId>
</dependency>

MQTT接受订阅的主题

$SYS/brokers/+/clients/#

处理设备上下线事件

获取 EMQX 消息主题

// 从消息请求头中获取消息主题topic
String topic = String.valueOf(message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC));

获取topic最后的节点字符串

以下方式通过主题 topic 来获取 ClientId 

// topic最后的节点字符串
String lastPart = extractLastPart(topic);
// 获取消息内容
String payload = StrUtil.str((byte[]) message.getPayload(), "utf-8");
// 判断设备是上线或下线消息
if ("connected".equals(lastPart)) {
    // 设备上线消息
    clientId = extractClientIdFromTopic(topic);
    log.info("设备上线提醒 -> IMEI:{}", clientId);
} else if ("disconnected".equals(lastPart)) {
    // 设备下线消息
    clientId = extractClientIdFromTopic(topic);
    log.info("设备下线警告 -> IMEI:{}", clientId);
}
/**
 * 获取最后一个节点
 *
 * @param topic 主题
 * @return 节点内容
 */
public static String extractLastPart(String topic) {
    // 使用split方法将字符串根据'/'分割成数组
    String[] parts = topic.split("/");
    // 获取最后一个元素
    String lastPart = parts[parts.length - 1];
    return lastPart;
}
/**
 * 从Topic中提取ClientId
 *
 * @param topic 主题
 * @return ClientId
 */
public static String extractClientIdFromTopic(String topic) {
    // 使用正则表达式匹配主题中的ClientId
    String regex = "\\$SYS/brokers/[^/]+/clients/([^/]+)/(connected|disconnected)";
    Pattern pattern = Pattern.compile(regex);
    Matcher matcher = pattern.matcher(topic);
    // matcher.groupCount() 是一个方法，用于获取正则表达式中定义的组数。在正则表达式中，使用括号 () 来定义捕获组。在这个情况下，正则表达式 \\$SYS/brokers/[^/]+/clients/([^/]+)/(connected|disconnected) 中有两组，分别是括号内的 ([^/]+) 部分和 (connected|disconnected) 部分。matcher.groupCount() 返回的是正则表达式中捕获组的数量
    if (matcher.matches() && matcher.groupCount() == 2) {
        // 如果正则匹配成功，提取ClientId并返回
        return matcher.group(1);
    } else {
        // 如果匹配失败，返回null或者抛出异常，视情况而定
        return null;
}

当然你也可以通过解析 payload 来获取更多详细信息，可参照官方文档：客户端上下线事件

主题 (Topic)说明
${clientid}/connected上线事件。当任意客户端上线时，EMQX 就会发布该主题的消息
${clientid}/disconnected下线事件。当任意客户端下线时，EMQX 就会发布该主题的消息

connected 事件消息的 Payload 解析成 JSON 格式如下:

{
    "username": "foo",
    "ts": 1625572213873,
    "sockport": 1883,
    "proto_ver": 4,
    "proto_name": "MQTT",
    "keepalive": 60,
    "ipaddress": "127.0.0.1",
    "expiry_interval": 0,
    "connected_at": 1625572213873,
    "connack": 0,
    "clientid": "emqtt-8348fe27a87976ad4db3",
    "clean_start": true
}

disconnected 事件消息的 Payload 解析成 JSON 格式如下:

{
    "username": "foo",
    "ts": 1625572213873,
    "sockport": 1883,
    "reason": "tcp_closed",
    "proto_ver": 4,
    "proto_name": "MQTT",
    "ipaddress": "127.0.0.1",
    "disconnected_at": 1625572213873,
    "clientid": "emqtt-8348fe27a87976ad4db3"
}

可以解析 JOSN 数据拿到 clientid ，注意此处使用的 JSON 解析工具是 Hutool 的。

// 获取消息内容
String payload = StrUtil.str((byte[]) message.getPayload(), "utf-8");
// 解析JSON字符串
JSONObject payloadJsonObject = JSONUtil.parseObj(payload);
// 获取ClientId
String clientid = payloadJsonObject.getStr("clientid");

实现效果

总结

  • 修改 EMQ X ACL 配置： 你在 EMQ X 中修改了 acl.config 文件，添加了相应的 ACL 规则，允许订阅 $SYS/brokers/+/clients/# 主题的所有客户端。这个步骤允许你在开源版本中访问关键的设备连接信息。
  • 创建 Spring Boot 应用程序： 通过创建一个 Spring Boot 应用程序，你建立了一个连接到 EMQ X 代理的桥梁。这个应用程序充当了监听器，用于订阅 $SYS/brokers/+/clients/# 主题，以实时感知设备的连接和断开事件。
  • 实时设备监控： 你的解决方案允许你实时监控设备的连接状态，而无需依赖 EMQ X 企业版的内置功能。这使你能够快速响应设备状态的变化，并采取相应的行动，如发送通知或执行自定义操作。
  • 定制性： 通过这个方法，你能够灵活地定制解决方案，以满足你的特定需求。你可以根据设备状态采取不同的操作，因此具有更大的灵活性。
  • 避免功能限制： 尽管 EMQ X 4.4.10 开源版本没有内置设备上下线提醒模块，但你成功地绕过了这个限制，通过自定义配置和应用程序开发来实现了所需的功能。
  • 无需升级或许可证： 你的解决方案不需要升级到 EMQ X 企业版或购买额外的许可证。这使得你可以在开源版本的基础上构建所需的功能。
  • 总而言之，你的解决方案是一种巧妙的方式，通过修改配置和创建一个自定义的 Spring Boot 应用程序，实现了设备上下线状态的监控和管理。这个方法不仅解决了技术挑战，还提供了灵活性和定制性，以满足你的特定需求。这是一个创造性的方法，适用于需要在 EMQ X 开源版本中实现设备监控的情况。

以上就是SpringBoot实现EMQ设备的上下线告警的详细内容，更多关于SpringBoot EMQ上下线告警的资料请关注好代码网其它相关文章！

您可能有感兴趣的文章
springboot如何使用EMQX(MQTT协议)的如何实现
SpringBoot整合RabbitMQ示例详解
SpringBoot中如何使用 RabbitMQ的好代码教程详解
SpringBoot整合RabbitMQ 手动应答(简单demo)
Spring Boot系列好代码教程之7步集成RabbitMQ的方法