diff --git a/credit-tracker.properties b/credit-tracker.properties index 3f9b9fc..3eaf575 100644 --- a/credit-tracker.properties +++ b/credit-tracker.properties @@ -75,19 +75,29 @@ ct.encoding=windows-1251 #----------------------------Pipeline settings----------------------------# -#???-?? ????????? ?? xml ?? ???? ???????????? +#batch size for saving in phone_fid_map ct.saver-subject-batch-size=15000 -ct.pipeline.subject-batch-size=5000 -#????????? ? PHONE ? NAME + ?????? ? ???? SIGNALS_PHONE_FID_MAP +#limit on select phone_fid_map +ct.pipeline.subject-batch-size=10000 + +#STAGES SETTINGS +#select NAME + FID BY PHONE ct.pipeline.name-enricher-stage-concurrency=20 -#????????? ? ID ? ?????? ????? xml ??? ??. ??????? +ct.pipeline.name-enricher-queue-capacity=20 +#select ID BY FID 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 + +#Backoff settings ct.pipeline.error-attempt-count=5 -#????? ???????? ????? ????????? ??????? ct.pipeline.error-attempt-delay=2s -#??????? ?? ???? ??????? +ct.pipeline.retryable-exceptions=org.springframework.dao.DataAccessException,org.springframework.dao.DataIntegrityViolationException + +#Pipeline global settings ct.pipeline.global-timeout=2h -#???-?? ??????????? ???????????? ?????????????? ???????? -ct.pipeline.parallel-requests=3 -ct.pipeline.retryable-exceptions=org.springframework.dao.DataAccessException,org.springframework.dao.DataIntegrityViolationException \ No newline at end of file +ct.pipeline.parallel-requests=3 \ No newline at end of file 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 17dcbb8..63df0a7 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 @@ -6,9 +6,13 @@ import org.springframework.boot.context.properties.bind.DefaultValue; public record PipelineSettings( @DefaultValue("5000") Integer subjectBatchSize, - @DefaultValue("3") Integer saveSubjectStageConcurrency, @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, @DefaultValue("5m") Duration globalTimeout, diff --git a/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java b/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java index 79ad85f..45db916 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java +++ b/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java @@ -2,30 +2,26 @@ package ru.nbch.credit_tracker.config.task; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.model.impl.PartFilePayload; import ru.nbch.credit_tracker.model.impl.SubjectPayload; @Configuration public class StageConfig { - @Bean("saveSubjectQueueFactory") - public StageQueueFactory saveSubjectQueueFactory() { - return new StageQueueFactory<>(50); - } - @Bean("nameEnricherQueueFactory") - public StageQueueFactory nameEnricherQueueFactory() { - return new StageQueueFactory<>(10); + public StageQueueFactory nameEnricherQueueFactory(CreditTrackerSettings creditTrackerSettings) { + return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameEnricherQueueCapacity()); } @Bean("idEnricherQueueFactory") - public StageQueueFactory idEnricherQueueFactory() { - return new StageQueueFactory<>(10); + public StageQueueFactory idEnricherQueueFactory(CreditTrackerSettings creditTrackerSettings) { + return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameEnricherQueueCapacity()); } @Bean("partFileQueueFactory") - public StageQueueFactory partFileQueueFactory() { - return new StageQueueFactory<>(100); + public StageQueueFactory partFileQueueFactory(CreditTrackerSettings creditTrackerSettings) { + return new StageQueueFactory<>(creditTrackerSettings.getPipeline().mergerQueueCapacity()); } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java index 13678c1..0dd6e0e 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java @@ -9,6 +9,7 @@ import java.util.concurrent.TimeUnit; import java.util.function.BiFunction; import java.util.function.Supplier; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.stereotype.Component; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; @@ -30,7 +31,7 @@ import ru.nbch.credit_tracker.utils.error.ExceptionUtils; @Component @Slf4j -public class PipelinePackageManager { +public class PipelinePackageManager implements InitializingBean { private final SignalRestService signalRestService; private final FlagsEnricher flagsEnricherV2; private final StageQueueFactory idEnricherQueueFactory; @@ -48,11 +49,7 @@ public class PipelinePackageManager { return t; }); - private final ExecutorService subjectSaverExecutor = Executors.newFixedThreadPool(3, r -> { - Thread t = new Thread(r); - t.setName("subject-saver-%d".formatted(t.threadId())); - return t; - }); + private final ExecutorService subjectSaverExecutor; private final Semaphore semaphore; private final PipelineBackoff pipelineBackoff; private final Saver saver; @@ -80,9 +77,13 @@ public class PipelinePackageManager { this.fidNameEnricher = fidNameEnricher; this.creditTrackerSettings = creditTrackerSettings; this.semaphore = new Semaphore(creditTrackerSettings.getPipeline().parallelRequests()); - log.info("pipeline settings: {}", creditTrackerSettings.getPipeline()); this.pipelineBackoff = pipelineBackoff; this.saver = saver; + this.subjectSaverExecutor = Executors.newFixedThreadPool(creditTrackerSettings.getPipeline().parallelRequests(), r -> { + Thread t = new Thread(r); + t.setName("subject-saver-%d".formatted(t.threadId())); + return t; + }); } public void buildAndRun(Long packageId, Path pkgPath) throws Exception { @@ -147,4 +148,9 @@ public class PipelinePackageManager { throw e; } } + + @Override + public void afterPropertiesSet() throws Exception { + log.info("pipeline settings: {}", creditTrackerSettings.getPipeline()); + } } 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 a124648..334f75f 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 @@ -12,6 +12,7 @@ import java.util.concurrent.TimeUnit; import java.util.function.Consumer; 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.FlagDetail; import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper; import ru.nbch.credit_tracker.model.SubjectProfile; @@ -26,12 +27,12 @@ public class FlagsEnricher implements Consumer { private final ExecutorService ioPool; public FlagsEnricher(FlagsService2 flagsService, - PhoneFidMapMapper phoneFidMapMapper) { + CreditTrackerSettings creditTrackerSettings) { this.flagsService = flagsService; - this.ioPool = new ThreadPoolExecutor(10, + this.ioPool = new ThreadPoolExecutor(creditTrackerSettings.getPipeline().flagEnricherStageConcurrency(), 10, 0L, TimeUnit.MINUTES, - new LinkedBlockingQueue<>(100), + new LinkedBlockingQueue<>(creditTrackerSettings.getPipeline().flagEnricherQueueCapacity()), r -> { Thread t = new Thread(r); t.setName("flags-enricher-%d".formatted(t.threadId())); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/SaveSubjectHandler.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/SaveSubjectHandler.java deleted file mode 100644 index f579ca0..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/SaveSubjectHandler.java +++ /dev/null @@ -1,67 +0,0 @@ -package ru.nbch.credit_tracker.service.task.impl; - -import java.util.List; -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.PhoneFidMap; -import ru.nbch.credit_tracker.enums.Status; -import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; -import ru.nbch.credit_tracker.model.SubjectProfile; -import ru.nbch.credit_tracker.model.impl.SubjectPayload; -import ru.nbch.credit_tracker.service.indic.PhoneService2; -import ru.nbch.credit_tracker.service.task.Handler; - -/** - * Стадия для сопоставления FID к номеру телефона и регистрации пакета в базе данных. - * Так же выполняется поиск и обогащение перс.данными из NAME для дальнейшей обработки. - */ -@Service -@Slf4j -public class SaveSubjectHandler implements Handler { - private final PhoneService2 phoneService2; - private final UserPackageMapper userPackageMapper; - private final CreditTrackerSettings creditTrackerSettings; - - public SaveSubjectHandler(PhoneService2 phoneService2, - UserPackageMapper userPackageMapper, - CreditTrackerSettings creditTrackerSettings) { - this.phoneService2 = phoneService2; - this.userPackageMapper = userPackageMapper; - this.creditTrackerSettings = creditTrackerSettings; - } - - @Override - public SubjectPayload handle(SubjectPayload taskPayload) { - List subjects = taskPayload.getSubjects(); - Long packageId = taskPayload.getPackageId(); - log.debug("SaveSubjectStage handle by packageId: {}, size of bucket for save: {}", taskPayload.getPackageId(), subjects.size()); - List subjectForSave = subjects.stream().map(subject -> PhoneFidMap.builder() - .subjectId(subject.getIdentifier()) - .packageId(packageId) - .phone(subject.getPhone()) - .build()).toList(); - - List ids = phoneService2.registerSubjects(subjectForSave); - for (int i = 0; i < subjects.size(); i++) { - subjects.get(i).setMapId(ids.get(i)); - } - return taskPayload; - } - - @Override - public void handleOnPoison(SubjectPayload poison) { - log.trace("Got poison; set package on PROCESSING by packageId: {}", poison.getPackageId()); - userPackageMapper.updatePackageStatus(poison.getPackageId(), Status.PROCESSING.name()); - } - - @Override - public int preferredConcurrency() { - return creditTrackerSettings.getPipeline().saveSubjectStageConcurrency(); - } - - @Override - public SubjectPayload cookPoison() { - return SubjectPayload.POISON; - } -}