java

关注公众号 jb51net

关闭
首页 > 软件编程 > java > SpringBoot EMQ上下线告警

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

作者：花伤情犹在

EMQX 的上下线系统消息通知功能在客户端连接成功或者客户端断开连接,需要实现设备的上下线状态监控,所以本文给大家介绍了如何通过SpringBoot实现EMQ设备的上下线告警,文中有详细的代码示例,需要的朋友可以参考下

前言

上下线通知

我遇到了一个难题，即在使用 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 ，提供了可靠的消息传递机制，使设备能够相互通信，同时也可与后端应用程序集成，使其成为物联网生态系统中的重要组成部分。

环境

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/#"]}.

综合起来，这个规则允许所有的客户端订阅以 $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");

实现效果

总结

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

您可能感兴趣的文章:
阅读全文