mk1990
2 years ago
committed by
GitHub
GitHub
34 changed files with 368 additions and 331 deletions
@ -0,0 +1,91 @@ |
|||
package com.genersoft.iot.vmp.gb28181.session; |
|||
|
|||
import com.genersoft.iot.vmp.gb28181.bean.*; |
|||
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder; |
|||
import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage; |
|||
import com.genersoft.iot.vmp.vmanager.bean.WVPResult; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.scheduling.annotation.Scheduled; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
import java.time.Instant; |
|||
import java.util.*; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
/** |
|||
* @author lin |
|||
*/ |
|||
@Component |
|||
public class RecordDataCatch { |
|||
|
|||
public static Map<String, RecordInfo> data = new ConcurrentHashMap<>(); |
|||
|
|||
@Autowired |
|||
private DeferredResultHolder deferredResultHolder; |
|||
|
|||
|
|||
public int put(String deviceId, String sn, int sumNum, List<RecordItem> recordItems) { |
|||
String key = deviceId + sn; |
|||
RecordInfo recordInfo = data.get(key); |
|||
if (recordInfo == null) { |
|||
recordInfo = new RecordInfo(); |
|||
recordInfo.setDeviceId(deviceId); |
|||
recordInfo.setSn(sn.trim()); |
|||
recordInfo.setSumNum(sumNum); |
|||
recordInfo.setRecordList(Collections.synchronizedList(new ArrayList<>())); |
|||
recordInfo.setLastTime(Instant.now()); |
|||
recordInfo.getRecordList().addAll(recordItems); |
|||
data.put(key, recordInfo); |
|||
}else { |
|||
// 同一个设备的通道同步请求只考虑一个，其他的直接忽略
|
|||
if (!Objects.equals(sn.trim(), recordInfo.getSn())) { |
|||
return 0; |
|||
} |
|||
recordInfo.getRecordList().addAll(recordItems); |
|||
recordInfo.setLastTime(Instant.now()); |
|||
} |
|||
return recordInfo.getRecordList().size(); |
|||
} |
|||
|
|||
@Scheduled(fixedRate = 5 * 1000) //每5秒执行一次, 发现数据5秒未更新则移除数据并认为数据接收超时
|
|||
private void timerTask(){ |
|||
Set<String> keys = data.keySet(); |
|||
// 获取五秒前的时刻
|
|||
Instant instantBefore5S = Instant.now().minusMillis(TimeUnit.SECONDS.toMillis(5)); |
|||
for (String key : keys) { |
|||
RecordInfo recordInfo = data.get(key); |
|||
// 超过五秒收不到消息任务超时， 只更新这一部分数据
|
|||
if ( recordInfo.getLastTime().isBefore(instantBefore5S)) { |
|||
// 处理录像数据， 返回给前端
|
|||
String msgKey = DeferredResultHolder.CALLBACK_CMD_RECORDINFO + recordInfo.getDeviceId() + recordInfo.getSn(); |
|||
|
|||
WVPResult<RecordInfo> wvpResult = new WVPResult<>(); |
|||
wvpResult.setCode(0); |
|||
wvpResult.setMsg("success"); |
|||
// 对数据进行排序
|
|||
Collections.sort(recordInfo.getRecordList()); |
|||
wvpResult.setData(recordInfo); |
|||
|
|||
RequestMessage msg = new RequestMessage(); |
|||
msg.setKey(msgKey); |
|||
msg.setData(wvpResult); |
|||
deferredResultHolder.invokeAllResult(msg); |
|||
data.remove(key); |
|||
} |
|||
} |
|||
} |
|||
|
|||
public boolean isComplete(String deviceId, String sn) { |
|||
RecordInfo recordInfo = data.get(deviceId + sn); |
|||
return recordInfo != null && recordInfo.getRecordList().size() == recordInfo.getSumNum(); |
|||
} |
|||
|
|||
public RecordInfo getRecordInfo(String deviceId, String sn) { |
|||
return data.get(deviceId + sn); |
|||
} |
|||
|
|||
public void remove(String deviceId, String sn) { |
|||
data.remove(deviceId + sn); |
|||
} |
|||
} |
|||
@ -1,76 +0,0 @@ |
|||
package com.genersoft.iot.vmp.gb28181.transmit.callback; |
|||
|
|||
import com.genersoft.iot.vmp.gb28181.bean.RecordInfo; |
|||
import com.genersoft.iot.vmp.gb28181.bean.RecordItem; |
|||
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd.RecordInfoResponseMessageHandler; |
|||
import com.genersoft.iot.vmp.utils.redis.RedisUtil; |
|||
import org.slf4j.Logger; |
|||
|
|||
import java.util.ArrayList; |
|||
import java.util.Comparator; |
|||
import java.util.List; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
@SuppressWarnings("unchecked") |
|||
public class CheckForAllRecordsThread extends Thread { |
|||
|
|||
private String key; |
|||
|
|||
private RecordInfo recordInfo; |
|||
|
|||
private RedisUtil redis; |
|||
|
|||
private Logger logger; |
|||
|
|||
private DeferredResultHolder deferredResultHolder; |
|||
|
|||
public CheckForAllRecordsThread(String key, RecordInfo recordInfo) { |
|||
this.key = key; |
|||
this.recordInfo = recordInfo; |
|||
} |
|||
|
|||
@Override |
|||
public void run() { |
|||
|
|||
String cacheKey = this.key; |
|||
|
|||
for (long stop = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); stop > System.nanoTime();) { |
|||
List<Object> cacheKeys = redis.scan(cacheKey + "_*"); |
|||
List<RecordItem> totalRecordList = new ArrayList<RecordItem>(); |
|||
for (int i = 0; i < cacheKeys.size(); i++) { |
|||
totalRecordList.addAll((List<RecordItem>) redis.get(cacheKeys.get(i).toString())); |
|||
} |
|||
if (totalRecordList.size() < this.recordInfo.getSumNum()) { |
|||
logger.info("已获取" + totalRecordList.size() + "项录像数据，共" + this.recordInfo.getSumNum() + "项"); |
|||
} else { |
|||
logger.info("录像数据已全部获取，共 {} 项", this.recordInfo.getSumNum()); |
|||
this.recordInfo.setRecordList(totalRecordList); |
|||
for (int i = 0; i < cacheKeys.size(); i++) { |
|||
redis.del(cacheKeys.get(i).toString()); |
|||
} |
|||
break; |
|||
} |
|||
} |
|||
// 自然顺序排序, 元素进行升序排列
|
|||
this.recordInfo.getRecordList().sort(Comparator.naturalOrder()); |
|||
RequestMessage msg = new RequestMessage(); |
|||
msg.setKey(DeferredResultHolder.CALLBACK_CMD_RECORDINFO + recordInfo.getDeviceId() + recordInfo.getSn()); |
|||
msg.setData(recordInfo); |
|||
deferredResultHolder.invokeAllResult(msg); |
|||
logger.info("处理完成，返回结果"); |
|||
RecordInfoResponseMessageHandler.threadNameList.remove(cacheKey); |
|||
} |
|||
|
|||
public void setRedis(RedisUtil redis) { |
|||
this.redis = redis; |
|||
} |
|||
|
|||
public void setDeferredResultHolder(DeferredResultHolder deferredResultHolder) { |
|||
this.deferredResultHolder = deferredResultHolder; |
|||
} |
|||
|
|||
public void setLogger(Logger logger) { |
|||
this.logger = logger; |
|||
} |
|||
|
|||
} |
|||
@ -1,57 +1,58 @@ |
|||
package com.genersoft.iot.vmp.utils; |
|||
|
|||
import java.text.ParseException; |
|||
|
|||
import java.text.SimpleDateFormat; |
|||
import java.util.Date; |
|||
import java.time.Instant; |
|||
import java.time.LocalDate; |
|||
import java.time.LocalDateTime; |
|||
import java.time.ZoneId; |
|||
import java.time.format.DateTimeFormatter; |
|||
import java.time.format.DateTimeParseException; |
|||
import java.time.temporal.TemporalAccessor; |
|||
|
|||
import java.util.Locale; |
|||
|
|||
/** |
|||
* 全局时间工具类 |
|||
* @author swwheihei |
|||
* @author lin |
|||
*/ |
|||
public class DateUtil { |
|||
|
|||
private static final String yyyy_MM_dd_T_HH_mm_ss_SSSXXX = "yyyy-MM-dd'T'HH:mm:ss"; |
|||
private static final String yyyy_MM_dd_HH_mm_ss = "yyyy-MM-dd HH:mm:ss"; |
|||
public static final String yyyy_MM_dd_HH_mm_ss = "yyyy-MM-dd HH:mm:ss"; |
|||
|
|||
public static final SimpleDateFormat formatISO8601 = new SimpleDateFormat(yyyy_MM_dd_T_HH_mm_ss_SSSXXX, Locale.getDefault()); |
|||
public static final SimpleDateFormat format = new SimpleDateFormat(yyyy_MM_dd_HH_mm_ss, Locale.getDefault()); |
|||
|
|||
public static String yyyy_MM_dd_HH_mm_ssToISO8601(String formatTime) { |
|||
public static final DateTimeFormatter formatterISO8601 = DateTimeFormatter.ofPattern(yyyy_MM_dd_T_HH_mm_ss_SSSXXX, Locale.getDefault()).withZone(ZoneId.systemDefault()); |
|||
public static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern(yyyy_MM_dd_HH_mm_ss, Locale.getDefault()).withZone(ZoneId.systemDefault()); |
|||
|
|||
try { |
|||
return formatISO8601.format(format.parse(formatTime)); |
|||
} catch (ParseException e) { |
|||
e.printStackTrace(); |
|||
} |
|||
return ""; |
|||
public static String yyyy_MM_dd_HH_mm_ssToISO8601(String formatTime) { |
|||
return formatterISO8601.format(formatter.parse(formatTime)); |
|||
} |
|||
|
|||
public static String ISO8601Toyyyy_MM_dd_HH_mm_ss(String formatTime) { |
|||
return formatter.format(formatterISO8601.parse(formatTime)); |
|||
|
|||
try { |
|||
return format.format(formatISO8601.parse(formatTime)); |
|||
} catch (ParseException e) { |
|||
e.printStackTrace(); |
|||
} |
|||
return ""; |
|||
} |
|||
|
|||
public static long yyyy_MM_dd_HH_mm_ssToTimestamp(String formatTime) { |
|||
//设置要读取的时间字符串格式
|
|||
Date date; |
|||
try { |
|||
date = format.parse(formatTime); |
|||
Long timestamp=date.getTime()/1000; |
|||
//转换为Date类
|
|||
return timestamp; |
|||
} catch (ParseException e) { |
|||
e.printStackTrace(); |
|||
} |
|||
return 0; |
|||
TemporalAccessor temporalAccessor = formatter.parse(formatTime); |
|||
Instant instant = Instant.from(temporalAccessor); |
|||
return instant.getEpochSecond(); |
|||
} |
|||
|
|||
public static String getNow() { |
|||
return format.format(System.currentTimeMillis()); |
|||
LocalDateTime nowDateTime = LocalDateTime.now(); |
|||
return formatter.format(nowDateTime); |
|||
} |
|||
|
|||
public static boolean verification(String timeStr, DateTimeFormatter dateTimeFormatter) { |
|||
try { |
|||
LocalDate.parse(timeStr, dateTimeFormatter); |
|||
return true; |
|||
}catch (DateTimeParseException exception) { |
|||
return false; |
|||
} |
|||
} |
|||
} |
|||
|
|||
Loading…
Reference in new issue