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

Added ReportNameRegistry to store generated report names for hits writing

Browse source
This commit is contained in:
Mikhail Trofimov 2026-07-05 14:15:41 +03:00
parent 2d75350f75
commit 0cfe976670
5 changed files with 132 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
src/main/java/ru/nbch/credit_tracker/model/SignalKafkaEvent.java
View file

@ -5,6 +5,7 @@ import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage;
import ru.nbch.credit_tracker.enums.EventType;
import java.time.LocalDateTime;
import java.util.Map;
@Getter
@Setter
@ -13,6 +14,7 @@ import java.time.LocalDateTime;
@AllArgsConstructor
public class SignalKafkaEvent {
private EventType eventType;
private Map<String, String> reportNames;
private UserPackagePhoneData userPackagePhoneData;
private UserPackageLegalData userPackageLegalData;

53
src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java
View file

@ -49,17 +49,20 @@ public class KafkaMonitoringService {
private final SourceService sourceService;
private final SegmentMapService segmentMapService;
private final HitsService hitsService;
private final ReportNameRegistry reportNameRegistry;
public KafkaMonitoringService(@Qualifier("signalKafkaEventQueue") BlockingQueue<SignalKafkaEvent> signalEventsQueue,
PackageRecordsService packageRecordsService,
SourceService sourceService,
SegmentMapService segmentMapService,
CreditTrackerSettings settings,
HitsService hitsService) {
HitsService hitsService,
ReportNameRegistry reportNameRegistry) {
this.signalEventsQueue = signalEventsQueue;
this.packageRecordsService = packageRecordsService;
this.sourceService = sourceService;
this.segmentMapService = segmentMapService;
this.reportNameRegistry = reportNameRegistry;
this.permits = Integer.parseInt(settings.getKafkaConsumerSettings().concurency()); // количество равно concurency листенера
this.hitsService = hitsService;
this.processingPermits = new Semaphore(permits);
@ -133,14 +136,14 @@ public class KafkaMonitoringService {
});
combinedFuture.thenAcceptAsync(hits -> {
if (!hits.isEmpty()) {
try {
hitsService.insertHits(hits);
} catch (Exception e) {
log.error("Failed to save hits. {}", ExceptionUtils.getStackTrace(e));
}
}
}, dbExecutor);
if (!hits.isEmpty()) {
try {
hitsService.insertHits(hits);
} catch (Exception e) {
log.error("Failed to save hits. {}", ExceptionUtils.getStackTrace(e));
}
}
}, dbExecutor);
try {
combinedFuture.get(1, TimeUnit.MINUTES);
@ -225,12 +228,17 @@ public class KafkaMonitoringService {
Integer segment = segmentMap.get(memberCode.substring(0, 6));
event.setSegment(segment != null ? segment : -1);
putToQueue(event);
String reportKey = userPackagePhoneData.getUserId().substring(0, 6)
+ "_" + userPackagePhoneData.getPackageName();
String reportFileName = resolveNameAndEnqueue(reportKey, event);
if (reportFileName == null) {
continue; // событие не встало в очередь — hit не пишем
}
Hit hit = Hit.builder()
.packageId(userPackagePhoneData.getPackageId())
.subjectId(subjectId)
.signalType(signalType)
.reportFileName("") // TODO
.reportFileName(reportFileName.replace(".xml", ".zip"))
.inqT(message.getInquiryType())
.busCategory(busCategory)
.inqDt(message.getInquiryDate())
@ -325,12 +333,17 @@ public class KafkaMonitoringService {
Integer segment = segmentMap.get(memberCode.substring(0, 6));
event.setSegment(segment != null ? segment : -1);
putToQueue(event);
String reportKey = userPackageLegalData.getUserId().substring(0, 6)
+ "_" + userPackageLegalData.getPackageName();
String reportFileName = resolveNameAndEnqueue(reportKey, event);
if (reportFileName == null) {
continue; // событие не встало в очередь — hit не пишем
}
Hit hit = Hit.builder()
.packageId(userPackageLegalData.getPackageId())
.subjectId(subjectId)
.signalType(signalType)
.reportFileName("") // TODO
.reportFileName(reportFileName.replace(".xml", ".zip"))
.inqT(message.getInquiryType())
.busCategory(busCategory)
.inqDt(message.getInquiryDate())
@ -357,12 +370,18 @@ public class KafkaMonitoringService {
(inquiryType == 3 && billingCode.equalsIgnoreCase("CHSP"));
}
private void putToQueue(SignalKafkaEvent event) {
/**
* Резолвит имя результирующего файла и ставит событие в очередь.
*
* @return имя xml-файла или null, если поток был прерван и событие не встало в очередь
*/
private String resolveNameAndEnqueue(String reportKey, SignalKafkaEvent event) {
try {
signalEventsQueue.put(event);
return reportNameRegistry.resolveNameAndEnqueue(reportKey, event);
} catch (InterruptedException e) {
log.error("Error while putting signal event", e);
Thread.currentThread().interrupt();
return null;
}
}
@ -374,9 +393,7 @@ public class KafkaMonitoringService {
processingPermits.acquire(permits);
try {
SignalKafkaEvent rotateEvent = new SignalKafkaEvent();
rotateEvent.setEventType(EventType.ROTATE);
signalEventsQueue.put(rotateEvent);
reportNameRegistry.enqueueRotate();
log.info("ROTATE event sent to queue");
SignalKafkaEvent stopEvent = new SignalKafkaEvent();

72
src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/ReportNameRegistry.java Normal file
View file

@ -0,0 +1,72 @@
package ru.nbch.credit_tracker.service.mode.monitoring;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;
import ru.nbch.credit_tracker.enums.EventType;
import ru.nbch.credit_tracker.model.SignalKafkaEvent;
import ru.nbch.credit_tracker.utils.CTUtil;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
/**
* Реестр имён результирующих файлов текущего цикла ротации.
* Имя фиксируется при первом сигнале цикла для пакета (timestamp — момент первого сигнала).
* <p>
* событие, получившее имя эпохи N, встаёт в очередь раньше, чем ROTATE,
* закрывающий эпоху N. Поэтому WRITE-события кладутся в очередь только через
* {@link #resolveNameAndEnqueue}, а ROTATE — только через {@link #enqueueRotate}.
*/
@Component
public class ReportNameRegistry {
// ключ: membercode_packageName; значение: финальное имя xml-файла
private final ConcurrentHashMap<String, String> names = new ConcurrentHashMap<>();
private final ReadWriteLock lock = new ReentrantReadWriteLock();
private final BlockingQueue<SignalKafkaEvent> signalEventsQueue;
public ReportNameRegistry(@Qualifier("signalKafkaEventQueue") BlockingQueue<SignalKafkaEvent> signalEventsQueue) {
this.signalEventsQueue = signalEventsQueue;
}
/**
* Резолвит имя файла для пакета и атомарно с этим ставит WRITE-событие в очередь.
*
* @param key membercode_packageName
* @return имя xml-файла, в который попадёт сигнал
*/
public String resolveNameAndEnqueue(String key, SignalKafkaEvent event) throws InterruptedException {
lock.readLock().lock();
try {
String name = names.computeIfAbsent(key, k ->
String.format("%s_%s.xml", k,
LocalDateTime.now().format(CTUtil.responseReportDateTimeFormatter)));
signalEventsQueue.put(event);
return name;
} finally {
lock.readLock().unlock();
}
}
/**
* Закрывает текущую эпоху: снапшот имён уезжает внутри ROTATE-события,
* реестр очищается атомарно с постановкой события в очередь.
*/
public void enqueueRotate() throws InterruptedException {
lock.writeLock().lock();
try {
SignalKafkaEvent rotateEvent = new SignalKafkaEvent();
rotateEvent.setEventType(EventType.ROTATE);
rotateEvent.setReportNames(new HashMap<>(names));
names.clear();
signalEventsQueue.put(rotateEvent);
} finally {
lock.writeLock().unlock();
}
}
}

14
src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/RotateSheduler.java
View file

@ -3,14 +3,10 @@ package ru.nbch.credit_tracker.service.mode.monitoring;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.stereotype.Component;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import ru.nbch.credit_tracker.enums.EventType;
import ru.nbch.credit_tracker.model.SignalKafkaEvent;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@ -23,12 +19,12 @@ public class RotateSheduler {
);
private volatile boolean isShuttingdDown = false;
private final BlockingQueue<SignalKafkaEvent> signalEventsQueue;
private final ReportNameRegistry reportNameRegistry;
private final CreditTrackerSettings config;
public RotateSheduler(@Qualifier("signalKafkaEventQueue") BlockingQueue<SignalKafkaEvent> signalEventsQueue,
public RotateSheduler(ReportNameRegistry reportNameRegistry,
CreditTrackerSettings config) {
this.signalEventsQueue = signalEventsQueue;
this.reportNameRegistry = reportNameRegistry;
this.config = config;
}
@ -44,9 +40,7 @@ public class RotateSheduler {
return;
}
try {
SignalKafkaEvent rotateEvent = new SignalKafkaEvent();
rotateEvent.setEventType(EventType.ROTATE);
signalEventsQueue.put(rotateEvent);
reportNameRegistry.enqueueRotate();
} catch (InterruptedException e) {
log.error("Error while putting signal event", e);
Thread.currentThread().interrupt();

29
src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java
View file

@ -34,6 +34,7 @@ import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@ -85,7 +86,7 @@ public class SignalKafkaEventWriterService {
}
if (event.getEventType() == EventType.ROTATE) {
rotate();
rotate(event.getReportNames());
}
}
}
@ -134,14 +135,16 @@ public class SignalKafkaEventWriterService {
}
/**
* Упаковывает все результирующие xml в архив
* Упаковывает все результирующие xml в архив.
*
* @param reportNames снапшот имён файлов закрываемой эпохи (ключ: membercode_packageName)
*/
private void rotate() {
private void rotate(Map<String, String> reportNames) {
List<UserPackage> packagesToUpdate = new ArrayList<>();
List<UserPackage> userPackages = userPackageMapper.findAllPackagesByStatus(Status.ON_MONITORING.name());
for (UserPackage userPackage : userPackages) {
try {
String reportName = movedAndZip(userPackage);
String reportName = movedAndZip(userPackage, reportNames);
userPackage.setLastDownloadedReport(reportName);
packagesToUpdate.add(userPackage);
} catch (IOException e) {
@ -167,7 +170,7 @@ public class SignalKafkaEventWriterService {
}
}
private String movedAndZip(UserPackage userPackage) throws IOException {
private String movedAndZip(UserPackage userPackage, Map<String, String> reportNames) throws IOException {
String finalName = "";
String commonMembercode = userPackage.getUserId().substring(0, 6);
Path tmpFilePath = Path.of(config.getLoadDir(),
@ -184,11 +187,17 @@ public class SignalKafkaEventWriterService {
if (!Files.exists(rootReportPath)) {
Files.createDirectories(rootReportPath);
}
LocalDateTime newReportDate = LocalDateTime.now();
finalName = String.format("%s_%s_%s.xml",
commonMembercode,
userPackage.getPackageName(),
newReportDate.format(CTUtil.responseReportDateTimeFormatter));
String reportKey = commonMembercode + "_" + userPackage.getPackageName();
finalName = reportNames != null ? reportNames.get(reportKey) : null;
if (finalName == null) {
// tmp-файл есть, а имени в реестре нет —
// генерируем имя по времени ротации
finalName = String.format("%s_%s_%s.xml",
commonMembercode,
userPackage.getPackageName(),
LocalDateTime.now().format(CTUtil.responseReportDateTimeFormatter));
}
Path responsePath = rootReportPath.resolve(finalName);