package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager;
|
|
import cn.hutool.core.util.StrUtil;
|
import io.netty.handler.codec.mqtt.MqttQoS;
|
import io.vertx.core.buffer.Buffer;
|
import io.vertx.mqtt.MqttEndpoint;
|
import lombok.Data;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.stereotype.Component;
|
|
import java.util.ArrayList;
|
import java.util.List;
|
import java.util.Map;
|
import java.util.concurrent.ConcurrentHashMap;
|
|
/**
|
* IoT 网关 MQTT 连接管理器
|
* <p>
|
* 统一管理 MQTT 连接的认证状态、设备会话和消息发送功能：
|
* 1. 管理 MQTT 连接的认证状态
|
* 2. 管理设备会话和在线状态
|
* 3. 管理消息发送到设备
|
*
|
* @author 芋道源码
|
*/
|
@Slf4j
|
@Component
|
public class IotMqttConnectionManager {
|
|
/**
|
* 未知地址常量（当获取端点地址失败时使用）
|
*/
|
private static final String UNKNOWN_ADDRESS = "unknown";
|
|
/**
|
* 连接信息映射：MqttEndpoint -> 连接信息
|
*/
|
private final Map<MqttEndpoint, ConnectionInfo> connectionMap = new ConcurrentHashMap<>();
|
|
/**
|
* 设备 ID -> MqttEndpoint 的映射
|
*/
|
private final Map<Long, MqttEndpoint> deviceEndpointMap = new ConcurrentHashMap<>();
|
|
/**
|
* 安全获取 endpoint 地址
|
* <p>
|
* 优先从缓存获取地址，缓存为空时再尝试实时获取
|
*
|
* @param endpoint MQTT 连接端点
|
* @return 地址字符串，获取失败时返回 "unknown"
|
*/
|
public String getEndpointAddress(MqttEndpoint endpoint) {
|
String realTimeAddress = UNKNOWN_ADDRESS;
|
if (endpoint == null) {
|
return realTimeAddress;
|
}
|
|
// 1. 优先从缓存获取（避免连接关闭时的异常）
|
ConnectionInfo connectionInfo = connectionMap.get(endpoint);
|
if (connectionInfo != null && StrUtil.isNotBlank(connectionInfo.getRemoteAddress())) {
|
return connectionInfo.getRemoteAddress();
|
}
|
|
// 2. 缓存为空时尝试实时获取
|
try {
|
realTimeAddress = endpoint.remoteAddress().toString();
|
} catch (Exception ignored) {
|
// 连接已关闭，忽略异常
|
}
|
return realTimeAddress;
|
}
|
|
/**
|
* 注册设备连接（包含认证信息）
|
*
|
* @param endpoint MQTT 连接端点
|
* @param connectionInfo 连接信息
|
*/
|
public void registerConnection(MqttEndpoint endpoint, ConnectionInfo connectionInfo) {
|
Long deviceId = connectionInfo.getDeviceId();
|
// 如果设备已有其他连接，先清理旧连接
|
MqttEndpoint oldEndpoint = deviceEndpointMap.get(deviceId);
|
if (oldEndpoint != null && oldEndpoint != endpoint) {
|
log.info("[registerConnection][设备已有其他连接，断开旧连接，设备 ID: {}，旧连接: {}]",
|
deviceId, getEndpointAddress(oldEndpoint));
|
// 先清理映射，再关闭连接（避免旧连接处理器干扰）
|
connectionMap.remove(oldEndpoint);
|
oldEndpoint.close();
|
}
|
|
// 注册新连接
|
connectionMap.put(endpoint, connectionInfo);
|
deviceEndpointMap.put(deviceId, endpoint);
|
log.info("[registerConnection][注册设备连接，设备 ID: {}，连接: {}，productKey: {}，deviceName: {}]",
|
deviceId, getEndpointAddress(endpoint), connectionInfo.getProductKey(), connectionInfo.getDeviceName());
|
}
|
|
/**
|
* 注销设备连接
|
*
|
* @param endpoint MQTT 连接端点
|
*/
|
public void unregisterConnection(MqttEndpoint endpoint) {
|
ConnectionInfo connectionInfo = connectionMap.remove(endpoint);
|
if (connectionInfo == null) {
|
return;
|
}
|
Long deviceId = connectionInfo.getDeviceId();
|
deviceEndpointMap.remove(deviceId);
|
log.info("[unregisterConnection][注销设备连接，设备 ID: {}，连接: {}]", deviceId, getEndpointAddress(endpoint));
|
}
|
|
/**
|
* 获取连接信息
|
*/
|
public ConnectionInfo getConnectionInfo(MqttEndpoint endpoint) {
|
return connectionMap.get(endpoint);
|
}
|
|
/**
|
* 根据设备 ID 获取连接信息
|
*
|
* @param deviceId 设备 ID
|
* @return 连接信息
|
*/
|
public ConnectionInfo getConnectionInfoByDeviceId(Long deviceId) {
|
// 通过设备 ID 获取连接端点
|
MqttEndpoint endpoint = getDeviceEndpoint(deviceId);
|
if (endpoint == null) {
|
return null;
|
}
|
// 获取连接信息
|
return getConnectionInfo(endpoint);
|
}
|
|
/**
|
* 发送消息到设备
|
*
|
* @param deviceId 设备 ID
|
* @param topic 主题
|
* @param payload 消息内容
|
* @param qos 服务质量
|
* @param retain 是否保留消息
|
* @return 是否发送成功
|
*/
|
public boolean sendToDevice(Long deviceId, String topic, byte[] payload, int qos, boolean retain) {
|
MqttEndpoint endpoint = deviceEndpointMap.get(deviceId);
|
if (endpoint == null) {
|
log.warn("[sendToDevice][设备离线，无法发送消息，设备 ID: {}，主题: {}]", deviceId, topic);
|
return false;
|
}
|
|
try {
|
endpoint.publish(topic, Buffer.buffer(payload), MqttQoS.valueOf(qos), false, retain);
|
log.debug("[sendToDevice][发送消息成功，设备 ID: {}，主题: {}，QoS: {}]", deviceId, topic, qos);
|
return true;
|
} catch (Exception e) {
|
log.error("[sendToDevice][发送消息失败，设备 ID: {}，主题: {}，错误: {}]", deviceId, topic, e.getMessage());
|
return false;
|
}
|
}
|
|
/**
|
* 获取设备连接端点
|
*/
|
public MqttEndpoint getDeviceEndpoint(Long deviceId) {
|
return deviceEndpointMap.get(deviceId);
|
}
|
|
/**
|
* 关闭所有连接
|
*/
|
public void closeAll() {
|
// 1. 先复制再清空，避免 closeHandler 回调时并发修改
|
List<MqttEndpoint> endpoints = new ArrayList<>(connectionMap.keySet());
|
connectionMap.clear();
|
deviceEndpointMap.clear();
|
// 2. 关闭所有连接（closeHandler 中 unregisterConnection 发现 map 为空会安全跳过）
|
for (MqttEndpoint endpoint : endpoints) {
|
try {
|
endpoint.close();
|
} catch (Exception ignored) {
|
// 连接可能已关闭，忽略异常
|
}
|
}
|
}
|
|
/**
|
* 连接信息
|
*/
|
@Data
|
public static class ConnectionInfo {
|
|
/**
|
* 设备 ID
|
*/
|
private Long deviceId;
|
/**
|
* 产品 Key
|
*/
|
private String productKey;
|
/**
|
* 设备名称
|
*/
|
private String deviceName;
|
|
/**
|
* 连接地址
|
*/
|
private String remoteAddress;
|
|
}
|
|
}
|
