package com.zy.asrs.service.impl;
|
|
import com.alibaba.fastjson.JSON;
|
import com.alibaba.fastjson.serializer.SerializerFeature;
|
import com.zy.asrs.domain.replay.DeviceReplayChunkMeta;
|
import com.zy.asrs.domain.replay.DeviceReplayManifest;
|
import com.zy.asrs.domain.replay.DeviceReplayStreamKey;
|
import com.zy.asrs.entity.DeviceDataLog;
|
import com.zy.asrs.service.DeviceLogReplayManifestService;
|
import lombok.extern.slf4j.Slf4j;
|
import jakarta.annotation.PreDestroy;
|
import org.springframework.beans.factory.annotation.Value;
|
import org.springframework.stereotype.Service;
|
|
import java.io.BufferedReader;
|
import java.nio.charset.StandardCharsets;
|
import java.nio.file.Files;
|
import java.nio.file.StandardCopyOption;
|
import java.nio.file.Path;
|
import java.nio.file.Paths;
|
import java.time.ZoneId;
|
import java.util.ArrayList;
|
import java.util.Comparator;
|
import java.util.LinkedHashSet;
|
import java.util.List;
|
import java.util.Objects;
|
import java.util.Set;
|
import java.util.concurrent.ConcurrentHashMap;
|
import java.util.concurrent.ExecutorService;
|
import java.util.concurrent.Executors;
|
import java.util.concurrent.RejectedExecutionException;
|
import java.util.concurrent.ScheduledExecutorService;
|
import java.util.concurrent.ScheduledFuture;
|
import java.util.concurrent.TimeUnit;
|
import java.util.concurrent.atomic.AtomicBoolean;
|
import java.util.concurrent.atomic.AtomicInteger;
|
import java.util.regex.Matcher;
|
import java.util.regex.Pattern;
|
import java.util.stream.Collectors;
|
import java.util.stream.Stream;
|
|
@Slf4j
|
@Service
|
public class DeviceLogReplayManifestServiceImpl implements DeviceLogReplayManifestService {
|
|
private static final String MANIFEST_DIR_NAME = ".replay-manifest";
|
private static final int MANIFEST_SCHEMA_VERSION = 1;
|
private static final Pattern DEFAULT_FILE_PATTERN = Pattern.compile("^([^_]+)_(\\d+)_(\\d{8})_(\\d+)\\.log$");
|
private static final Pattern DEVP_FILE_PATTERN = Pattern.compile("^([^_]+)_(\\d+)_station_(\\d+)_(\\d{8})_(\\d+)\\.log$");
|
private static final long MANIFEST_STALE_THRESHOLD_MS = 5_000L;
|
private static final long EXECUTOR_SHUTDOWN_WAIT_MS = 5_000L;
|
private static final ZoneId REPLAY_ZONE = ZoneId.systemDefault();
|
|
@Value("${deviceLogStorage.loggingPath}")
|
private String loggingPath;
|
|
@Value("${deviceLogStorage.manifestRefreshDebounceMs:1000}")
|
private long manifestRefreshDebounceMs;
|
|
private final ExecutorService dayRefreshExecutor = Executors.newSingleThreadExecutor(runnable -> {
|
Thread thread = new Thread(runnable, "replay-manifest-refresh");
|
thread.setDaemon(true);
|
return thread;
|
});
|
|
private final ScheduledExecutorService streamRefreshExecutor = Executors.newSingleThreadScheduledExecutor(runnable -> {
|
Thread thread = new Thread(runnable, "replay-manifest-stream-refresh");
|
thread.setDaemon(true);
|
return thread;
|
});
|
|
private final ConcurrentHashMap<String, Boolean> refreshingDays = new ConcurrentHashMap<>();
|
private final ConcurrentHashMap<String, Boolean> refreshingStreams = new ConcurrentHashMap<>();
|
private final ConcurrentHashMap<String, StreamRefreshState> streamRefreshStateByStream = new ConcurrentHashMap<>();
|
private final AtomicBoolean shuttingDown = new AtomicBoolean(false);
|
|
@Override
|
public DeviceReplayManifest loadManifest(DeviceReplayStreamKey streamKey, boolean autoBuild) {
|
if (streamKey == null || streamKey.getDay() == null) {
|
return null;
|
}
|
Path manifestPath = resolveManifestPath(streamKey);
|
if (Files.exists(manifestPath)) {
|
try {
|
String json = Files.readString(manifestPath, StandardCharsets.UTF_8);
|
return JSON.parseObject(json, DeviceReplayManifest.class);
|
} catch (Exception e) {
|
log.warn("读取 replay manifest 失败，准备重建, path={}", manifestPath, e);
|
}
|
}
|
if (!autoBuild) {
|
return null;
|
}
|
return rebuildManifest(streamKey);
|
}
|
|
@Override
|
public List<DeviceReplayManifest> loadDayManifests(String day, boolean autoBuild) {
|
List<DeviceReplayManifest> manifests = new ArrayList<>();
|
for (DeviceReplayStreamKey streamKey : discoverStreamKeys(day, autoBuild)) {
|
DeviceReplayManifest manifest = loadManifest(streamKey, autoBuild);
|
if (manifest != null) {
|
manifests.add(manifest);
|
}
|
}
|
manifests.sort(Comparator
|
.comparing(DeviceReplayManifest::getFirstSampleTimeMs, Comparator.nullsLast(Long::compareTo))
|
.thenComparing(DeviceReplayManifest::getType, Comparator.nullsLast(String::compareTo))
|
.thenComparing(DeviceReplayManifest::getDeviceNo, Comparator.nullsLast(Integer::compareTo))
|
.thenComparing(DeviceReplayManifest::getStationId, Comparator.nullsLast(Integer::compareTo)));
|
return manifests;
|
}
|
|
@Override
|
public List<DeviceReplayManifest> refreshDayManifestsIfStale(String day) {
|
if (day == null || day.isBlank()) {
|
return new ArrayList<>();
|
}
|
List<DeviceReplayStreamKey> staleStreamKeys = resolveStaleStreamKeys(day);
|
if (staleStreamKeys.isEmpty()) {
|
return loadDayManifests(day, false);
|
}
|
for (DeviceReplayStreamKey streamKey : staleStreamKeys) {
|
refreshStreamManifestIfStale(streamKey);
|
}
|
return loadDayManifests(day, false);
|
}
|
|
@Override
|
public void scheduleDayManifestRefresh(String day) {
|
if (day == null || day.isBlank()) {
|
return;
|
}
|
if (shuttingDown.get()) {
|
return;
|
}
|
if (refreshingDays.putIfAbsent(day, Boolean.TRUE) != null) {
|
return;
|
}
|
try {
|
dayRefreshExecutor.submit(() -> {
|
try {
|
refreshDayManifestsIfStale(day);
|
} catch (Exception e) {
|
log.warn("异步刷新当天 replay manifest 失败, day={}", day, e);
|
} finally {
|
refreshingDays.remove(day);
|
}
|
});
|
} catch (RejectedExecutionException rejectedExecutionException) {
|
refreshingDays.remove(day);
|
if (!shuttingDown.get()) {
|
log.warn("提交当天 replay manifest 刷新任务失败, day={}", day, rejectedExecutionException);
|
}
|
}
|
}
|
|
@Override
|
public DeviceReplayManifest rebuildManifest(DeviceReplayStreamKey streamKey) {
|
if (streamKey == null || streamKey.getDay() == null) {
|
return null;
|
}
|
List<ParsedReplayFile> files = resolveStreamFiles(streamKey);
|
if (files.isEmpty()) {
|
return null;
|
}
|
DeviceReplayManifest manifest = new DeviceReplayManifest();
|
manifest.setSchemaVersion(MANIFEST_SCHEMA_VERSION);
|
manifest.setDay(streamKey.getDay());
|
manifest.setType(streamKey.getType());
|
manifest.setDeviceNo(streamKey.getDeviceNo());
|
manifest.setStationId(streamKey.getStationId());
|
manifest.setSampleMode("SNAPSHOT");
|
manifest.setTotalSamples(0);
|
|
long generatedSeqCursor = 0L;
|
int chunkIndex = 0;
|
for (ParsedReplayFile parsedReplayFile : files) {
|
DeviceReplayChunkMeta chunkMeta = new DeviceReplayChunkMeta();
|
chunkMeta.setChunkIndex(chunkIndex++);
|
chunkMeta.setFileName(parsedReplayFile.file.getFileName().toString());
|
chunkMeta.setRelativePath(buildRelativeLogPath(streamKey, parsedReplayFile.file.getFileName().toString()));
|
chunkMeta.setSealed(Boolean.TRUE);
|
chunkMeta.setSampleCount(0);
|
try (BufferedReader reader = Files.newBufferedReader(parsedReplayFile.file, StandardCharsets.UTF_8)) {
|
String line;
|
while ((line = reader.readLine()) != null) {
|
if (line == null || line.isBlank()) {
|
continue;
|
}
|
DeviceDataLog logItem;
|
try {
|
logItem = JSON.parseObject(line, DeviceDataLog.class);
|
} catch (Exception parseError) {
|
log.debug("跳过损坏的 replay manifest 日志行, file={}", parsedReplayFile.file, parseError);
|
continue;
|
}
|
if (!matchesStream(streamKey, logItem)) {
|
continue;
|
}
|
generatedSeqCursor++;
|
long sampleSeq = resolveSampleSeq(logItem, generatedSeqCursor);
|
long sampleTimeMs = resolveSampleTimeMs(logItem);
|
if (chunkMeta.getFirstSampleSeq() == null) {
|
chunkMeta.setFirstSampleSeq(sampleSeq);
|
chunkMeta.setFirstSampleTimeMs(sampleTimeMs);
|
}
|
chunkMeta.setLastSampleSeq(sampleSeq);
|
chunkMeta.setLastSampleTimeMs(sampleTimeMs);
|
chunkMeta.setSampleCount(chunkMeta.getSampleCount() + 1);
|
}
|
} catch (Exception e) {
|
log.warn("构建 replay manifest chunk 失败, file={}", parsedReplayFile.file, e);
|
continue;
|
}
|
if (chunkMeta.getSampleCount() == null || chunkMeta.getSampleCount() <= 0) {
|
continue;
|
}
|
manifest.getChunks().add(chunkMeta);
|
manifest.setTotalSamples(manifest.getTotalSamples() + chunkMeta.getSampleCount());
|
if (manifest.getFirstSampleSeq() == null) {
|
manifest.setFirstSampleSeq(chunkMeta.getFirstSampleSeq());
|
manifest.setFirstSampleTimeMs(chunkMeta.getFirstSampleTimeMs());
|
}
|
manifest.setLastSampleSeq(chunkMeta.getLastSampleSeq());
|
manifest.setLastSampleTimeMs(chunkMeta.getLastSampleTimeMs());
|
}
|
if (manifest.getChunks().isEmpty()) {
|
return null;
|
}
|
writeManifest(streamKey, manifest);
|
return manifest;
|
}
|
|
@Override
|
public void scheduleManifestRefreshForLog(String day, DeviceDataLog logItem) {
|
DeviceReplayStreamKey streamKey = DeviceReplayStreamKey.fromLog(day, logItem);
|
if (streamKey == null) {
|
return;
|
}
|
StreamRefreshState refreshState = getOrCreateStreamRefreshState(streamKey.streamId());
|
synchronized (refreshState.monitor) {
|
refreshState.latestStreamKey = streamKey;
|
}
|
if (shuttingDown.get()) {
|
refreshStreamManifest(streamKey);
|
return;
|
}
|
scheduleStreamManifestRefresh(streamKey);
|
}
|
|
@Override
|
public void refreshManifestForLog(String day, DeviceDataLog logItem) {
|
scheduleManifestRefreshForLog(day, logItem);
|
}
|
|
@PreDestroy
|
public void shutdownRefreshExecutor() {
|
if (!shuttingDown.compareAndSet(false, true)) {
|
return;
|
}
|
shutdownExecutor(dayRefreshExecutor, "replay-manifest-refresh");
|
shutdownExecutor(streamRefreshExecutor, "replay-manifest-stream-refresh");
|
List<DeviceReplayStreamKey> pendingStreamKeys = streamRefreshStateByStream.values().stream()
|
.map(refreshState -> refreshState.latestStreamKey)
|
.filter(Objects::nonNull)
|
.collect(Collectors.toList());
|
for (DeviceReplayStreamKey streamKey : pendingStreamKeys) {
|
try {
|
refreshStreamManifest(streamKey);
|
} catch (Exception e) {
|
log.warn("关闭前补刷 replay manifest 失败, streamId={}", streamKey.streamId(), e);
|
}
|
}
|
streamRefreshStateByStream.clear();
|
}
|
|
private void writeManifest(DeviceReplayStreamKey streamKey, DeviceReplayManifest manifest) {
|
Path manifestPath = resolveManifestPath(streamKey);
|
Path tempPath = manifestPath.resolveSibling(manifestPath.getFileName().toString() + ".tmp");
|
try {
|
Files.createDirectories(manifestPath.getParent());
|
String json = JSON.toJSONString(manifest, SerializerFeature.PrettyFormat);
|
Files.writeString(tempPath, json, StandardCharsets.UTF_8);
|
Files.move(tempPath, manifestPath, StandardCopyOption.REPLACE_EXISTING, StandardCopyOption.ATOMIC_MOVE);
|
} catch (Exception e) {
|
log.warn("写入 replay manifest 失败, path={}", manifestPath, e);
|
try {
|
Files.deleteIfExists(tempPath);
|
} catch (Exception ignored) {
|
}
|
}
|
}
|
|
private Set<DeviceReplayStreamKey> discoverStreamKeys(String day, boolean allowLogScan) {
|
LinkedHashSet<DeviceReplayStreamKey> streamKeys = new LinkedHashSet<>();
|
Path manifestDayDir = Paths.get(loggingPath, day, MANIFEST_DIR_NAME);
|
if (Files.exists(manifestDayDir)) {
|
try (Stream<Path> stream = Files.walk(manifestDayDir, 3)) {
|
stream.filter(Files::isRegularFile)
|
.filter(path -> path.getFileName().toString().endsWith(".json"))
|
.forEach(path -> {
|
DeviceReplayStreamKey streamKey = parseManifestPath(day, path);
|
if (streamKey != null) {
|
streamKeys.add(streamKey);
|
}
|
});
|
} catch (Exception e) {
|
log.warn("扫描 replay manifest 失败, day={}", day, e);
|
}
|
}
|
if (!allowLogScan) {
|
return streamKeys;
|
}
|
Path dayDir = Paths.get(loggingPath, day);
|
if (!Files.exists(dayDir)) {
|
return streamKeys;
|
}
|
try (Stream<Path> stream = Files.walk(dayDir, 3)) {
|
stream.filter(Files::isRegularFile)
|
.filter(path -> path.getFileName().toString().endsWith(".log"))
|
.forEach(path -> {
|
ParsedReplayFile parsed = parseReplayFile(path.getFileName().toString());
|
if (parsed == null || !day.equals(parsed.day)) {
|
return;
|
}
|
streamKeys.add(DeviceReplayStreamKey.of(parsed.day, parsed.type, parsed.deviceNo, parsed.stationId));
|
});
|
} catch (Exception e) {
|
log.warn("扫描 replay stream 失败, day={}", day, e);
|
}
|
return streamKeys;
|
}
|
|
private void scheduleStreamManifestRefresh(DeviceReplayStreamKey streamKey) {
|
if (streamKey == null || streamKey.getDay() == null) {
|
return;
|
}
|
long requestTime = System.currentTimeMillis();
|
StreamRefreshState refreshState = getOrCreateStreamRefreshState(streamKey.streamId());
|
refreshState.activeUsers.incrementAndGet();
|
try {
|
synchronized (refreshState.monitor) {
|
refreshState.lastScheduleTimeMs = requestTime;
|
refreshState.latestStreamKey = streamKey;
|
scheduleDebouncedStreamRefreshLocked(refreshState, streamKey, requestTime, normalizeDebounceDelay(manifestRefreshDebounceMs));
|
}
|
} finally {
|
if (refreshState.activeUsers.decrementAndGet() == 0) {
|
tryCleanupStreamRefreshState(streamKey.streamId(), refreshState);
|
}
|
}
|
}
|
|
private void scheduleDebouncedStreamRefreshLocked(StreamRefreshState refreshState,
|
DeviceReplayStreamKey streamKey,
|
long requestTime,
|
long delayMs) {
|
if (streamKey == null || streamKey.getDay() == null) {
|
return;
|
}
|
ScheduledFuture<?> existingFuture = refreshState.scheduledFuture;
|
if (existingFuture != null && !existingFuture.isDone()) {
|
return;
|
}
|
try {
|
refreshState.scheduledFuture = streamRefreshExecutor.schedule(
|
() -> executeScheduledStreamRefresh(streamKey, requestTime),
|
normalizeDebounceDelay(delayMs),
|
TimeUnit.MILLISECONDS
|
);
|
} catch (RejectedExecutionException rejectedExecutionException) {
|
refreshState.scheduledFuture = null;
|
if (shuttingDown.get()) {
|
refreshStreamManifest(streamKey);
|
return;
|
}
|
throw rejectedExecutionException;
|
}
|
}
|
|
private void executeScheduledStreamRefresh(DeviceReplayStreamKey streamKey, long scheduledRequestTime) {
|
if (streamKey == null || streamKey.getDay() == null) {
|
return;
|
}
|
String streamId = streamKey.streamId();
|
StreamRefreshState refreshState = streamRefreshStateByStream.get(streamId);
|
if (refreshState == null) {
|
return;
|
}
|
refreshState.activeUsers.incrementAndGet();
|
try {
|
DeviceReplayStreamKey latestStreamKey;
|
long latestRequestTime;
|
synchronized (refreshState.monitor) {
|
refreshState.scheduledFuture = null;
|
latestRequestTime = refreshState.lastScheduleTimeMs > 0 ? refreshState.lastScheduleTimeMs : scheduledRequestTime;
|
latestStreamKey = resolveLatestStreamKey(refreshState, streamKey);
|
long remainingDelayMs = calculateRemainingDebounceDelay(latestRequestTime);
|
if (remainingDelayMs > 0) {
|
scheduleDebouncedStreamRefreshLocked(refreshState, latestStreamKey, latestRequestTime, remainingDelayMs);
|
return;
|
}
|
}
|
if (refreshingStreams.putIfAbsent(streamId, Boolean.TRUE) != null) {
|
return;
|
}
|
|
long refreshStartTime = System.currentTimeMillis();
|
try {
|
refreshStreamManifest(latestStreamKey);
|
} catch (Exception e) {
|
log.warn("异步刷新 replay manifest stream 失败, streamId={}", streamId, e);
|
} finally {
|
refreshingStreams.remove(streamId);
|
scheduleTrailingStreamRefreshIfNeeded(refreshState, latestStreamKey, refreshStartTime);
|
}
|
} finally {
|
if (refreshState.activeUsers.decrementAndGet() == 0) {
|
tryCleanupStreamRefreshState(streamId, refreshState);
|
}
|
}
|
}
|
|
private void scheduleTrailingStreamRefreshIfNeeded(StreamRefreshState refreshState,
|
DeviceReplayStreamKey fallbackStreamKey,
|
long refreshStartTime) {
|
synchronized (refreshState.monitor) {
|
long latestRequestTime = refreshState.lastScheduleTimeMs;
|
if (latestRequestTime <= 0L) {
|
return;
|
}
|
if (latestRequestTime <= refreshStartTime) {
|
if (refreshState.scheduledFuture == null || refreshState.scheduledFuture.isDone()) {
|
refreshState.lastScheduleTimeMs = 0L;
|
refreshState.latestStreamKey = null;
|
}
|
return;
|
}
|
DeviceReplayStreamKey latestStreamKey = resolveLatestStreamKey(refreshState, fallbackStreamKey);
|
scheduleDebouncedStreamRefreshLocked(
|
refreshState,
|
latestStreamKey,
|
latestRequestTime,
|
calculateRemainingDebounceDelay(latestRequestTime)
|
);
|
}
|
}
|
|
private long calculateRemainingDebounceDelay(long requestTime) {
|
long debounceDelayMs = normalizeDebounceDelay(manifestRefreshDebounceMs);
|
if (debounceDelayMs <= 0) {
|
return 0L;
|
}
|
long elapsedMs = System.currentTimeMillis() - requestTime;
|
if (elapsedMs >= debounceDelayMs) {
|
return 0L;
|
}
|
return debounceDelayMs - elapsedMs;
|
}
|
|
private long normalizeDebounceDelay(long delayMs) {
|
return Math.max(delayMs, 0L);
|
}
|
|
private DeviceReplayManifest refreshStreamManifestIfStale(DeviceReplayStreamKey streamKey) {
|
return refreshStreamManifest(streamKey, true);
|
}
|
|
private DeviceReplayManifest refreshStreamManifest(DeviceReplayStreamKey streamKey) {
|
return refreshStreamManifest(streamKey, false);
|
}
|
|
private DeviceReplayManifest refreshStreamManifest(DeviceReplayStreamKey streamKey, boolean onlyIfStale) {
|
if (streamKey == null) {
|
return null;
|
}
|
String streamId = streamKey.streamId();
|
StreamRefreshState refreshState = getOrCreateStreamRefreshState(streamId);
|
refreshState.activeUsers.incrementAndGet();
|
try {
|
synchronized (refreshState.monitor) {
|
if (onlyIfStale && !isManifestStale(streamKey)) {
|
return loadManifest(streamKey, false);
|
}
|
return rebuildManifest(streamKey);
|
}
|
} finally {
|
if (refreshState.activeUsers.decrementAndGet() == 0) {
|
tryCleanupStreamRefreshState(streamId, refreshState);
|
}
|
}
|
}
|
|
private StreamRefreshState getOrCreateStreamRefreshState(String streamId) {
|
return streamRefreshStateByStream.computeIfAbsent(streamId, key -> new StreamRefreshState());
|
}
|
|
private DeviceReplayStreamKey resolveLatestStreamKey(StreamRefreshState refreshState,
|
DeviceReplayStreamKey fallbackStreamKey) {
|
DeviceReplayStreamKey latestStreamKey = refreshState.latestStreamKey;
|
return latestStreamKey == null ? fallbackStreamKey : latestStreamKey;
|
}
|
|
private void tryCleanupStreamRefreshState(String streamId, StreamRefreshState refreshState) {
|
if (shuttingDown.get() || refreshState.activeUsers.get() > 0) {
|
return;
|
}
|
synchronized (refreshState.monitor) {
|
if (shuttingDown.get() || refreshState.activeUsers.get() > 0) {
|
return;
|
}
|
ScheduledFuture<?> scheduledFuture = refreshState.scheduledFuture;
|
if (scheduledFuture != null && !scheduledFuture.isDone()) {
|
return;
|
}
|
if (refreshState.lastScheduleTimeMs > 0L || refreshState.latestStreamKey != null) {
|
return;
|
}
|
if (refreshingStreams.containsKey(streamId)) {
|
return;
|
}
|
streamRefreshStateByStream.remove(streamId, refreshState);
|
}
|
}
|
|
private void shutdownExecutor(ExecutorService executorService, String executorName) {
|
executorService.shutdown();
|
try {
|
if (executorService.awaitTermination(EXECUTOR_SHUTDOWN_WAIT_MS, TimeUnit.MILLISECONDS)) {
|
return;
|
}
|
List<Runnable> droppedTasks = executorService.shutdownNow();
|
log.warn("等待 {} 关闭超时，强制停止剩余任务, droppedTaskCount={}", executorName, droppedTasks.size());
|
if (!executorService.awaitTermination(EXECUTOR_SHUTDOWN_WAIT_MS, TimeUnit.MILLISECONDS)) {
|
log.warn("{} 强制停止后仍未完全退出", executorName);
|
}
|
} catch (InterruptedException interruptedException) {
|
Thread.currentThread().interrupt();
|
List<Runnable> droppedTasks = executorService.shutdownNow();
|
log.warn("等待 {} 关闭被中断，已强制停止剩余任务, droppedTaskCount={}", executorName, droppedTasks.size(), interruptedException);
|
}
|
}
|
|
private List<DeviceReplayStreamKey> resolveStaleStreamKeys(String day) {
|
Set<DeviceReplayStreamKey> streamKeys = discoverStreamKeys(day, true);
|
List<DeviceReplayStreamKey> staleStreamKeys = new ArrayList<>();
|
for (DeviceReplayStreamKey streamKey : streamKeys) {
|
if (isManifestStale(streamKey)) {
|
staleStreamKeys.add(streamKey);
|
}
|
}
|
return staleStreamKeys;
|
}
|
|
private boolean isManifestStale(DeviceReplayStreamKey streamKey) {
|
if (streamKey == null) {
|
return false;
|
}
|
long latestLogModifiedTime = resolveLatestLogModifiedTime(streamKey);
|
if (latestLogModifiedTime <= 0) {
|
return false;
|
}
|
Path manifestPath = resolveManifestPath(streamKey);
|
if (!Files.exists(manifestPath)) {
|
return true;
|
}
|
try {
|
long manifestModifiedTime = Files.getLastModifiedTime(manifestPath).toMillis();
|
return latestLogModifiedTime - manifestModifiedTime > MANIFEST_STALE_THRESHOLD_MS;
|
} catch (Exception e) {
|
log.warn("判断 replay manifest 是否过期失败, path={}", manifestPath, e);
|
return true;
|
}
|
}
|
|
private long resolveLatestLogModifiedTime(DeviceReplayStreamKey streamKey) {
|
long latestModifiedTime = 0L;
|
for (ParsedReplayFile replayFile : resolveStreamFiles(streamKey)) {
|
try {
|
latestModifiedTime = Math.max(latestModifiedTime, Files.getLastModifiedTime(replayFile.file).toMillis());
|
} catch (Exception e) {
|
log.debug("读取 replay 日志文件修改时间失败, file={}", replayFile.file, e);
|
}
|
}
|
return latestModifiedTime;
|
}
|
|
private DeviceReplayStreamKey parseManifestPath(String day, Path manifestPath) {
|
if (manifestPath == null || manifestPath.getParent() == null) {
|
return null;
|
}
|
Path typePath = manifestPath.getParent();
|
Path filePath = manifestPath.getFileName();
|
if (typePath == null || filePath == null) {
|
return null;
|
}
|
String type = typePath.getFileName() == null ? null : typePath.getFileName().toString();
|
String fileName = filePath.toString();
|
Matcher stationMatcher = Pattern.compile("^(\\d+)_station_(\\d+)\\.json$").matcher(fileName);
|
if (stationMatcher.matches()) {
|
return DeviceReplayStreamKey.of(day, type, Integer.parseInt(stationMatcher.group(1)), Integer.parseInt(stationMatcher.group(2)));
|
}
|
Matcher defaultMatcher = Pattern.compile("^(\\d+)\\.json$").matcher(fileName);
|
if (defaultMatcher.matches()) {
|
return DeviceReplayStreamKey.of(day, type, Integer.parseInt(defaultMatcher.group(1)), null);
|
}
|
return null;
|
}
|
|
private List<ParsedReplayFile> resolveStreamFiles(DeviceReplayStreamKey streamKey) {
|
Path deviceDir = Paths.get(loggingPath, streamKey.getDay(), streamKey.getType(), String.valueOf(streamKey.getDeviceNo()));
|
if (!Files.exists(deviceDir)) {
|
return new ArrayList<>();
|
}
|
try (Stream<Path> stream = Files.list(deviceDir)) {
|
return stream.filter(Files::isRegularFile)
|
.map(path -> parseReplayFile(path.getFileName().toString(), path))
|
.filter(parsed -> parsed != null
|
&& streamKey.getDay().equals(parsed.day)
|
&& streamKey.getType().equals(parsed.type)
|
&& streamKey.getDeviceNo().equals(parsed.deviceNo)
|
&& equalsNullable(streamKey.getStationId(), parsed.stationId))
|
.sorted(Comparator.comparing(parsed -> parsed.index))
|
.collect(Collectors.toList());
|
} catch (Exception e) {
|
log.warn("解析 replay stream 文件失败, streamKey={}", streamKey.streamId(), e);
|
return new ArrayList<>();
|
}
|
}
|
|
private Path resolveManifestPath(DeviceReplayStreamKey streamKey) {
|
String fileName = streamKey.getStationId() == null
|
? streamKey.getDeviceNo() + ".json"
|
: streamKey.getDeviceNo() + "_station_" + streamKey.getStationId() + ".json";
|
return Paths.get(loggingPath, streamKey.getDay(), MANIFEST_DIR_NAME, streamKey.getType(), fileName);
|
}
|
|
private String buildRelativeLogPath(DeviceReplayStreamKey streamKey, String fileName) {
|
return streamKey.getType() + "/" + streamKey.getDeviceNo() + "/" + fileName;
|
}
|
|
private long resolveSampleSeq(DeviceDataLog logItem, long fallback) {
|
if (logItem != null && logItem.getSampleSeq() != null && logItem.getSampleSeq() > 0) {
|
return logItem.getSampleSeq();
|
}
|
return fallback;
|
}
|
|
private long resolveSampleTimeMs(DeviceDataLog logItem) {
|
if (logItem != null && logItem.getSampleTimeMs() != null && logItem.getSampleTimeMs() > 0) {
|
return logItem.getSampleTimeMs();
|
}
|
if (logItem != null && logItem.getCreateTime() != null) {
|
return logItem.getCreateTime().getTime();
|
}
|
return 0L;
|
}
|
|
private boolean matchesStream(DeviceReplayStreamKey streamKey, DeviceDataLog logItem) {
|
if (streamKey == null || logItem == null) {
|
return false;
|
}
|
if (!streamKey.getType().equals(logItem.getType())) {
|
return false;
|
}
|
if (!streamKey.getDeviceNo().equals(logItem.getDeviceNo())) {
|
return false;
|
}
|
return equalsNullable(streamKey.getStationId(), logItem.getStationId());
|
}
|
|
private boolean equalsNullable(Integer left, Integer right) {
|
if (left == null) {
|
return right == null;
|
}
|
return left.equals(right);
|
}
|
|
private ParsedReplayFile parseReplayFile(String fileName) {
|
return parseReplayFile(fileName, null);
|
}
|
|
private ParsedReplayFile parseReplayFile(String fileName, Path filePath) {
|
Matcher devpMatcher = DEVP_FILE_PATTERN.matcher(fileName);
|
if (devpMatcher.matches()) {
|
ParsedReplayFile parsed = new ParsedReplayFile();
|
parsed.type = devpMatcher.group(1);
|
parsed.deviceNo = Integer.parseInt(devpMatcher.group(2));
|
parsed.stationId = Integer.parseInt(devpMatcher.group(3));
|
parsed.day = devpMatcher.group(4);
|
parsed.index = Integer.parseInt(devpMatcher.group(5));
|
parsed.file = filePath;
|
return parsed;
|
}
|
Matcher defaultMatcher = DEFAULT_FILE_PATTERN.matcher(fileName);
|
if (defaultMatcher.matches()) {
|
ParsedReplayFile parsed = new ParsedReplayFile();
|
parsed.type = defaultMatcher.group(1);
|
parsed.deviceNo = Integer.parseInt(defaultMatcher.group(2));
|
parsed.day = defaultMatcher.group(3);
|
parsed.index = Integer.parseInt(defaultMatcher.group(4));
|
parsed.file = filePath;
|
return parsed;
|
}
|
return null;
|
}
|
|
private static class ParsedReplayFile {
|
private String type;
|
private Integer deviceNo;
|
private Integer stationId;
|
private String day;
|
private Integer index;
|
private Path file;
|
}
|
|
private static class StreamRefreshState {
|
private final Object monitor = new Object();
|
private final AtomicInteger activeUsers = new AtomicInteger(0);
|
private volatile long lastScheduleTimeMs;
|
private volatile DeviceReplayStreamKey latestStreamKey;
|
private volatile ScheduledFuture<?> scheduledFuture;
|
}
|
}
|
