diff --git a/apm-collector/apm-collector-analysis/analysis-segment-parser/segment-parser-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/segment/parser/provider/buffer/SegmentBufferManager.java b/apm-collector/apm-collector-analysis/analysis-segment-parser/segment-parser-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/segment/parser/provider/buffer/SegmentBufferManager.java index d7add8e48..09fc97c1d 100644 --- a/apm-collector/apm-collector-analysis/analysis-segment-parser/segment-parser-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/segment/parser/provider/buffer/SegmentBufferManager.java +++ b/apm-collector/apm-collector-analysis/analysis-segment-parser/segment-parser-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/segment/parser/provider/buffer/SegmentBufferManager.java @@ -19,6 +19,7 @@ package org.apache.skywalking.apm.collector.analysis.segment.parser.provider.buffer; import java.io.*; +import java.nio.channels.FileLock; import org.apache.skywalking.apm.collector.core.module.ModuleManager; import org.apache.skywalking.apm.collector.core.util.*; import org.apache.skywalking.apm.network.proto.UpstreamSegment; @@ -40,11 +41,14 @@ public enum SegmentBufferManager { try { OffsetManager.INSTANCE.initialize(); if (new File(BufferFileConfig.BUFFER_PATH).mkdirs()) { + tryLock(); newDataFile(); } else if (BufferFileConfig.BUFFER_FILE_CLEAN_WHEN_RESTART) { deleteFiles(); + tryLock(); newDataFile(); } else { + tryLock(); String writeFileName = OffsetManager.INSTANCE.getWriteFileName(); if (StringUtils.isNotEmpty(writeFileName)) { File dataFile = new File(BufferFileConfig.BUFFER_PATH + writeFileName); @@ -77,6 +81,21 @@ public enum SegmentBufferManager { } } + private void tryLock() { + logger.info("try to lock buffer directory."); + FileLock lock = null; + + try { + lock = new FileOutputStream(BufferFileConfig.BUFFER_PATH + "lock").getChannel().tryLock(); + } catch (IOException e) { + logger.error(e.getMessage(), e); + } + + if (lock == null) { + throw new RuntimeException("The buffer directory is reading or writing by another thread."); + } + } + private void newDataFile() throws IOException { logger.debug("getOrCreate new segment buffer file"); String timeBucket = String.valueOf(TimeBucketUtils.INSTANCE.getSecondTimeBucket(System.currentTimeMillis()));