From 77f217e0f81c454df467c706ea084b5cf9842ab5 Mon Sep 17 00:00:00 2001 From: Mikhail Trofimov Date: Thu, 9 Jul 2026 16:51:21 +0300 Subject: [PATCH] Minor refactoring --- .../monitoring/KafkaMonitoringService.java | 187 ++++++++++-------- .../mode/monitoring/ReportNameRegistry.java | 8 +- 2 files changed, 117 insertions(+), 78 deletions(-) diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java index 16325b8..c041be3 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java @@ -198,10 +198,6 @@ public class KafkaMonitoringService { } for (PhonesMap phonesMap : phonesMapList) { - // флаги просрочек - Integer flagDebt5000 = phonesMap.getFlagDebt5000(); - Integer flag90days = phonesMap.getFlag90days(); - // в userPackagePhoneDataMap только нужные номера, те что не вошли в userPackagePhoneDataList игнорируются List userPackagePhoneDataList = userPackagePhoneDataMap.get(phonesMap.getPhone()); @@ -210,44 +206,30 @@ public class KafkaMonitoringService { } for (UserPackagePhoneData userPackagePhoneData : userPackagePhoneDataList) { - String subjectId = userPackagePhoneData.getSubjectId(); String signalTypes = userPackagePhoneData.getSignalTypes(); if (signalTypes.contains(Integer.toString(signalType))) { - SignalKafkaEvent event = new SignalKafkaEvent(); - event.setEventType(EventType.WRITE); - event.setNatural(true); - event.setUserPackagePhoneData(userPackagePhoneData); - event.setPackageMessage(message); - event.setSubjectId(subjectId); - event.setFlagDebt5000(flagDebt5000); - event.setFlag90days(flag90days); - event.setSignalType(Integer.toString(signalType)); - event.setBusCategory(signalType == 3 ? busCategory : null); - Integer segment = segmentMap.get(memberCode.substring(0, 6)); - event.setSegment(segment != null ? segment : -1); + SignalKafkaEvent event = buildSignalKafkaEvent( + true, message, + phonesMap, userPackagePhoneData, + null, null, // legal person fields + signalType, busCategory, segment + ); String reportKey = userPackagePhoneData.getUserId().substring(0, 6) + "_" + userPackagePhoneData.getPackageName(); - String reportFileName = resolveNameAndEnqueue(reportKey, event); + String reportFileName = reportNameRegistry.resolveNameAndEnqueue(reportKey, event); if (reportFileName == null) { continue; // событие не встало в очередь — hit не пишем } - Hit hit = Hit.builder() - .packageId(userPackagePhoneData.getPackageId()) - .subjectId(subjectId) - .signalType(signalType) - .reportFileName(reportFileName.replace(".xml", ".zip")) - .inqT(message.getInquiryType()) - .busCategory(busCategory) - .inqDt(message.getInquiryDate()) - .inqP(message.getInquiryPurpose()) - .inqA(message.getInquiryAmount()) - .segment(segment != null ? segment : -1) - .recordedAt(LocalDateTime.now()) - .phone(userPackagePhoneData.getPhone()) - .build(); + Hit hit = buildHit( + true, message, + userPackagePhoneData, + null, + signalType, busCategory, + segment, reportFileName + ); hits.add(hit); } } @@ -303,10 +285,6 @@ public class KafkaMonitoringService { } for (LegalsMap legalsMap : legalsMapList) { - // флаги просрочек - Integer flag90days = legalsMap.getFlag90days(); - LocalDateTime lastRejectedDate = legalsMap.getLastRejectedDate(); - // в userPackageLegalDataMap только нужные inn, regnum, те что не вошли в userPackageLegalDataList игнорируются List userPackageLegalDataList = userPackageLegalDataMap.get(Pair.of(legalsMap.getInn(), legalsMap.getRegNum())); @@ -315,45 +293,32 @@ public class KafkaMonitoringService { } for (UserPackageLegalData userPackageLegalData : userPackageLegalDataList) { - String subjectId = userPackageLegalData.getSubjectId(); String signalTypes = userPackageLegalData.getSignalTypes(); if (signalTypes.contains(Integer.toString(signalType))) { - SignalKafkaEvent event = new SignalKafkaEvent(); - event.setEventType(EventType.WRITE); - event.setNatural(false); - event.setUserPackageLegalData(userPackageLegalData); - event.setPackageMessage(message); - event.setSubjectId(subjectId); - event.setFlag90days(flag90days); - event.setSignalType(Integer.toString(signalType)); - event.setBusCategory(signalType == 3 ? busCategory : null); - event.setLastRejectedDate(lastRejectedDate); - Integer segment = segmentMap.get(memberCode.substring(0, 6)); - event.setSegment(segment != null ? segment : -1); + + SignalKafkaEvent event = buildSignalKafkaEvent( + false, message, + null, null, // np fields + legalsMap, + userPackageLegalData, + signalType, busCategory, segment + ); String reportKey = userPackageLegalData.getUserId().substring(0, 6) + "_" + userPackageLegalData.getPackageName(); - String reportFileName = resolveNameAndEnqueue(reportKey, event); + String reportFileName = reportNameRegistry.resolveNameAndEnqueue(reportKey, event); if (reportFileName == null) { continue; // событие не встало в очередь — hit не пишем } - Hit hit = Hit.builder() - .packageId(userPackageLegalData.getPackageId()) - .subjectId(subjectId) - .signalType(signalType) - .reportFileName(reportFileName.replace(".xml", ".zip")) - .inqT(message.getInquiryType()) - .busCategory(busCategory) - .inqDt(message.getInquiryDate()) - .inqP(message.getInquiryPurpose()) - .inqA(message.getInquiryAmount()) - .segment(segment != null ? segment : -1) - .recordedAt(LocalDateTime.now()) - .inn(userPackageLegalData.getInn()) - .regNum(userPackageLegalData.getRegNum()) - .build(); + Hit hit = buildHit( + false, message, + null, + userPackageLegalData, + signalType, busCategory, + segment, reportFileName + ); hits.add(hit); } } @@ -370,18 +335,86 @@ public class KafkaMonitoringService { (inquiryType == 3 && billingCode.equalsIgnoreCase("CHSP")); } - /** - * Резолвит имя результирующего файла и ставит событие в очередь. - * - * @return имя xml-файла или null, если поток был прерван и событие не встало в очередь - */ - private String resolveNameAndEnqueue(String reportKey, SignalKafkaEvent event) { - try { - return reportNameRegistry.resolveNameAndEnqueue(reportKey, event); - } catch (InterruptedException e) { - log.error("Error while putting signal event", e); - Thread.currentThread().interrupt(); - return null; + private SignalKafkaEvent buildSignalKafkaEvent(boolean isNatural, PackageMessage message, + PhonesMap phonesMap, UserPackagePhoneData userPackagePhoneData, + LegalsMap legalsMap, UserPackageLegalData userPackageLegalData, + int signalType, String busCategory, Integer segment) { + if (isNatural) { + // флаги просрочек + Integer flagDebt5000 = phonesMap.getFlagDebt5000(); + Integer flag90days = phonesMap.getFlag90days(); + + String subjectId = userPackagePhoneData.getSubjectId(); + + return SignalKafkaEvent.builder() + .eventType(EventType.WRITE) + .natural(true) + .userPackagePhoneData(userPackagePhoneData) + .packageMessage(message) + .subjectId(subjectId) + .flagDebt5000(flagDebt5000) + .flag90days(flag90days) + .signalType(Integer.toString(signalType)) + .busCategory(signalType == 3 ? busCategory : null) + .segment(segment != null ? segment : -1) + .build(); + } else { + // флаги просрочек + Integer flag90days = legalsMap.getFlag90days(); + LocalDateTime lastRejectedDate = legalsMap.getLastRejectedDate(); + + String subjectId = userPackageLegalData.getSubjectId(); + return SignalKafkaEvent.builder() + .eventType(EventType.WRITE) + .natural(false) + .userPackageLegalData(userPackageLegalData) + .packageMessage(message) + .subjectId(subjectId) + .flag90days(flag90days) + .signalType(Integer.toString(signalType)) + .busCategory(signalType == 3 ? busCategory : null) + .lastRejectedDate(lastRejectedDate) + .segment(segment != null ? segment : -1) + .build(); + } + } + + private Hit buildHit(boolean isNatural, PackageMessage message, + UserPackagePhoneData userPackagePhoneData, + UserPackageLegalData userPackageLegalData, + int signalType, String busCategory, + Integer segment, String reportFileName) { + if (isNatural) { + return Hit.builder() + .packageId(userPackagePhoneData.getPackageId()) + .subjectId(userPackagePhoneData.getSubjectId()) + .signalType(signalType) + .reportFileName(reportFileName.replace(".xml", ".zip")) + .inqT(message.getInquiryType()) + .busCategory(busCategory) + .inqDt(message.getInquiryDate()) + .inqP(message.getInquiryPurpose()) + .inqA(message.getInquiryAmount()) + .segment(segment != null ? segment : -1) + .recordedAt(LocalDateTime.now()) + .phone(userPackagePhoneData.getPhone()) + .build(); + } else { + return Hit.builder() + .packageId(userPackageLegalData.getPackageId()) + .subjectId(userPackageLegalData.getSubjectId()) + .signalType(signalType) + .reportFileName(reportFileName.replace(".xml", ".zip")) + .inqT(message.getInquiryType()) + .busCategory(busCategory) + .inqDt(message.getInquiryDate()) + .inqP(message.getInquiryPurpose()) + .inqA(message.getInquiryAmount()) + .segment(segment != null ? segment : -1) + .recordedAt(LocalDateTime.now()) + .inn(userPackageLegalData.getInn()) + .regNum(userPackageLegalData.getRegNum()) + .build(); } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/ReportNameRegistry.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/ReportNameRegistry.java index f92b752..9d23d47 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/ReportNameRegistry.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/ReportNameRegistry.java @@ -1,5 +1,6 @@ package ru.nbch.credit_tracker.service.mode.monitoring; +import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.stereotype.Component; import ru.nbch.credit_tracker.enums.EventType; @@ -21,6 +22,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock; * закрывающий эпоху N. Поэтому WRITE-события кладутся в очередь только через * {@link #resolveNameAndEnqueue}, а ROTATE — только через {@link #enqueueRotate}. */ +@Slf4j @Component public class ReportNameRegistry { @@ -40,7 +42,7 @@ public class ReportNameRegistry { * @param key membercode_packageName * @return имя xml-файла, в который попадёт сигнал */ - public String resolveNameAndEnqueue(String key, SignalKafkaEvent event) throws InterruptedException { + public String resolveNameAndEnqueue(String key, SignalKafkaEvent event) { lock.readLock().lock(); try { String name = names.computeIfAbsent(key, k -> @@ -48,6 +50,10 @@ public class ReportNameRegistry { LocalDateTime.now().format(CTUtil.responseReportDateTimeFormatter))); signalEventsQueue.put(event); return name; + } catch (InterruptedException e) { + log.error("Error while putting signal event", e); + Thread.currentThread().interrupt(); + return null; } finally { lock.readLock().unlock(); }