fix:优化物联设备mqtt订阅
HuangHuiKang
4 days ago
parent
0e5d780de8
commit
566acc12eb
@ -0,0 +1,30 @@
|
||||
package cn.iocoder.yudao.module.iot.framework.mqtt.config;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.eclipse.paho.client.mqttv3.MqttClient;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import javax.annotation.PreDestroy;
|
||||
import javax.annotation.Resource;
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class MqttClientShutdown {
|
||||
|
||||
@Resource
|
||||
private MqttClient mqttClient;
|
||||
|
||||
@PreDestroy
|
||||
public void shutdown() {
|
||||
try {
|
||||
if (mqttClient.isConnected()) {
|
||||
mqttClient.disconnectForcibly(1000, 1000);
|
||||
}
|
||||
mqttClient.close();
|
||||
log.info("MQTT客户端已关闭");
|
||||
} catch (Exception e) {
|
||||
log.warn("MQTT客户端关闭失败: {}", e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,164 @@
|
||||
package cn.iocoder.yudao.module.iot.framework.mqtt.consumer;
|
||||
|
||||
import cn.iocoder.yudao.module.iot.controller.admin.devicemodelrules.vo.PointRulesRespVO;
|
||||
import cn.iocoder.yudao.module.iot.dal.dataobject.device.DeviceDO;
|
||||
import cn.iocoder.yudao.module.iot.dal.dataobject.devicecontactmodel.DeviceContactModelDO;
|
||||
import cn.iocoder.yudao.module.iot.dal.dataobject.devicepointrules.DevicePointRulesDO;
|
||||
import cn.iocoder.yudao.module.iot.dal.mysql.device.DeviceMapper;
|
||||
import cn.iocoder.yudao.module.iot.dal.mysql.devicecontactmodel.DeviceContactModelMapper;
|
||||
import cn.iocoder.yudao.module.iot.dal.mysql.devicepointrules.DevicePointRulesMapper;
|
||||
import com.alibaba.fastjson.JSON;
|
||||
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
|
||||
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Getter;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
import org.springframework.context.annotation.Lazy;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import javax.annotation.Resource;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Optional;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class IotMqttRuntimeCache {
|
||||
|
||||
@Resource
|
||||
@Lazy
|
||||
private DeviceMapper deviceMapper;
|
||||
|
||||
@Resource
|
||||
private DeviceContactModelMapper deviceContactModelMapper;
|
||||
|
||||
@Resource
|
||||
private DevicePointRulesMapper devicePointRulesMapper;
|
||||
|
||||
private final ConcurrentMap<String, Optional<DeviceDO>> deviceByTopic = new ConcurrentHashMap<>();
|
||||
private final ConcurrentMap<Long, List<DeviceContactModelDO>> pointsByDeviceId = new ConcurrentHashMap<>();
|
||||
private final ConcurrentMap<Long, Map<String, List<CachedPointRule>>> pointRulesByDeviceId = new ConcurrentHashMap<>();
|
||||
private final ConcurrentMap<Long, Optional<DevicePointRulesDO>> countRuleByDeviceId = new ConcurrentHashMap<>();
|
||||
|
||||
public DeviceDO getDeviceByTopic(String topic) {
|
||||
if (StringUtils.isBlank(topic)) {
|
||||
return null;
|
||||
}
|
||||
return deviceByTopic.computeIfAbsent(topic, this::loadDeviceByTopic).orElse(null);
|
||||
}
|
||||
|
||||
public List<DeviceContactModelDO> getDevicePoints(Long deviceId) {
|
||||
if (deviceId == null) {
|
||||
return Collections.emptyList();
|
||||
}
|
||||
return pointsByDeviceId.computeIfAbsent(deviceId, this::loadDevicePoints);
|
||||
}
|
||||
|
||||
public List<CachedPointRule> getPointRules(Long deviceId, String attributeCode) {
|
||||
if (deviceId == null || StringUtils.isBlank(attributeCode)) {
|
||||
return Collections.emptyList();
|
||||
}
|
||||
return pointRulesByDeviceId.computeIfAbsent(deviceId, this::loadPointRulesByAttributeCode)
|
||||
.getOrDefault(attributeCode, Collections.emptyList());
|
||||
}
|
||||
|
||||
public DevicePointRulesDO getLatestCountRule(Long deviceId) {
|
||||
if (deviceId == null) {
|
||||
return null;
|
||||
}
|
||||
return countRuleByDeviceId.computeIfAbsent(deviceId, this::loadLatestCountRule).orElse(null);
|
||||
}
|
||||
|
||||
public void refreshDevice(Long deviceId) {
|
||||
if (deviceId == null) {
|
||||
return;
|
||||
}
|
||||
pointsByDeviceId.remove(deviceId);
|
||||
pointRulesByDeviceId.remove(deviceId);
|
||||
countRuleByDeviceId.remove(deviceId);
|
||||
deviceByTopic.entrySet().removeIf(entry -> entry.getValue().map(device -> deviceId.equals(device.getId())).orElse(false));
|
||||
}
|
||||
|
||||
public void refreshTopic(String topic) {
|
||||
if (StringUtils.isNotBlank(topic)) {
|
||||
deviceByTopic.remove(topic);
|
||||
}
|
||||
}
|
||||
|
||||
public void clearAll() {
|
||||
deviceByTopic.clear();
|
||||
pointsByDeviceId.clear();
|
||||
pointRulesByDeviceId.clear();
|
||||
countRuleByDeviceId.clear();
|
||||
}
|
||||
|
||||
private Optional<DeviceDO> loadDeviceByTopic(String topic) {
|
||||
return Optional.ofNullable(deviceMapper.selectOne(Wrappers.<DeviceDO>lambdaQuery()
|
||||
.eq(DeviceDO::getTopic, topic)
|
||||
.last("LIMIT 1")));
|
||||
}
|
||||
|
||||
private List<DeviceContactModelDO> loadDevicePoints(Long deviceId) {
|
||||
List<DeviceContactModelDO> points = deviceContactModelMapper.selectList(Wrappers.<DeviceContactModelDO>lambdaQuery()
|
||||
.eq(DeviceContactModelDO::getDeviceId, deviceId));
|
||||
if (CollectionUtils.isEmpty(points)) {
|
||||
return Collections.emptyList();
|
||||
}
|
||||
return Collections.unmodifiableList(new ArrayList<>(points));
|
||||
}
|
||||
|
||||
private Map<String, List<CachedPointRule>> loadPointRulesByAttributeCode(Long deviceId) {
|
||||
List<DevicePointRulesDO> rules = devicePointRulesMapper.selectList(Wrappers.<DevicePointRulesDO>lambdaQuery()
|
||||
.eq(DevicePointRulesDO::getDeviceId, deviceId)
|
||||
.orderByDesc(DevicePointRulesDO::getCreateTime));
|
||||
if (CollectionUtils.isEmpty(rules)) {
|
||||
return Collections.emptyMap();
|
||||
}
|
||||
List<CachedPointRule> parsedRules = new ArrayList<>();
|
||||
for (DevicePointRulesDO rule : rules) {
|
||||
if (StringUtils.isBlank(rule.getFieldRule())) {
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
List<PointRulesRespVO> pointRules = JSON.parseArray(rule.getFieldRule(), PointRulesRespVO.class);
|
||||
if (CollectionUtils.isEmpty(pointRules)) {
|
||||
continue;
|
||||
}
|
||||
for (PointRulesRespVO pointRule : pointRules) {
|
||||
if (StringUtils.isBlank(pointRule.getCode())) {
|
||||
continue;
|
||||
}
|
||||
parsedRules.add(new CachedPointRule(rule, pointRule));
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.warn("MQTT规则解析失败 deviceId={}, ruleId={}, reason={}", deviceId, rule.getId(), e.getMessage());
|
||||
}
|
||||
}
|
||||
if (parsedRules.isEmpty()) {
|
||||
return Collections.emptyMap();
|
||||
}
|
||||
return parsedRules.stream().collect(Collectors.groupingBy(rule -> rule.getPointRule().getCode()));
|
||||
}
|
||||
|
||||
private Optional<DevicePointRulesDO> loadLatestCountRule(Long deviceId) {
|
||||
return Optional.ofNullable(devicePointRulesMapper.selectOne(Wrappers.<DevicePointRulesDO>lambdaQuery()
|
||||
.eq(DevicePointRulesDO::getDeviceId, deviceId)
|
||||
.eq(DevicePointRulesDO::getIdentifier, "COUNT")
|
||||
.orderByDesc(DevicePointRulesDO::getUpdateTime)
|
||||
.last("LIMIT 1")));
|
||||
}
|
||||
|
||||
@Getter
|
||||
@AllArgsConstructor
|
||||
public static class CachedPointRule {
|
||||
private final DevicePointRulesDO rule;
|
||||
private final PointRulesRespVO pointRule;
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Reference in New Issue