15 changed files with 540 additions and 49 deletions
@ -0,0 +1 @@ |
|||
17 |
|||
@ -0,0 +1,82 @@ |
|||
package com.techsor.datacenter.sender.dao; |
|||
|
|||
import com.techsor.datacenter.sender.entitiy.f10.F10JournalLog; |
|||
import jakarta.annotation.Resource; |
|||
import org.springframework.jdbc.core.JdbcTemplate; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
@Component |
|||
public class F10Dao { |
|||
|
|||
@Resource |
|||
private JdbcTemplate jdbcTemplate ; |
|||
|
|||
public void insertF10Log(F10JournalLog f10JournalLog) { |
|||
String sql = "INSERT IGNORE INTO f10_journal_log (" + |
|||
"device_id, company_id, raw_hex, raw_text, serial_no, mode, " + |
|||
"signal_seq, receive_date, receive_time, channel_no, test_flag, block_no, block_name, " + |
|||
"signal_type1, signal_type2, signal_label, signal_status, " + |
|||
"course_no, area_code, course_code, standby_name, physical_addr, " + |
|||
"customer_name, phone, remark, display_color, error_reason, line_type, area_name, " + |
|||
"send_date, send_time, customer_no, display_data1, display_data2, " + |
|||
"instruction_flag, signal_code, card_no, card_type, mansion_building, mansion_room, option_field, " + |
|||
"created_at, created_time, receive_timestamp, expired_flag, alert_history_id" + |
|||
") VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"; |
|||
|
|||
jdbcTemplate.update(sql, |
|||
f10JournalLog.getDeviceId(), |
|||
f10JournalLog.getCompanyId(), |
|||
f10JournalLog.getRawHex(), |
|||
f10JournalLog.getRawText(), |
|||
f10JournalLog.getSerialNo(), |
|||
f10JournalLog.getMode(), |
|||
|
|||
f10JournalLog.getSignalSeq(), |
|||
f10JournalLog.getReceiveDate(), |
|||
f10JournalLog.getReceiveTime(), |
|||
f10JournalLog.getChannelNo(), |
|||
f10JournalLog.getTestFlag(), |
|||
f10JournalLog.getBlockNo(), |
|||
f10JournalLog.getBlockName(), |
|||
|
|||
f10JournalLog.getSignalType1(), |
|||
f10JournalLog.getSignalType2(), |
|||
f10JournalLog.getSignalLabel(), |
|||
f10JournalLog.getSignalStatus(), |
|||
|
|||
f10JournalLog.getCourseNo(), |
|||
f10JournalLog.getAreaCode(), |
|||
f10JournalLog.getCourseCode(), |
|||
f10JournalLog.getStandbyName(), |
|||
f10JournalLog.getPhysicalAddr(), |
|||
|
|||
f10JournalLog.getCustomerName(), |
|||
f10JournalLog.getPhone(), |
|||
f10JournalLog.getRemark(), |
|||
f10JournalLog.getDisplayColor(), |
|||
f10JournalLog.getErrorReason(), |
|||
f10JournalLog.getLineType(), |
|||
f10JournalLog.getAreaName(), |
|||
|
|||
f10JournalLog.getSendDate(), |
|||
f10JournalLog.getSendTime(), |
|||
f10JournalLog.getCustomerNo(), |
|||
f10JournalLog.getDisplayData1(), |
|||
f10JournalLog.getDisplayData2(), |
|||
|
|||
f10JournalLog.getInstructionFlag(), |
|||
f10JournalLog.getSignalCode(), |
|||
f10JournalLog.getCardNo(), |
|||
f10JournalLog.getCardType(), |
|||
f10JournalLog.getMansionBuilding(), |
|||
f10JournalLog.getMansionRoom(), |
|||
f10JournalLog.getOptionField(), |
|||
|
|||
f10JournalLog.getCreatedAt(), |
|||
new java.sql.Timestamp(f10JournalLog.getCreatedAt()), |
|||
f10JournalLog.getReceiveTimestamp(), |
|||
f10JournalLog.getExpiredFlag(), |
|||
f10JournalLog.getAlertHistoryId() |
|||
); |
|||
} |
|||
} |
|||
@ -0,0 +1,87 @@ |
|||
package com.techsor.datacenter.sender.entitiy.f10; |
|||
|
|||
import lombok.Data; |
|||
|
|||
import java.time.LocalDateTime; |
|||
|
|||
@Data |
|||
public class F10JournalLog { |
|||
|
|||
// ===== 主键 =====
|
|||
private Long id; |
|||
|
|||
// ===== 设备 =====
|
|||
private String deviceId; |
|||
private Long companyId; |
|||
|
|||
// ===== 原始数据 =====
|
|||
private String rawHex; |
|||
private String rawText; |
|||
|
|||
// ===== 报文头 =====
|
|||
private String serialNo; |
|||
private String mode; |
|||
|
|||
// ===== 1~35字段 =====
|
|||
private String signalSeq; |
|||
private String receiveDate; |
|||
private String receiveTime; |
|||
|
|||
private String channelNo; |
|||
private String testFlag; |
|||
private String blockNo; |
|||
private String blockName; |
|||
|
|||
private String signalType1; |
|||
private String signalType2; |
|||
private String signalLabel; |
|||
private String signalStatus; |
|||
|
|||
private String courseNo; |
|||
private String areaCode; |
|||
private String courseCode; |
|||
|
|||
private String standbyName; |
|||
private String physicalAddr; |
|||
|
|||
private String customerName; |
|||
private String phone; |
|||
private String remark; |
|||
|
|||
private String displayColor; |
|||
private String errorReason; |
|||
private String lineType; |
|||
|
|||
private String areaName; |
|||
|
|||
private String sendDate; |
|||
private String sendTime; |
|||
|
|||
private String customerNo; |
|||
|
|||
private String displayData1; |
|||
private String displayData2; |
|||
|
|||
private String instructionFlag; |
|||
private String signalCode; |
|||
|
|||
private String cardNo; |
|||
private String cardType; |
|||
|
|||
private String mansionBuilding; |
|||
private String mansionRoom; |
|||
|
|||
private String optionField; |
|||
|
|||
// ===== 系统字段 =====
|
|||
private Long createdAt; // 毫秒时间戳
|
|||
private LocalDateTime createdTime; // 标准时间
|
|||
|
|||
// 接收时间戳（JST）
|
|||
private Long receiveTimestamp; |
|||
|
|||
// 是否过期告警（0=正常，1=过期）
|
|||
private Integer expiredFlag; |
|||
|
|||
private Long alertHistoryId; |
|||
} |
|||
@ -0,0 +1,13 @@ |
|||
package com.techsor.datacenter.sender.entitiy.f10; |
|||
|
|||
import lombok.Data; |
|||
|
|||
@Data |
|||
public class RedisStreamEntity { |
|||
|
|||
private int partition;; |
|||
private String hexRawData; |
|||
private String payload; |
|||
private long ts; |
|||
|
|||
} |
|||
@ -0,0 +1,246 @@ |
|||
package com.techsor.datacenter.sender.service; |
|||
|
|||
import cn.hutool.core.collection.CollectionUtil; |
|||
import com.techsor.datacenter.sender.components.CommonOpt; |
|||
import com.techsor.datacenter.sender.components.GuavaRedisCache; |
|||
import com.techsor.datacenter.sender.config.DataSourceContextHolder; |
|||
import com.techsor.datacenter.sender.dao.DashboardAlertDao; |
|||
import com.techsor.datacenter.sender.dao.DeviceDao; |
|||
import com.techsor.datacenter.sender.dao.F10Dao; |
|||
import com.techsor.datacenter.sender.entitiy.DeviceEntity; |
|||
import com.techsor.datacenter.sender.entitiy.DynamodbEntity; |
|||
import com.techsor.datacenter.sender.entitiy.f10.F10JournalLog; |
|||
import com.techsor.datacenter.sender.entitiy.f10.RedisStreamEntity; |
|||
import com.techsor.datacenter.sender.utils.RedisUtils; |
|||
import jakarta.annotation.Resource; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.stereotype.Service; |
|||
|
|||
import java.nio.charset.Charset; |
|||
import java.time.LocalDateTime; |
|||
import java.time.ZoneId; |
|||
import java.time.format.DateTimeFormatter; |
|||
import java.util.Arrays; |
|||
import java.util.List; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
@Slf4j |
|||
@Service("F10Service") |
|||
public class F10Service { |
|||
|
|||
private static final Charset SHIFT_JIS = Charset.forName("Shift-JIS"); |
|||
|
|||
private static final ZoneId JST = ZoneId.of("Asia/Tokyo"); |
|||
|
|||
@Autowired |
|||
private CommonOpt commonOpt; |
|||
@Resource |
|||
private GuavaRedisCache guavaRedisCache; |
|||
@Resource |
|||
private RedisUtils redisUtils; |
|||
|
|||
@Autowired |
|||
private F10Dao f10Dao; |
|||
@Autowired |
|||
private DeviceDao deviceDao; |
|||
@Autowired |
|||
private DashboardAlertDao dashboardAlertDao; |
|||
|
|||
|
|||
|
|||
public void handle(RedisStreamEntity rawEntity) { |
|||
String hexLog = rawEntity.getHexRawData().trim(); |
|||
byte[] src = hexStringToByteArray(hexLog); |
|||
//(STX:1 + Serial:4 + Mode:2 + 正文:320 + ETX:1)
|
|||
if (src.length < 328) { |
|||
log.warn("原始报文长度不足328字节"); |
|||
return; |
|||
} |
|||
|
|||
try { |
|||
long currentTs = System.currentTimeMillis(); |
|||
|
|||
F10JournalLog f10JournalLog = parseToEntity(rawEntity.getHexRawData()); |
|||
f10JournalLog.setCreatedAt(currentTs); |
|||
|
|||
//前缀_26契約先番号_4チャンネル番号
|
|||
//前缀还没定，先【26契約先番号_4チャンネル番号】
|
|||
f10JournalLog.setDeviceId(f10JournalLog.getCustomerNo() + "_" + f10JournalLog.getChannelNo()); |
|||
|
|||
long receivedCurrentTs = parseToTimestamp(f10JournalLog.getReceiveDate(), f10JournalLog.getReceiveTime()); |
|||
f10JournalLog.setReceiveTimestamp(receivedCurrentTs); |
|||
|
|||
String redisKey = getLatestAlarmKey(f10JournalLog.getDeviceId()); |
|||
// 从 Redis 取值
|
|||
String latestTsStr = redisUtils.get(redisKey); |
|||
Long latestTs = null; |
|||
if (latestTsStr != null) { |
|||
try { |
|||
latestTs = Long.parseLong(latestTsStr); |
|||
} catch (Exception e) { |
|||
log.error("Redis数据异常，无法转换为Long: key={}, value={}", redisKey, latestTsStr); |
|||
} |
|||
} |
|||
|
|||
boolean isExpired = false; |
|||
// ===== 判断逻辑 =====
|
|||
if (latestTs == null) { |
|||
// 首次
|
|||
redisUtils.add(redisKey, String.valueOf(receivedCurrentTs)); |
|||
} else { |
|||
if (receivedCurrentTs > latestTs) { |
|||
// 新数据
|
|||
redisUtils.add(redisKey, String.valueOf(receivedCurrentTs)); |
|||
} else if (receivedCurrentTs == latestTs) { |
|||
// 重复数据 → 直接丢弃
|
|||
log.warn("重复告警，丢弃 deviceId={}", f10JournalLog.getDeviceId()); |
|||
return; |
|||
} else { |
|||
// 旧数据 → 标记过期
|
|||
isExpired = true; |
|||
} |
|||
} |
|||
f10JournalLog.setExpiredFlag(isExpired ? 1 : 0); |
|||
|
|||
String topCompanyId= commonOpt.getTopCompanyId(f10JournalLog.getDeviceId()); |
|||
if (topCompanyId==null || topCompanyId.equals("0")){ |
|||
log.error("no companyId is found========================>>>> {}",f10JournalLog.getDeviceId()); |
|||
return; |
|||
} |
|||
DataSourceContextHolder.clearCurrentDataSourceKey(); |
|||
DataSourceContextHolder.setCurrentDataSourceKey("dataSourceForCompany_"+topCompanyId); |
|||
|
|||
log.info("Use datasource for company:"+topCompanyId); |
|||
List<DeviceEntity> deviceEntity=this.deviceDao.getDeviceInfoByDeviceId(f10JournalLog.getDeviceId()); |
|||
this.guavaRedisCache.incrementDailyDeviceIdCount(f10JournalLog.getDeviceId()); |
|||
if (CollectionUtil.isEmpty(deviceEntity)){ |
|||
log.error("no deviceEntity is found========================>>>> {}",f10JournalLog.getDeviceId()); |
|||
return; |
|||
} |
|||
|
|||
Long companyId = deviceEntity.get(0).getCompanyId(); |
|||
f10JournalLog.setCompanyId(companyId); |
|||
|
|||
if (!isExpired){ |
|||
DynamodbEntity entity = new DynamodbEntity(); |
|||
entity.setDeviceId(f10JournalLog.getDeviceId()); |
|||
entity.setReceive_ts(currentTs); |
|||
Long dbId = dashboardAlertDao.insertAlertHistory(entity, 2); |
|||
f10JournalLog.setAlertHistoryId(dbId); |
|||
} |
|||
|
|||
|
|||
f10Dao.insertF10Log(f10JournalLog); |
|||
|
|||
} catch (Exception e) { |
|||
log.error("F10解析入库失败", e); |
|||
} |
|||
} |
|||
|
|||
public static String getLatestAlarmKey(String deviceId) { |
|||
return "f10:latest_alarm:" + deviceId; |
|||
} |
|||
|
|||
public static F10JournalLog parseToEntity(String hexLog) { |
|||
byte[] src = hexStringToByteArray(hexLog); |
|||
|
|||
int offset = 0; |
|||
|
|||
// 跳过 STX
|
|||
offset += 1; |
|||
|
|||
String serialNo = new String(Arrays.copyOfRange(src, offset, offset + 4), SHIFT_JIS); |
|||
String mode = new String(Arrays.copyOfRange(src, offset, offset + 2), SHIFT_JIS); |
|||
|
|||
int bodyStart = 7; |
|||
int bodyEnd = src.length - 1; |
|||
byte[] body = Arrays.copyOfRange(src, bodyStart, bodyEnd); |
|||
String rawText = new String(body, SHIFT_JIS).trim(); |
|||
|
|||
F10JournalLog f10JournalLog = new F10JournalLog(); |
|||
f10JournalLog.setRawHex(hexLog); |
|||
f10JournalLog.setRawText(rawText); |
|||
f10JournalLog.setSerialNo(serialNo); |
|||
f10JournalLog.setMode(mode); |
|||
|
|||
offset = 7; |
|||
|
|||
f10JournalLog.setSignalSeq(getTargetData(src, offset, 9)); offset += 9; |
|||
f10JournalLog.setReceiveDate(getTargetData(src, offset, 8)); offset += 8; |
|||
f10JournalLog.setReceiveTime(getTargetData(src, offset, 6)); offset += 6; |
|||
|
|||
f10JournalLog.setChannelNo(getTargetData(src, offset, 3)); offset += 3; |
|||
f10JournalLog.setTestFlag(getTargetData(src, offset, 1)); offset += 1; |
|||
f10JournalLog.setBlockNo(getTargetData(src, offset, 2)); offset += 2; |
|||
f10JournalLog.setBlockName(getTargetData(src, offset, 30)); offset += 30; |
|||
|
|||
f10JournalLog.setSignalType1(getTargetData(src, offset, 2)); offset += 2; |
|||
f10JournalLog.setSignalType2(getTargetData(src, offset, 2)); offset += 2; |
|||
f10JournalLog.setSignalLabel(getTargetData(src, offset, 2)); offset += 2; |
|||
f10JournalLog.setSignalStatus(getTargetData(src, offset, 1)); offset += 1; |
|||
|
|||
f10JournalLog.setCourseNo(getTargetData(src, offset, 3)); offset += 3; |
|||
f10JournalLog.setAreaCode(getTargetData(src, offset, 2)); offset += 2; |
|||
f10JournalLog.setCourseCode(getTargetData(src, offset, 3)); offset += 3; |
|||
|
|||
f10JournalLog.setStandbyName(getTargetData(src, offset, 20)); offset += 20; |
|||
f10JournalLog.setPhysicalAddr(getTargetData(src, offset, 8)); offset += 8; |
|||
|
|||
f10JournalLog.setCustomerName(getTargetData(src, offset, 30)); offset += 30; |
|||
f10JournalLog.setPhone(getTargetData(src, offset, 20)); offset += 20; |
|||
f10JournalLog.setRemark(getTargetData(src, offset, 60)); offset += 60; |
|||
|
|||
f10JournalLog.setDisplayColor(getTargetData(src, offset, 2)); offset += 2; |
|||
f10JournalLog.setErrorReason(getTargetData(src, offset, 1)); offset += 1; |
|||
f10JournalLog.setLineType(getTargetData(src, offset, 2)); offset += 2; |
|||
|
|||
f10JournalLog.setAreaName(getTargetData(src, offset, 12)); offset += 12; |
|||
|
|||
f10JournalLog.setSendDate(getTargetData(src, offset, 8)); offset += 8; |
|||
f10JournalLog.setSendTime(getTargetData(src, offset, 6)); offset += 6; |
|||
|
|||
f10JournalLog.setCustomerNo(getTargetData(src, offset, 8)); offset += 8; |
|||
|
|||
f10JournalLog.setDisplayData1(getTargetData(src, offset, 20)); offset += 20; |
|||
f10JournalLog.setDisplayData2(getTargetData(src, offset, 20)); offset += 20; |
|||
|
|||
f10JournalLog.setInstructionFlag(getTargetData(src, offset, 1)); offset += 1; |
|||
f10JournalLog.setSignalCode(getTargetData(src, offset, 4)); offset += 4; |
|||
|
|||
f10JournalLog.setCardNo(getTargetData(src, offset, 7)); offset += 7; |
|||
f10JournalLog.setCardType(getTargetData(src, offset, 1)); offset += 1; |
|||
|
|||
f10JournalLog.setMansionBuilding(getTargetData(src, offset, 2)); offset += 2; |
|||
f10JournalLog.setMansionRoom(getTargetData(src, offset, 4)); offset += 4; |
|||
|
|||
f10JournalLog.setOptionField(getTargetData(src, offset, 10)); |
|||
|
|||
return f10JournalLog; |
|||
} |
|||
|
|||
private static String getTargetData(byte[] src, int start, int len) { |
|||
return new String(Arrays.copyOfRange(src, start, start + len), SHIFT_JIS).trim(); |
|||
} |
|||
|
|||
private static byte[] hexStringToByteArray(String s) { |
|||
s = s.replace(" ", ""); |
|||
int len = s.length(); |
|||
byte[] data = new byte[len / 2]; |
|||
for (int i = 0; i < len; i += 2) { |
|||
data[i / 2] = (byte) ((Character.digit(s.charAt(i), 16) << 4) |
|||
+ Character.digit(s.charAt(i + 1), 16)); |
|||
} |
|||
return data; |
|||
} |
|||
|
|||
public static long parseToTimestamp(String date, String time) { |
|||
LocalDateTime ldt = LocalDateTime.parse( |
|||
date + time, |
|||
DateTimeFormatter.ofPattern("yyyyMMddHHmmss") |
|||
); |
|||
|
|||
return ldt.atZone(JST).toInstant().toEpochMilli(); |
|||
} |
|||
|
|||
} |
|||
Loading…
Reference in new issue