Sumuxi/fastbee
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

20260328:数据推送：最推送最新数据

Browse Source
This commit is contained in:
George 2026-03-28 20:44:42 +08:00
parent c25f288b6c
commit 8a16a35627
3 changed files with 67 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
fastbee-iot-data/src/main/java/com/fastbee/iot/data/push/ThingsModelPushHandler.java
View File

@ -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<Long, String> 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<String, ValueItem> 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<Long, String> openIdMap) {
List<String> 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<Long, String> 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<String> extractVarNames(String content) {

8
fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/cache/ITSLCache.java vendored
View File

@ -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<String, ValueItem> getSimpleThingsModelValueItemByCache(String redisKey);
}

53
fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/cache/impl/TSLCacheImpl.java vendored
View File

@ -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<String, Object> map = redisCache.getCacheMap(RedisKeyBuilder.buildTSLCacheKey(productId));
if (!CollectionUtils.isEmpty(map)) {
Map<String, ThingsModelValueItem> 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<String, ThingsModelValueItem> 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<String, Object> map = redisCache.getCacheMap(cacheKey);
if (!CollectionUtils.isEmpty(map)) {
Map<String, ThingsModelValueItem> 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<String, ThingsModelValueItem> 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<String, ValueItem> getSimpleThingsModelValueItemByCache(String redisKey) {
Map<String, Object> map = redisCache.getCacheMap(redisKey);
return buildValueItemMap(map);
}
@Nullable
private Map<String, ThingsModelValueItem> buildThingsModelValueItemMap(Map<String, Object> map) {
if (!CollectionUtils.isEmpty(map)) {
Map<String, ThingsModelValueItem> 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<String, ValueItem> buildValueItemMap(Map<String, Object> 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;
}
}