From 9fce65ea4f3308e21f1cfee9045abbcdaff0bce8 Mon Sep 17 00:00:00 2001 From: ialbert Date: Fri, 16 Jan 2026 17:47:41 +0300 Subject: [PATCH] update flags --- .gitignore | 1 + credit-tracker.properties | 7 ++- .../settings/CreditTrackerSettings.java | 2 + .../config/settings/FlagUpdaterSettings.java | 9 +++ .../config/settings/PipelineSettings.java | 4 +- .../mapper/signal/PhonesMapMapper.java | 2 + .../service/UpdatingFlagsService.java | 63 ++++++------------- .../service/task/PipelinePackageFabric.java | 8 +-- .../service/task/impl/FlagsEnricher.java | 62 +++++++----------- .../mybatis/mapper/signal/PhonesMapMapper.xml | 10 +++ 10 files changed, 77 insertions(+), 91 deletions(-) create mode 100644 src/main/java/ru/nbch/credit_tracker/config/settings/FlagUpdaterSettings.java diff --git a/.gitignore b/.gitignore index 667aaef..97b6019 100644 --- a/.gitignore +++ b/.gitignore @@ -31,3 +31,4 @@ build/ ### VS Code ### .vscode/ +tmp/ \ No newline at end of file diff --git a/credit-tracker.properties b/credit-tracker.properties index 857724b..4547704 100644 --- a/credit-tracker.properties +++ b/credit-tracker.properties @@ -90,9 +90,10 @@ ct.pipeline.id-enricher-stage-concurrency=10 ct.pipeline.id-enricher-queue-capacity=20 #Save part of signal xml file and merge; only one thread ct.pipeline.merger-queue-capacity=50 -#select and calc flags; instant run after name-enricher -ct.pipeline.flag-enricher-stage-concurrency=10 -ct.pipeline.flag-enricher-queue-capacity=50 + +#select and calc flags +ct.flag-updater.concurrency=10 +ct.flag-updater.queue-capacity=100 #Backoff settings ct.pipeline.error-attempt-count=5 diff --git a/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java b/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java index 5dcd2f8..e107c85 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java +++ b/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java @@ -23,6 +23,8 @@ public class CreditTrackerSettings { private XsdValidation xsdValidation; @NestedConfigurationProperty private PipelineSettings pipeline; + @NestedConfigurationProperty + private FlagUpdaterSettings flagUpdater; private String signalsRootUri; private String signalsToken; diff --git a/src/main/java/ru/nbch/credit_tracker/config/settings/FlagUpdaterSettings.java b/src/main/java/ru/nbch/credit_tracker/config/settings/FlagUpdaterSettings.java new file mode 100644 index 0000000..5dbd34d --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/config/settings/FlagUpdaterSettings.java @@ -0,0 +1,9 @@ +package ru.nbch.credit_tracker.config.settings; + +import org.springframework.boot.context.properties.bind.DefaultValue; + +public record FlagUpdaterSettings( + @DefaultValue("10") Integer concurrency, + @DefaultValue("100") Integer queueCapacity +) { +} diff --git a/src/main/java/ru/nbch/credit_tracker/config/settings/PipelineSettings.java b/src/main/java/ru/nbch/credit_tracker/config/settings/PipelineSettings.java index 63df0a7..95a1bd5 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/settings/PipelineSettings.java +++ b/src/main/java/ru/nbch/credit_tracker/config/settings/PipelineSettings.java @@ -4,14 +4,14 @@ import java.time.Duration; import java.util.List; import org.springframework.boot.context.properties.bind.DefaultValue; +//X это количество потоков Save +//тогда: Name должно быть ~3x, Fid должно быть ~2x чтобы они поспевали друг за другом? public record PipelineSettings( @DefaultValue("5000") Integer subjectBatchSize, @DefaultValue("15") Integer nameEnricherStageConcurrency, @DefaultValue("20") Integer nameEnricherQueueCapacity, @DefaultValue("10") Integer idEnricherStageConcurrency, @DefaultValue("15") Integer idEnricherQueueCapacity, - @DefaultValue("10") Integer flagEnricherStageConcurrency, - @DefaultValue("100") Integer flagEnricherQueueCapacity, @DefaultValue("50") Integer mergerQueueCapacity, @DefaultValue("5") Integer errorAttemptCount, @DefaultValue("2s") Duration errorAttemptDelay, diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java index 5b19ff9..338fb7a 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java @@ -10,6 +10,8 @@ public interface PhonesMapMapper { List getCachedData(@Param("phones") List phones); List searchedAtLessThan(@Param("daysCount") Integer daysCount, @Param("limit") int limit, @Param("offset") int offset); PhonesMap findByPhone(@Param("phone") Long phone); + List findEligibleForFlagsUpdate(@Param("lastId") Long lastId, + @Param("batchSize") Integer batchSize); void updateIdCachedData(@Param("json") String json); void updateNameCachedData(@Param("json") String json); void updateFlagCachedData(@Param("json") String json); diff --git a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java index ca5485a..0615c86 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java @@ -2,21 +2,12 @@ package ru.nbch.credit_tracker.service; import java.util.List; import lombok.extern.slf4j.Slf4j; -import org.slf4j.MDC; import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; -import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; -import ru.nbch.credit_tracker.entities.app.UserPackage; -import ru.nbch.credit_tracker.enums.Status; +import ru.nbch.credit_tracker.entities.app.PhonesMap; import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper; -import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; -import ru.nbch.credit_tracker.service.task.Pipeline; -import ru.nbch.credit_tracker.service.task.PipelinePackageFabric; -import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator; -import ru.nbch.credit_tracker.service.task.PipelineRecalcPDOrchestrator; -import ru.nbch.credit_tracker.service.task.SetupParam; -import ru.nbch.credit_tracker.service.task.impl.PipelineMode; +import ru.nbch.credit_tracker.service.task.impl.FlagsEnricher; /** * Сервис периодического обновления фидов и флагов просрочек по пакетам находящимся в статусе ON_MONITORING @@ -26,47 +17,31 @@ import ru.nbch.credit_tracker.service.task.impl.PipelineMode; @Service @EnableScheduling public class UpdatingFlagsService { - private final Long daysToUpdateFlags; - private final UserPackageMapper userPackageMapper; + private final static int BATCH_SIZE = 1000; private final PhonesMapMapper phonesMapMapper; - private final PipelinePackageFabric pipelinePackageManager; - private final PipelinePackageOrchestrator pipelineOrchestrator; - private final PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator; + private final FlagsEnricher flagsEnricherV2; - public UpdatingFlagsService(CreditTrackerSettings config, - UserPackageMapper userPackageMapper, - PhonesMapMapper phonesMapMapper, - PipelinePackageFabric pipelinePackageManager, - PipelinePackageOrchestrator pipelineOrchestrator, - PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator) { - this.daysToUpdateFlags = config.getDaysToUpdateFlags() != null ? config.getDaysToUpdateFlags() : 1; - this.userPackageMapper = userPackageMapper; + public UpdatingFlagsService(PhonesMapMapper phonesMapMapper, FlagsEnricher flagsEnricherV2) { this.phonesMapMapper = phonesMapMapper; - this.pipelinePackageManager = pipelinePackageManager; - this.pipelineOrchestrator = pipelineOrchestrator; - this.pipelineRecalcPDOrchestrator = pipelineRecalcPDOrchestrator; + this.flagsEnricherV2 = flagsEnricherV2; } @Scheduled(cron = "${ct.update-flags-cron-expression}") private void update() { - log.info("Start updating flags"); - List packages = userPackageMapper.findAllPackagesByStatus(Status.ON_MONITORING.name()); - for (UserPackage userPackage : packages) { - MDC.put("ru.nbch.credit_tracker.log.mdc.key", "[" + userPackage.getPackageName() + "]"); - log.info("Check on updating flags for Package {}", userPackage.getPackageName()); - boolean hasOldRow = userPackageMapper.hasOldRow(userPackage.getId(), daysToUpdateFlags.intValue()); - log.info("There are {} old rows", hasOldRow ? "exist" : "doesn't exist"); - if (hasOldRow) { - try { - userPackageMapper.updatePackageStatus(userPackage.getId(), Status.PROCESSING_REPEATED.name()); - Pipeline pipeline = pipelinePackageManager.build(new SetupParam(userPackage.getId(), null, PipelineMode.DEFAULT)); - pipelineOrchestrator.startOrDelay(pipeline, userPackage.getId()); - } catch (Exception e) { - log.error("Failed to build pipeline", e); - } + log.info("Start filling flags"); + long lastId = 0L; + int batchNumber = 1; + List phonesMaps; + while (true) { + phonesMaps = phonesMapMapper.findEligibleForFlagsUpdate(lastId, BATCH_SIZE); + log.trace("batch {}, signals_phones_map records loaded: {}", batchNumber, phonesMaps.size()); + if (phonesMaps.isEmpty()) { + break; } - MDC.remove("ru.nbch.credit_tracker.log.mdc.key"); + + flagsEnricherV2.accept(phonesMaps); + lastId = phonesMaps.getLast().getId(); + log.trace("batch {}, lastId {}", batchNumber++, lastId); } - log.info("Finish updating flags"); } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java index 53b3154..9abc35f 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java @@ -93,11 +93,11 @@ public class PipelinePackageFabric implements InitializingBean { public Pipeline buildRecalculationPD(SetupParam setupParam) throws Exception { return PipeLineBuilderV2.first(nameUpdaterQueueFactory.getObject(), idUpdaterQueueFactory.getObject()) - .handler(fidNameUpdater) - .andThen() + .handler(fidNameUpdater) + .andThen() .stage(noOpQueueFactory.getObject()) - .handler(fidIdUpdater) - .andThen() + .handler(fidIdUpdater) + .andThen() .fillInitialQueue(pipelineStarterFactory.apply(setupParam)) .pollFinishingQueue(noOpTerminatorFactory.get()) .backoff(pipelineBackoff) diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java index de1daea..799894e 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java @@ -7,20 +7,18 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; -import java.util.function.Consumer; import java.util.stream.Stream; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.entities.app.PhonesMap; import ru.nbch.credit_tracker.model.FlagData; -import ru.nbch.credit_tracker.model.SubjectProfile; -import ru.nbch.credit_tracker.model.impl.SubjectPayload; import ru.nbch.credit_tracker.service.indic.FlagsService2; import ru.nbch.credit_tracker.service.signal.PackageRecordsService; @Slf4j @Service -public class FlagsEnricher implements Consumer { +public class FlagsEnricher { private final FlagsService2 flagsService; private final PackageRecordsService packageRecordsService; @@ -31,10 +29,10 @@ public class FlagsEnricher implements Consumer { CreditTrackerSettings creditTrackerSettings) { this.flagsService = flagsService; this.packageRecordsService = packageRecordsService; - this.ioPool = new ThreadPoolExecutor(creditTrackerSettings.getPipeline().flagEnricherStageConcurrency(), - 10, + this.ioPool = new ThreadPoolExecutor(creditTrackerSettings.getFlagUpdater().concurrency(), + creditTrackerSettings.getFlagUpdater().concurrency(), 0L, TimeUnit.MINUTES, - new LinkedBlockingQueue<>(creditTrackerSettings.getPipeline().flagEnricherQueueCapacity()), + new LinkedBlockingQueue<>(creditTrackerSettings.getFlagUpdater().queueCapacity()), r -> { Thread t = new Thread(r); t.setName("flags-enricher-%d".formatted(t.threadId())); @@ -44,17 +42,11 @@ public class FlagsEnricher implements Consumer { ); } - public void process(SubjectPayload subjectPayload) { - process(subjectPayload, false); - } + public void process(List phonesMaps) { + log.debug("FlagsEnricher handle: size: {}", phonesMaps.size()); - public void process(SubjectPayload subjectPayload, boolean update) { - List subjects = subjectPayload.getSubjects(); - log.debug("FlagsEnricher handle by packageId: {}, size: {}", subjectPayload.getPackageId(), subjects.size()); - - List noFlagDataFids = subjects.stream() - .filter(subjectProfile -> subjectProfile.getFlagData() == null) - .map(SubjectProfile::getFid) + List noFlagDataFids = phonesMaps.stream() + .map(PhonesMap::getFid) .toList(); LocalDate yearAgo = LocalDate.now().minusYears(1); @@ -63,34 +55,28 @@ public class FlagsEnricher implements Consumer { Map flag90ByPastDueArrear = flagsService.findFlag90ByPastDueArrear(noFlagDataFids, yearAgo); Map flag90ByRuPaymt = flagsService.findFlag90ByRuPaymt(noFlagDataFids, yearAgo); - List flagCacheOnUpdate = subjects.stream() - .filter(subjectProfile -> subjectProfile.getFid() != null) - .flatMap(subject -> { - if (subject.getFlagData() == null) { - Integer debt5000 = flagsDebt5000.getOrDefault(subject.getFid(), 0); - int pastDue90 = flag90ByPastDueArrear.getOrDefault(subject.getFid(), 0); - int paymt90 = flag90ByRuPaymt.getOrDefault(subject.getFid(), 0); - Integer flagDays90 = Math.max(pastDue90, paymt90); + List flagCacheOnUpdate = phonesMaps.stream() + .flatMap(phonesMap -> { + Integer fid = phonesMap.getFid(); + Integer debt5000 = flagsDebt5000.getOrDefault(fid, 0); + int pastDue90 = flag90ByPastDueArrear.getOrDefault(fid, 0); + int paymt90 = flag90ByRuPaymt.getOrDefault(fid, 0); + Integer flagDays90 = Math.max(pastDue90, paymt90); - FlagData flagData = FlagData.builder() - .flag_90_12(flagDays90) - .flag_debt5000(debt5000) - .phone(subject.getPhone()) - .build(); - subject.setFlagData(flagData); - return Stream.of(flagData); - } - return Stream.empty(); + FlagData flagData = FlagData.builder() + .flag_90_12(flagDays90) + .flag_debt5000(debt5000) + .phone(phonesMap.getPhone()) + .build(); + return Stream.of(flagData); }).toList(); - packageRecordsService.updateFlagCache(flagCacheOnUpdate); } - @Override - public void accept(SubjectPayload subjectPayload) { + public void accept(List phonesMaps) { ThreadPoolExecutor casted = (ThreadPoolExecutor) ioPool; log.info("submitting payload. active tasks number: {}; queue size: {}", casted.getActiveCount(), casted.getQueue().size()); - this.ioPool.submit(() -> process(subjectPayload)); + this.ioPool.submit(() -> process(phonesMaps)); } } diff --git a/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml b/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml index 0ea836a..e4beb3a 100644 --- a/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml @@ -116,6 +116,16 @@ birth_dt = EXCLUDED.birth_dt + + update signals_phones_map dst set ser_num = src.ser_num,