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

# 進(jìn)入EMQX容器
docker exec -it emqx /bin/sh
# 進(jìn)入配置文件目錄
cd /opt/emqx/etc
# 查看acl配置文件
cat acl.conf
# 編輯acl配置文件
vi acl.conf

# 進(jìn)入配置文件目錄
cd /etc/emqx
# 查看acl配置文件
cat acl.conf
# 編輯acl配置文件
vi acl.conf

%%--------------------------------------------------------------------
%% [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}.

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

<!-- MQTT相關(guān)依賴 -->
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-stream</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-mqtt</artifactId>
</dependency>

$SYS/brokers/+/clients/#

// 從消息請(qǐng)求頭中獲取消息主題topic
String topic = String.valueOf(message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC));

// topic最后的節(jié)點(diǎn)字符串
String lastPart = extractLastPart(topic);
// 獲取消息內(nèi)容
String payload = StrUtil.str((byte[]) message.getPayload(), "utf-8");
// 判斷設(shè)備是上線或下線消息
if ("connected".equals(lastPart)) {
    // 設(shè)備上線消息
    clientId = extractClientIdFromTopic(topic);
    log.info("設(shè)備上線提醒 -> IMEI:{}", clientId);
} else if ("disconnected".equals(lastPart)) {
    // 設(shè)備下線消息
    clientId = extractClientIdFromTopic(topic);
    log.info("設(shè)備下線警告 -> IMEI:{}", clientId);
}
/**
 * 獲取最后一個(gè)節(jié)點(diǎn)
 *
 * @param topic 主題
 * @return 節(jié)點(diǎn)內(nèi)容
 */
public static String extractLastPart(String topic) {
    // 使用split方法將字符串根據(jù)'/'分割成數(shù)組
    String[] parts = topic.split("/");
    // 獲取最后一個(gè)元素
    String lastPart = parts[parts.length - 1];
    return lastPart;
}
/**
 * 從Topic中提取ClientId
 *
 * @param topic 主題
 * @return ClientId
 */
public static String extractClientIdFromTopic(String topic) {
    // 使用正則表達(dá)式匹配主題中的ClientId
    String regex = "\\$SYS/brokers/[^/]+/clients/([^/]+)/(connected|disconnected)";
    Pattern pattern = Pattern.compile(regex);
    Matcher matcher = pattern.matcher(topic);
    // matcher.groupCount() 是一個(gè)方法，用于獲取正則表達(dá)式中定義的組數(shù)。在正則表達(dá)式中，使用括號(hào) () 來(lái)定義捕獲組。在這個(gè)情況下，正則表達(dá)式 \\$SYS/brokers/[^/]+/clients/([^/]+)/(connected|disconnected) 中有兩組，分別是括號(hào)內(nèi)的 ([^/]+) 部分和 (connected|disconnected) 部分。matcher.groupCount() 返回的是正則表達(dá)式中捕獲組的數(shù)量
    if (matcher.matches() && matcher.groupCount() == 2) {
        // 如果正則匹配成功，提取ClientId并返回
        return matcher.group(1);
    } else {
        // 如果匹配失敗，返回null或者拋出異常，視情況而定
        return null;
}

{
    "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
}

{
    "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"
}

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