diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/push/ThingsModelPushHandler.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/push/ThingsModelPushHandler.java index dc5eb64c..1d5e37d9 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/push/ThingsModelPushHandler.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/push/ThingsModelPushHandler.java @@ -6,12 +6,12 @@ import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.common.core.notify.NotifySendResponse; import com.fastbee.common.core.notify.config.WeChatConfigParams; import com.fastbee.common.core.notify.msg.WechatMsgParams; -import com.fastbee.common.core.redis.RedisCache; import com.fastbee.common.core.redis.RedisKeyBuilder; import com.fastbee.common.enums.DataLevelEnum; import com.fastbee.common.utils.StringUtils; import com.fastbee.common.utils.wechat.WechatUtils; import com.fastbee.common.wechat.WeChatAppResult; +import com.fastbee.iot.cache.ITSLCache; import com.fastbee.iot.domain.PushModelItem; import com.fastbee.iot.domain.SocialUser; import com.fastbee.iot.domain.ThingsModelPushConfig; @@ -64,7 +64,7 @@ public class ThingsModelPushHandler { @Resource private ISocialUserService socialUserService; @Resource - private RedisCache redisCache; + private ITSLCache itslCache; /** * 按推送配置执行推送 @@ -129,16 +129,16 @@ public class ThingsModelPushHandler { } // ---------- 执行推送 ---------- - pushForDevice(config, msgParams, pushUrl, openIdMap); - - // 更新最近推送时间 - pushConfigService.updateLastPushTime(config.getId()); + if (pushForDevice(config, msgParams, pushUrl, openIdMap)) { + // 更新最近推送时间 + pushConfigService.updateLastPushTime(config.getId()); + } } /** * 对单台设备按等级分组推送 */ - private void pushForDevice(ThingsModelPushConfig config, WechatMsgParams msgParams, + private boolean pushForDevice(ThingsModelPushConfig config, WechatMsgParams msgParams, String pushUrl, Map openIdMap) { String serialNumber = config.getSerialNumber().toUpperCase(); String redisKey = RedisKeyBuilder.buildTSLVCacheKey(config.getProductId(), serialNumber); @@ -147,17 +147,18 @@ public class ThingsModelPushHandler { for (DataLevelEnum lvl : DataLevelEnum.values()) { levelTextMap.put(lvl, new ArrayList<>()); } - + Map cacheMap = itslCache.getSimpleThingsModelValueItemByCache(redisKey); for (PushModelItem item : config.getItems()) { - String cacheValue = redisCache.getCacheMapValue(redisKey, item.getIdentifier()); - if (StringUtils.isEmpty(cacheValue)) { + ValueItem valueItem = cacheMap.get(item.getIdentifier()); + if (valueItem == null) { continue; } - String value; - try { - value = JSON.parseObject(cacheValue, ValueItem.class).getValue(); - } catch (Exception e) { - value = cacheValue; + String value = valueItem.getValue(); + Date ts = valueItem.getTs(); + if (config.getLastPushTime() != null && ts != null && ts.before(config.getLastPushTime())) { + log.warn("[Push] configId={} sn={} item={} value={} 时间{}早于上次推送时间{},跳过", + config.getId(), serialNumber, item.getIdentifier(), value, ts, config.getLastPushTime()); + return false; } // 没有值 直接跳过 @@ -184,13 +185,14 @@ public class ThingsModelPushHandler { if (chosen == null) { log.debug("[Push] configId={} sn={} 无有效分级数据,跳过发送", config.getId(), serialNumber); - return; + return false; } String dataText = String.join(";", levelTextMap.get(chosen)); if (dataText.length() > 32) { dataText = dataText.substring(0, 29) + "..."; } - sendTemplateMsg(config, msgParams, serialNumber, chosen, dataText, pushUrl, openIdMap); + + return sendTemplateMsg(config, msgParams, serialNumber, chosen, dataText, pushUrl, openIdMap); } /** @@ -198,7 +200,7 @@ public class ThingsModelPushHandler { * 从通知模板的 content 字段提取 {{变量.DATA}} 变量名顺序, * 按顺序映射:第1个→设备名称,第2个→数据等级,第3个→监测数据。 */ - private void sendTemplateMsg(ThingsModelPushConfig config, WechatMsgParams msgParams, + private boolean sendTemplateMsg(ThingsModelPushConfig config, WechatMsgParams msgParams, String serialNumber, DataLevelEnum level, String dataText, String pushUrl, Map openIdMap) { List varNames = extractVarNames(msgParams.getContent()); @@ -227,6 +229,7 @@ public class ThingsModelPushHandler { sendMap.put(m.getKey(), new TemplateDataVo(m.getValue())); } + NotifySendResponse resp = null; for (Map.Entry userEntry : openIdMap.entrySet()) { String openId = userEntry.getValue(); WeChatPublicAccountPushVO vo = new WeChatPublicAccountPushVO(); @@ -237,15 +240,13 @@ public class ThingsModelPushHandler { vo.setUrl(msgParams.getRedirectUrl() + "pagesA/home/device/index?deviceId=" + config.getDeviceId()); } try { - NotifySendResponse resp = doPost(JSON.toJSONString(vo), pushUrl); - log.info("[Push] configId={} sn={} userId={} level={} status={}", - config.getId(), serialNumber, userEntry.getKey(), - level.getLabel(), resp.getStatus()); + resp = doPost(JSON.toJSONString(vo), pushUrl); } catch (Exception e) { log.error("[Push] configId={} sn={} userId={} 推送异常: {}", config.getId(), serialNumber, userEntry.getKey(), e.getMessage()); } } + return Objects.nonNull(resp) && resp.getStatus() == 1; } private List extractVarNames(String content) { diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/cache/ITSLCache.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/cache/ITSLCache.java index ff959769..e2aef953 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/cache/ITSLCache.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/cache/ITSLCache.java @@ -1,6 +1,7 @@ package com.fastbee.iot.cache; import com.fastbee.iot.model.ThingsModels.ThingsModelValueItem; +import com.fastbee.iot.model.ThingsModels.ValueItem; import java.util.List; import java.util.Map; @@ -135,4 +136,11 @@ public interface ITSLCache { */ ThingsModelValueItem getSingleDeviceCalculatedModel(String serialNumber, String identifier); + /** + * 根据redisKey获取简易物模型缓存值 + * + * @param redisKey + * @return + */ + Map getSimpleThingsModelValueItemByCache(String redisKey); } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/cache/impl/TSLCacheImpl.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/cache/impl/TSLCacheImpl.java index 877f416b..c1d8a9e3 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/cache/impl/TSLCacheImpl.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/cache/impl/TSLCacheImpl.java @@ -16,12 +16,14 @@ import com.fastbee.iot.model.ThingsModelItem.Datatype; import com.fastbee.iot.model.ThingsModels.IdentifierVO; import com.fastbee.iot.model.ThingsModels.PropertyDto; import com.fastbee.iot.model.ThingsModels.ThingsModelValueItem; +import com.fastbee.iot.model.ThingsModels.ValueItem; import com.fastbee.iot.service.IModbusConfigService; import com.fastbee.iot.service.IThingsModelService; import org.springframework.beans.BeanUtils; import org.springframework.stereotype.Service; import org.springframework.util.CollectionUtils; +import javax.annotation.Nullable; import javax.annotation.Resource; import java.util.*; import java.util.function.Function; @@ -183,14 +185,8 @@ public class TSLCacheImpl implements ITSLCache { assert !Objects.isNull(productId) : "产品id为空"; // redis获取物模型 Map map = redisCache.getCacheMap(RedisKeyBuilder.buildTSLCacheKey(productId)); - if (!CollectionUtils.isEmpty(map)) { - Map itemMap = map.values().stream().map(v -> JSON.parseObject(v.toString(), ThingsModelValueItem.class)).collect( - Collectors.toMap(ThingsModelValueItem::getId, Function.identity())); - for (ThingsModelValueItem item : itemMap.values()) { - convertModelName(item, SecurityUtils.getLanguage()); - } - return itemMap; - } + Map itemMap = buildThingsModelValueItemMap(map); + if (itemMap != null) return itemMap; return setCacheThMapByProductId(productId); } @@ -345,15 +341,8 @@ public class TSLCacheImpl implements ITSLCache { } String cacheKey = RedisKeyBuilder.buildDeviceTSLCacheKey(serialNumber); Map map = redisCache.getCacheMap(cacheKey); - if (!CollectionUtils.isEmpty(map)) { - Map itemMap = map.values().stream() - .map(v -> JSON.parseObject(v.toString(), ThingsModelValueItem.class)) - .collect(Collectors.toMap(ThingsModelValueItem::getId, Function.identity())); - for (ThingsModelValueItem item : itemMap.values()) { - convertModelName(item, SecurityUtils.getLanguage()); - } - return itemMap; - } + Map itemMap = buildThingsModelValueItemMap(map); + if (itemMap != null) return itemMap; // 缓存未命中,从数据库加载 return refreshDeviceCalculatedModelCache(serialNumber); } @@ -449,4 +438,34 @@ public class TSLCacheImpl implements ITSLCache { return item; } + @Override + public Map getSimpleThingsModelValueItemByCache(String redisKey) { + Map map = redisCache.getCacheMap(redisKey); + return buildValueItemMap(map); + } + + @Nullable + private Map buildThingsModelValueItemMap(Map map) { + if (!CollectionUtils.isEmpty(map)) { + Map itemMap = map.values().stream() + .map(v -> JSON.parseObject(v.toString(), ThingsModelValueItem.class)) + .collect(Collectors.toMap(ThingsModelValueItem::getId, Function.identity())); + for (ThingsModelValueItem item : itemMap.values()) { + convertModelName(item, SecurityUtils.getLanguage()); + } + return itemMap; + } + return null; + } + + + @Nullable + private Map buildValueItemMap(Map map) { + if (!CollectionUtils.isEmpty(map)) { + return map.values().stream() + .map(v -> JSON.parseObject(v.toString(), ValueItem.class)) + .collect(Collectors.toMap(ValueItem::getId, Function.identity())); + } + return null; + } }