diff --git a/credit-tracker.properties b/credit-tracker.properties index 4365883..7798bed 100644 --- a/credit-tracker.properties +++ b/credit-tracker.properties @@ -93,6 +93,15 @@ ct.pipeline.id-enricher-queue-capacity=20 #Save part of signal xml file and merge; only one thread ct.pipeline.merger-queue-capacity=50 +#STAGES SETTINGS +#select NAME + FID BY PHONE +ct.pipeline.name-updater-stage-concurrency=20 +ct.pipeline.name-updater-queue-capacity=20 +#select ID BY FID +ct.pipeline.id-updater-stage-concurrency=10 +ct.pipeline.id-updater-queue-capacity=20 + + #select and calc flags ct.flag-updater.concurrency=10 ct.flag-updater.queue-capacity=100 diff --git a/pom.xml b/pom.xml index c784c5c..3338c8e 100644 --- a/pom.xml +++ b/pom.xml @@ -233,14 +233,74 @@ org.springframework.boot spring-boot-maven-plugin - - - - org.projectlombok - lombok - - - + + + + repackage + + repackage + + + true + + + + build-common + package + + repackage + + + ru.nbch.credit_tracker.service.mode.common.CTMainApplication + main + true + + + org.projectlombok + lombok + + + + + + + build-monitoring + package + + repackage + + + ru.nbch.credit_tracker.service.mode.monitoring.CTMonitoringApplication + monitoring + true + + + org.projectlombok + lombok + + + + + + + build-migration + package + + repackage + + + ru.nbch.credit_tracker.service.mode.migration.CTMigrationApplication + migration + true + + + org.projectlombok + lombok + + + + + pl.project13.maven diff --git a/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java b/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java deleted file mode 100644 index 548e154..0000000 --- a/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java +++ /dev/null @@ -1,25 +0,0 @@ -package ru.nbch.credit_tracker; - -import org.springframework.boot.SpringApplication; -import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.context.ConfigurableApplicationContext; -import ru.nbch.credit_tracker.service.MonitoringService; -import ru.nbch.credit_tracker.service.StartupRestoreService; - -@SpringBootApplication -public class CreditTrackerApplication { - - public static void main(String[] args) { - ConfigurableApplicationContext ctx = SpringApplication.run(CreditTrackerApplication.class, args); -// if (args.length > 0) { -// String arg = args[0]; -// if (arg.equalsIgnoreCase("migration")) { -// } -// } - StartupRestoreService updateProcessingPackagesService = ctx.getBean(StartupRestoreService.class); - MonitoringService monitoringService = ctx.getBean(MonitoringService.class); - updateProcessingPackagesService.run(); - monitoringService.run(); - } - -} 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 95a1bd5..f6421b6 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 @@ -12,6 +12,10 @@ public record PipelineSettings( @DefaultValue("20") Integer nameEnricherQueueCapacity, @DefaultValue("10") Integer idEnricherStageConcurrency, @DefaultValue("15") Integer idEnricherQueueCapacity, + @DefaultValue("15") Integer nameUpdaterStageConcurrency, + @DefaultValue("20") Integer nameUpdaterQueueCapacity, + @DefaultValue("10") Integer idUpdaterStageConcurrency, + @DefaultValue("15") Integer idUpdaterQueueCapacity, @DefaultValue("50") Integer mergerQueueCapacity, @DefaultValue("5") Integer errorAttemptCount, @DefaultValue("2s") Duration errorAttemptDelay, diff --git a/src/main/java/ru/nbch/credit_tracker/config/task/ExecutorsConfig.java b/src/main/java/ru/nbch/credit_tracker/config/task/ExecutorsConfig.java index 158cc10..7ea51eb 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/task/ExecutorsConfig.java +++ b/src/main/java/ru/nbch/credit_tracker/config/task/ExecutorsConfig.java @@ -9,9 +9,9 @@ import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; @Configuration public class ExecutorsConfig { - @Bean + @Bean(destroyMethod = "shutdown") public ExecutorService pipelineExecutorService() { - return Executors.newFixedThreadPool(10, r -> { + return Executors.newFixedThreadPool(15, r -> { Thread t = new Thread(r); t.setName("pipe-starter-%d".formatted(t.threadId())); return t; 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 c78c000..cadf8ae 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 @@ -17,7 +17,7 @@ public class StageConfig { @Bean("nameUpdaterQueueFactory") public StageQueueFactory nameUpdaterQueueFactory(CreditTrackerSettings creditTrackerSettings) { - return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameEnricherQueueCapacity()); + return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameUpdaterQueueCapacity()); } @Bean("idEnricherQueueFactory") @@ -27,7 +27,7 @@ public class StageConfig { @Bean("idUpdaterQueueFactory") public StageQueueFactory idUpdaterQueueFactory(CreditTrackerSettings creditTrackerSettings) { - return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameEnricherQueueCapacity()); + return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameUpdaterQueueCapacity()); } @Bean("noOpQueueFactory") diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/common/CTMainApplication.java b/src/main/java/ru/nbch/credit_tracker/service/mode/common/CTMainApplication.java new file mode 100644 index 0000000..bf5db89 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/common/CTMainApplication.java @@ -0,0 +1,30 @@ +package ru.nbch.credit_tracker.service.mode.common; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.ComponentScan; +import org.springframework.context.annotation.FilterType; + +@SpringBootApplication +@ComponentScan( + value = {"ru.nbch.credit_tracker"}, + excludeFilters = { + @ComponentScan.Filter( + type = FilterType.REGEX, + pattern = { + "ru\\.nbch\\.credit_tracker\\.service\\.mode\\.migration..*", + "ru\\.nbch\\.credit_tracker\\.service\\.mode\\.monitoring..*" + } + ) + } +) +public class CTMainApplication { + + public static void main(String[] args) { + ConfigurableApplicationContext ctx = SpringApplication.run(CTMainApplication.class, args); + StartupRestoreService updateProcessingPackagesService = ctx.getBean(StartupRestoreService.class); + updateProcessingPackagesService.run(); + } + +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/StartupRestoreService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java similarity index 97% rename from src/main/java/ru/nbch/credit_tracker/service/StartupRestoreService.java rename to src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java index 4de6e2f..a692d50 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/StartupRestoreService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java @@ -1,4 +1,4 @@ -package ru.nbch.credit_tracker.service; +package ru.nbch.credit_tracker.service.mode.common; import java.nio.file.Path; import java.util.List; @@ -10,6 +10,7 @@ import ru.nbch.credit_tracker.entities.app.UserPackage; import ru.nbch.credit_tracker.enums.Status; import ru.nbch.credit_tracker.mapper.signal.PackagesRecordsMapper; import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; +import ru.nbch.credit_tracker.service.MonitoringErrorProcesingService; import ru.nbch.credit_tracker.service.task.Pipeline; import ru.nbch.credit_tracker.service.task.PipelinePackageFabric; import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator; diff --git a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFidAndPdService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidAndPdService.java similarity index 98% rename from src/main/java/ru/nbch/credit_tracker/service/UpdatingFidAndPdService.java rename to src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidAndPdService.java index 37b7735..df8ac46 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFidAndPdService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidAndPdService.java @@ -1,4 +1,4 @@ -package ru.nbch.credit_tracker.service; +package ru.nbch.credit_tracker.service.mode.common; import java.util.List; import java.util.concurrent.CompletableFuture; diff --git a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFlagsService.java similarity index 98% rename from src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java rename to src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFlagsService.java index 194e9de..b3eb6cb 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFlagsService.java @@ -1,4 +1,4 @@ -package ru.nbch.credit_tracker.service; +package ru.nbch.credit_tracker.service.mode.common; import java.util.List; import java.util.concurrent.atomic.AtomicBoolean; diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/migration/CTMigrationApplication.java b/src/main/java/ru/nbch/credit_tracker/service/mode/migration/CTMigrationApplication.java new file mode 100644 index 0000000..8313937 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/migration/CTMigrationApplication.java @@ -0,0 +1,32 @@ +package ru.nbch.credit_tracker.service.mode.migration; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.ComponentScan; +import org.springframework.context.annotation.FilterType; + +@SpringBootApplication +@ComponentScan( + value = {"ru.nbch.credit_tracker"}, + excludeFilters = { + @ComponentScan.Filter( + type = FilterType.REGEX, + pattern = { + "ru\\.nbch\\.credit_tracker\\.service\\.mode\\.common..*", + "ru\\.nbch\\.credit_tracker\\.service\\.mode\\.monitoring..*" + } + ) + } +) +public class CTMigrationApplication { + public static void main(String[] args) { + SpringApplication application = new SpringApplication(CTMigrationApplication.class); + application.setWebApplicationType(WebApplicationType.NONE); + ConfigurableApplicationContext ctx = application.run(args); + MigrationService migrationService = ctx.getBean(MigrationService.class); + migrationService.update(); + } + +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/MigrationService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/migration/MigrationService.java similarity index 64% rename from src/main/java/ru/nbch/credit_tracker/service/MigrationService.java rename to src/main/java/ru/nbch/credit_tracker/service/mode/migration/MigrationService.java index ebb39e3..5698031 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/MigrationService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/migration/MigrationService.java @@ -1,15 +1,13 @@ -package ru.nbch.credit_tracker.service; +package ru.nbch.credit_tracker.service.mode.migration; import java.util.concurrent.CompletableFuture; import lombok.extern.slf4j.Slf4j; +import org.springframework.boot.SpringApplication; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.stereotype.Service; -import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; -import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; -import ru.nbch.credit_tracker.service.signal.PhonesMapService; 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; @@ -20,21 +18,14 @@ import ru.nbch.credit_tracker.service.task.impl.PipelineMode; public class MigrationService { private final PipelinePackageFabric pipelinePackageManager; private final PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator; - private final PipelinePackageOrchestrator pipelinePackageOrchestrator; - private final UserPackageMapper userPackageMapper; - private final PhonesMapService phonesMapService; + private final ConfigurableApplicationContext context; - public MigrationService(CreditTrackerSettings config, - PipelinePackageFabric pipelinePackageManager, + public MigrationService(PipelinePackageFabric pipelinePackageManager, PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator, - PipelinePackageOrchestrator pipelinePackageOrchestrator, - UserPackageMapper userPackageMapper, - PhonesMapService phonesMapService) { + ConfigurableApplicationContext context) { this.pipelinePackageManager = pipelinePackageManager; this.pipelineRecalcPDOrchestrator = pipelineRecalcPDOrchestrator; - this.pipelinePackageOrchestrator = pipelinePackageOrchestrator; - this.userPackageMapper = userPackageMapper; - this.phonesMapService = phonesMapService; + this.context = context; } public void update() { @@ -51,6 +42,7 @@ public class MigrationService { log.info("pipeline finished"); pipeline.close(); pipelineRecalcPDOrchestrator.release(); + new Thread(this::shutdown, "migration-app-exit").start(); } }); } catch (Exception e) { @@ -58,4 +50,12 @@ public class MigrationService { } } + private void shutdown() { + int exitCode = SpringApplication.exit(context, () -> { + // no errors + return 0; + }); + System.exit(exitCode); + } + } diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/CTMonitoringApplication.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/CTMonitoringApplication.java new file mode 100644 index 0000000..9e6c42d --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/CTMonitoringApplication.java @@ -0,0 +1,31 @@ +package ru.nbch.credit_tracker.service.mode.monitoring; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.ComponentScan; +import org.springframework.context.annotation.FilterType; + +@SpringBootApplication +@ComponentScan( + value = {"ru.nbch.credit_tracker"}, + excludeFilters = { + @ComponentScan.Filter( + type = FilterType.REGEX, + pattern = { + "ru\\.nbch\\.credit_tracker\\.service\\.mode\\.common..*", + "ru\\.nbch\\.credit_tracker\\.service\\.mode\\.migration..*" + } + ) + } +) +public class CTMonitoringApplication { + public static void main(String[] args) { + SpringApplication application = new SpringApplication(CTMonitoringApplication.class); + application.setWebApplicationType(WebApplicationType.NONE); + ConfigurableApplicationContext ctx = application.run(args); + MonitoringService monitoringService = ctx.getBean(MonitoringService.class); + monitoringService.run(); + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/MonitoringService.java similarity index 99% rename from src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java rename to src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/MonitoringService.java index 14bfd1c..f7b8380 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/MonitoringService.java @@ -1,4 +1,4 @@ -package ru.nbch.credit_tracker.service; +package ru.nbch.credit_tracker.service.mode.monitoring; import java.io.BufferedInputStream; import java.io.IOException; diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java index bce267a..fde7692 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java @@ -68,7 +68,7 @@ public class FidIdUpdater implements Handler { @Override public int preferredConcurrency() { - return creditTrackerSettings.getPipeline().idEnricherStageConcurrency(); + return creditTrackerSettings.getPipeline().idUpdaterStageConcurrency(); } @Override diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java index 1e44469..359d4a7 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java @@ -92,7 +92,7 @@ public class FidNameUpdater implements Handler { @Override public int preferredConcurrency() { - return creditTrackerSettings.getPipeline().nameEnricherStageConcurrency(); + return creditTrackerSettings.getPipeline().nameUpdaterStageConcurrency(); } @Override diff --git a/src/main/resources/sql/v2.21.ddl b/src/main/resources/sql/v2.21.ddl index 7b9c12e..360e9af 100644 --- a/src/main/resources/sql/v2.21.ddl +++ b/src/main/resources/sql/v2.21.ddl @@ -42,4 +42,15 @@ update signals_phone_fid_map set phone_l = phone::BIGINT where phone is not null update signals_user_packages set signal_types = '1' where signal_types is null; update signals_user_packages set external_package_name = concat(substring(membercode, 1, 6), '_', package_name) where membercode is not null and package_name is not null ; +INSERT +INTO signals_phones_map +(fid, phone, flag_90_12, flag_debt5000, calculated_at) +SELECT distinct on (phone::bigint) fid::int, + phone::bigint, + flag_90_12, + flag_debt5000, + '2025-01-01' +FROM signals_phone_fid_map where fid is not null +order by phone, calculated_at desc; + insert into app_version values ('2.0') \ No newline at end of file