treshenkov/clearing
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

http://jira.mfd.msk:8088/browse/CLS-36 parallel file execution temp fix

Browse source
This commit is contained in:
ialbert 2022-12-12 18:54:33 +03:00
parent 7ddad9bd21
commit 3bc6c83b1c
1 changed files with 45 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

63
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/DBFImporterService.java
View file

@ -3,7 +3,6 @@ package ru.spcex.clearing.dbf.importer.services;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled; import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
@ -12,9 +11,8 @@ import ru.spcex.clearing.dbf.importer.logic.data.ResultContainer;
import ru.spcex.clearing.dbf.importer.logic.data.enums.ETable; import ru.spcex.clearing.dbf.importer.logic.data.enums.ETable;
import java.io.File; import java.io.File;
import java.util.LinkedList; import java.nio.file.Path;
import java.util.List; import java.util.*;
import java.util.Map;
import java.util.function.Function; import java.util.function.Function;
import java.util.stream.Collectors; import java.util.stream.Collectors;
import java.util.stream.Stream; import java.util.stream.Stream;
@ -26,6 +24,7 @@ public class DBFImporterService {
private final FileChecker fileChecker; private final FileChecker fileChecker;
private final ThreadPoolTaskExecutor executorService; private final ThreadPoolTaskExecutor executorService;
private final Processor processor; private final Processor processor;
private final Set<Path> filesCurrentlyInProcess;
public DBFImporterService(@Qualifier("fileChecker") FileChecker messageListener, public DBFImporterService(@Qualifier("fileChecker") FileChecker messageListener,
@Qualifier("executor") ThreadPoolTaskExecutor executorService, @Qualifier("executor") ThreadPoolTaskExecutor executorService,
@ -33,6 +32,7 @@ public class DBFImporterService {
this.fileChecker = messageListener; this.fileChecker = messageListener;
this.executorService = executorService; this.executorService = executorService;
this.processor = processor; this.processor = processor;
this.filesCurrentlyInProcess = new HashSet<>();
} }
@Scheduled(cron = "${import-dbf-service.scheduler.check-src-dir-cron}") @Scheduled(cron = "${import-dbf-service.scheduler.check-src-dir-cron}")
@ -43,25 +43,52 @@ public class DBFImporterService {
public void run(ETable specificTable) { public void run(ETable specificTable) {
log.info("adding import task {}", specificTable != null ? specificTable.name() : ""); log.info("adding import task {}", specificTable != null ? specificTable.name() : "");
executorService.execute(() -> { executorService.execute(() -> {
log.info("checking new files... {}", specificTable != null ? specificTable.name() : ""); Map<ETable, List<File>> newFiles = null;
Map<ETable, List<File>> newFiles = fileChecker.checkNewFiles(specificTable); try {
if (newFiles.size() > 0 && log.isDebugEnabled()) { log.info("checking new files... {}", specificTable != null ? specificTable.name() : "");
log.debug("following files will be processed {}", forLogging(newFiles)); newFiles = getFiles(specificTable);
} else if (newFiles.size() > 0) { if (newFiles.size() > 0 && log.isDebugEnabled()) {
log.info("following files will be processed {}", forLoggingSizeOnly(newFiles)); log.debug("following files will be processed {}", forLogging(newFiles));
} else { } else if (newFiles.size() > 0) {
log.info("no files were found"); log.info("following files will be processed {}", forLoggingSizeOnly(newFiles));
} } else {
for (Map.Entry<ETable, List<File>> newFilesEntry : newFiles.entrySet()) { log.info("no files were found");
ETable currTable = newFilesEntry.getKey(); }
List<File> fileList = newFilesEntry.getValue(); for (Map.Entry<ETable, List<File>> newFilesEntry : newFiles.entrySet()) {
for (File dbfFile : fileList) { ETable currTable = newFilesEntry.getKey();
processor.process(ResultContainer.createNewTask(currTable, dbfFile)); List<File> fileList = newFilesEntry.getValue();
for (File dbfFile : fileList) {
processor.process(ResultContainer.createNewTask(currTable, dbfFile));
}
}
} finally {
if (newFiles != null && newFiles.size() > 0) {
cleanFiles(newFiles);
} }
} }
}); });
} }
private synchronized Map<ETable, List<File>> getFiles(ETable specificTable) {
Map<ETable, List<File>> newFiles = fileChecker.checkNewFiles(specificTable);
Map<ETable, List<File>> newFilesFiltered = newFiles.entrySet()
.stream()
.map(entry ->
new AbstractMap.SimpleEntry<>(entry.getKey(), entry.getValue()
.stream()
.filter(file -> !filesCurrentlyInProcess.contains(file.toPath()))
.collect(Collectors.toList())))
.collect(Collectors.toMap(AbstractMap.SimpleEntry::getKey, AbstractMap.SimpleEntry::getValue));
newFilesFiltered.forEach((table, files)
-> files.forEach(file -> filesCurrentlyInProcess.add(file.toPath())));
return newFilesFiltered;
}
private synchronized void cleanFiles(Map<ETable, List<File>> filesFromTask) {
filesFromTask.forEach((table, files)
-> files.forEach(file -> filesCurrentlyInProcess.remove(file.toPath())));
}
private static String forLogging(Map<ETable, List<File>> files) { private static String forLogging(Map<ETable, List<File>> files) {
return files return files
.entrySet() .entrySet()