Skip to content

黄嘉阳 / iot-platform-server

Go to a project
Commit 15decda0 by huangjy
Options

fix: 呼吸心率设备数据上报缓存问题

parent 79600242
Inline Side-by-side
Showing with 33 additions and 30 deletions
server-service/src/main/java/com/makeit/service/platform/elder/impl/PlatElderRealTimeServiceImpl.java
...@@ -194,11 +194,6 @@ public class PlatElderRealTimeServiceImpl implements PlatElderRealTimeService { ...@@ -194,11 +194,6 @@ public class PlatElderRealTimeServiceImpl implements PlatElderRealTimeService {
Map<Object, Object> entries = redisTemplate.opsForHash().entries(DEVICE_BR_ANALYSIS + platDevice.getOriDeviceId()); Map<Object, Object> entries = redisTemplate.opsForHash().entries(DEVICE_BR_ANALYSIS + platDevice.getOriDeviceId());
Collection<Object> values = entries.values(); Collection<Object> values = entries.values();
if (values.size() >= sleepTimeActionDuration) { if (values.size() >= sleepTimeActionDuration) {
Predicate<Object> predicate = entity -> {
JSONObject result = JSON.parseObject(entity.toString());
Integer bodymove = Integer.valueOf(result.getString("bodymove"));
return bodymove.compareTo(sleepTimeActionThreshold) <= 0;
};
int count = 0; int count = 0;
for (Object entity : values) { for (Object entity : values) {
JSONObject result = JSON.parseObject(entity.toString()); JSONObject result = JSON.parseObject(entity.toString());
......
server-web/src/main/java/com/makeit/mqtt/PushCallback.java
...@@ -153,42 +153,21 @@ public class PushCallback implements MqttCallback { ...@@ -153,42 +153,21 @@ public class PushCallback implements MqttCallback {
Long timestamp = device.getTimestamp(); Long timestamp = device.getTimestamp();
long currentSecond = timestamp / 1000; long currentSecond = timestamp / 1000;
// TODO 先这样判断 // 先通过产品名称判断
if (REPORT_PROPERTY.equals(device.getMessageType()) && headers.getProductName().contains("呼吸")) { if (REPORT_PROPERTY.equals(device.getMessageType()) && headers.getProductName().contains("呼吸")) {
// 缓存呼吸设备某段时间的数据，hash 最大长度 duration cacheBrDeviceData(device, currentSecond);
long duration = getSleepTimeActionDuration();
long size = redisTemplate.opsForHash().size(DEVICE_BR_ANALYSIS + device.getDeviceId());
if (size == duration) {
redisTemplate.opsForHash().delete(DEVICE_BR_ANALYSIS + device.getDeviceId(),currentSecond - duration);
}else if (size - duration >= 5) {
redisTemplate.delete(Objects.requireNonNull(redisTemplate.keys(DEVICE_BR_ANALYSIS + device.getDeviceId())));
}
redisTemplate.opsForHash().put(DEVICE_BR_ANALYSIS + device.getDeviceId(), currentSecond,JSON.toJSONString(device.getProperties()));
redisTemplate.expire(DEVICE_BR_ANALYSIS + device.getDeviceId(),duration + 3,TimeUnit.MINUTES);
redisTemplate.opsForValue().set(DEVICE_BR_DATA + device.getDeviceId(),JSON.toJSONString(device.getProperties()), redisTemplate.opsForValue().set(DEVICE_BR_DATA + device.getDeviceId(),JSON.toJSONString(device.getProperties()),
5000, TimeUnit.MILLISECONDS); 5000, TimeUnit.MILLISECONDS);
} }
if (REPORT_PROPERTY.equals(device.getMessageType()) && headers.getProductName().contains("空间")) { if (REPORT_PROPERTY.equals(device.getMessageType()) && headers.getProductName().contains("空间")) {
long maxSize = 10L; cacheSpaceFallDeviceData(DEVICE_SPACE_TEMP_DATA, device, currentSecond);
Long size = redisTemplate.opsForHash().size(DEVICE_SPACE_TEMP_DATA + device.getDeviceId());
redisTemplate.opsForHash().put(DEVICE_SPACE_TEMP_DATA + device.getDeviceId(), currentSecond,JSON.toJSONString(device.getProperties()));
redisTemplate.expire(DEVICE_SPACE_TEMP_DATA + device.getDeviceId(),maxSize + 1,TimeUnit.SECONDS);
if (size == maxSize) {
redisTemplate.opsForHash().delete(DEVICE_SPACE_TEMP_DATA + device.getDeviceId(),currentSecond - maxSize);
}
redisTemplate.opsForValue().set(DEVICE_SPACE_DATA + device.getDeviceId(),JSON.toJSONString(device.getProperties()), redisTemplate.opsForValue().set(DEVICE_SPACE_DATA + device.getDeviceId(),JSON.toJSONString(device.getProperties()),
5000, TimeUnit.MILLISECONDS); 5000, TimeUnit.MILLISECONDS);
} }
if (REPORT_PROPERTY.equals(device.getMessageType()) && headers.getProductName().contains("跌倒")) { if (REPORT_PROPERTY.equals(device.getMessageType()) && headers.getProductName().contains("跌倒")) {
long maxSize = 10L; cacheSpaceFallDeviceData(DEVICE_FALL_TEMP_DATA, device, currentSecond);
Long size = redisTemplate.opsForHash().size(DEVICE_FALL_TEMP_DATA + device.getDeviceId());
redisTemplate.opsForHash().put(DEVICE_FALL_TEMP_DATA + device.getDeviceId(), currentSecond,JSON.toJSONString(device.getProperties()));
redisTemplate.expire(DEVICE_FALL_TEMP_DATA + device.getDeviceId(),maxSize + 1,TimeUnit.SECONDS);
if (size == maxSize) {
redisTemplate.opsForHash().delete(DEVICE_FALL_TEMP_DATA + device.getDeviceId(),currentSecond - maxSize);
}
redisTemplate.opsForValue().set(DEVICE_FALL_DATA + device.getDeviceId(),JSON.toJSONString(device.getProperties()), redisTemplate.opsForValue().set(DEVICE_FALL_DATA + device.getDeviceId(),JSON.toJSONString(device.getProperties()),
5000, TimeUnit.MILLISECONDS); 5000, TimeUnit.MILLISECONDS);
...@@ -251,6 +230,35 @@ public class PushCallback implements MqttCallback { ...@@ -251,6 +230,35 @@ public class PushCallback implements MqttCallback {
} }
private void cacheSpaceFallDeviceData(String deviceSpaceTempData, DeviceInfo device, long currentSecond) {
long maxSize = 10L;
String key = deviceSpaceTempData + device.getDeviceId();
Long size = redisTemplate.opsForHash().size(key);
redisTemplate.opsForHash().put(key, currentSecond,JSON.toJSONString(device.getProperties()));
redisTemplate.expire(key,maxSize + 1,TimeUnit.SECONDS);
if (size == maxSize) {
redisTemplate.opsForHash().delete(key, currentSecond - maxSize);
}
}
private void cacheBrDeviceData(DeviceInfo device, long currentSecond) {
// 缓存呼吸设备某段时间的数据，hash 最大长度 duration
String key = DEVICE_BR_ANALYSIS + device.getDeviceId();
long duration = getSleepTimeActionDuration();
long size = redisTemplate.opsForHash().size(key);
long differenceValue = currentSecond - duration;
if (size == duration) {
redisTemplate.opsForHash().delete(key, differenceValue);
}else if (size - duration >= 5) {
// 兼容设备上报数据时间不准确问题
for (long i = differenceValue; i < differenceValue - 10; i--) {
redisTemplate.opsForHash().delete(key, i);
}
}
redisTemplate.opsForHash().put(key, currentSecond,JSON.toJSONString(device.getProperties()));
redisTemplate.expire(key,duration + 3,TimeUnit.MINUTES);
}
private Long getSleepTimeActionDuration() { private Long getSleepTimeActionDuration() {
String sleepTimeActionDuration = ""; String sleepTimeActionDuration = "";
String result = redisTemplate.opsForValue().get(SLEEP_ANALYSIS_KEY); String result = redisTemplate.opsForValue().get(SLEEP_ANALYSIS_KEY);
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment