From 0f98d8d28d5e696122389f80612a7425d1452009 Mon Sep 17 00:00:00 2001 From: etreschenkov Date: Tue, 25 Nov 2025 15:59:51 +0300 Subject: [PATCH] add new tables --- credit-tracker.properties | 30 +++++--- .../settings/CreditTrackerSettings.java | 1 + .../entities/app/UserPackage.java | 2 +- .../credit_tracker/facade/SignalFacadeV3.java | 2 +- .../mapper/signal/PhoneFidMapMapper.java | 2 +- .../repository/PhoneRepository.java | 9 +-- .../service/UpdatingFlagsService.java | 6 +- .../service/indic/PhoneService2.java | 4 ++ .../service/task/PipelinePackageManager.java | 29 ++++++-- .../service/task/impl/KickPipeLineImpl.java | 43 ++++++++++- .../service/task/impl/Saver.java | 72 +++++++++++++++++++ .../ru/nbch/credit_tracker/utils/LogUtil.java | 2 + .../mapper/signal/UserPackageMapper.xml | 4 +- 13 files changed, 175 insertions(+), 31 deletions(-) create mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/impl/Saver.java diff --git a/credit-tracker.properties b/credit-tracker.properties index 0ccb297..3f9b9fc 100644 --- a/credit-tracker.properties +++ b/credit-tracker.properties @@ -11,7 +11,7 @@ ct.indic-data-source-settings.driver=com.ibm.db2.jcc.DB2Driver ct.indic-data-source-settings.url=${DB_URL_INDIC:jdbc:db2://10.230.227.100:2668/cprosd22:currentSchema=INDIC;} ct.indic-data-source-settings.username=${USERNAME_INDIC:ros9appl} ct.indic-data-source-settings.password=${PASSWORD_INDIC:R0s9Appl} -ct.indic-data-source-settings.maximum-pool-size=${MAXIMUM_POOL_SIZE_INDIC:10} +ct.indic-data-source-settings.maximum-pool-size=${MAXIMUM_POOL_SIZE_INDIC:30} ct.indic-data-source-settings.minimum-idle=${MINIMUM_IDLE_INDIC:10} ct.indic-data-source-settings.connection-timeout=${CONNECTION_TIMEOUT_INDIC:10000} ct.indic-data-source-settings.validation-timeout=${VALIDATION_TIMEOUT_INDIC:5000} @@ -25,7 +25,7 @@ ct.signal-data-source-settings.driver=org.postgresql.Driver ct.signal-data-source-settings.url=${DB_URL_SIGNAL:jdbc:postgresql://10.200.200.133:5432/credit_tracker} ct.signal-data-source-settings.username=${USERNAME_SIGNAL:credit_tracker_user} ct.signal-data-source-settings.password=${PASSWORD_SIGNAL:-} -ct.signal-data-source-settings.maximum-pool-size=${MAXIMUM_POOL_SIZE_SIGNAL:10} +ct.signal-data-source-settings.maximum-pool-size=${MAXIMUM_POOL_SIZE_SIGNAL:40} ct.signal-data-source-settings.minimum-idle=${MINIMUM_IDLE_SIGNAL:10} ct.signal-data-source-settings.connection-timeout=${CONNECTION_TIMEOUT_SIGNAL:10000} ct.signal-data-source-settings.validation-timeout=${VALIDATION_TIMEOUT_SIGNAL:5000} @@ -39,7 +39,7 @@ ct.scoring-data-source-settings.driver=com.ibm.db2.jcc.DB2Driver ct.scoring-data-source-settings.url=${DB_URL_SCORING:jdbc:db2://10.230.227.101:2668/cprosd22:currentSchema=TEST_2_INDIC;} ct.scoring-data-source-settings.username=${USERNAME_SCORING:tester} ct.scoring-data-source-settings.password=${PASSWORD_SCORING:tester} -ct.scoring-data-source-settings.maximum-pool-size=${MAXIMUM_POOL_SIZE_SCORING:10} +ct.scoring-data-source-settings.maximum-pool-size=${MAXIMUM_POOL_SIZE_SCORING:30} ct.scoring-data-source-settings.minimum-idle=${MINIMUM_IDLE_SCORING:10} ct.scoring-data-source-settings.connection-timeout=${CONNECTION_TIMEOUT_SCORING:10000} ct.scoring-data-source-settings.validation-timeout=${VALIDATION_TIMEOUT_SCORING:5000} @@ -56,7 +56,7 @@ spring.servlet.multipart.max-request-size=${ACCEPTED_ARCHIVE_SIZE:2048MB} ct.xsd-validation.monitoring-request-path=monitoring-request.xsd ct.xsd-validation.signal-package-path=signal-package.xsd -ct.signals-root-uri=${SIGNALS_ROOT_URI:http://10.230.227.102:11094/signalRUalfa/} +ct.signals-root-uri=${SIGNALS_ROOT_URI:http://127.0.01:11094/signalRUalfa/} ct.signals-token=${SIGNALS_TOKEN:gttcHxwcGojoAvZK} ct.signals-user=${SIGNALS_USER:0001ZZ000001} @@ -73,7 +73,21 @@ ct.db-version=${DB_VERSION:1234} ct.encoding=windows-1251 -#Batch size and global pipeline timeout -ct.batchSize=1000 -#in minute -ct.timeout=30 \ No newline at end of file + +#----------------------------Pipeline settings----------------------------# +#???-?? ????????? ?? xml ?? ???? ???????????? +ct.saver-subject-batch-size=15000 +ct.pipeline.subject-batch-size=5000 +#????????? ? PHONE ? NAME + ?????? ? ???? SIGNALS_PHONE_FID_MAP +ct.pipeline.name-enricher-stage-concurrency=20 +#????????? ? ID ? ?????? ????? xml ??? ??. ??????? +ct.pipeline.id-enricher-stage-concurrency=10 +#?????????? ??????? ??? ?????? +ct.pipeline.error-attempt-count=5 +#????? ???????? ????? ????????? ??????? +ct.pipeline.error-attempt-delay=2s +#??????? ?? ???? ??????? +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 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 03dd774..4d57e97 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 @@ -35,4 +35,5 @@ public class CreditTrackerSettings { private Long daysToUpdateFlags; private String encoding; private String dbVersion; + private Integer saverSubjectBatchSize; } diff --git a/src/main/java/ru/nbch/credit_tracker/entities/app/UserPackage.java b/src/main/java/ru/nbch/credit_tracker/entities/app/UserPackage.java index 2fbac8a..b703a30 100644 --- a/src/main/java/ru/nbch/credit_tracker/entities/app/UserPackage.java +++ b/src/main/java/ru/nbch/credit_tracker/entities/app/UserPackage.java @@ -17,5 +17,5 @@ public class UserPackage { private String status; private LocalDateTime deactivatedAt; private String lastDownloadedReport; - private LocalDateTime monitoringAt; +// private LocalDateTime monitoringAt; } diff --git a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java index 0664d10..c117ca5 100644 --- a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java +++ b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java @@ -128,7 +128,7 @@ public class SignalFacadeV3 { Long packageId = signalService.registerPackage(userId, packageName); LogUtil.addMdcLogKey("%s %d".formatted(packageName, packageId)); - log.info("Package registered. PackageId {}. Status: {}", packageId, Status.PROCESSING.name()); + log.info("Package registered. PackageId {}. Status: {}", packageId, Status.ACCEPTED.name()); pipelinePackageManager.buildAndRun(packageId, parsedFilePath); } catch (Exception e) { try { diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java index 3ca07f4..9ccdaf1 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java @@ -17,7 +17,7 @@ public interface PhoneFidMapMapper { void setFidJson(@Param("json") String json); void registerWithId(@Param("phoneFidMap") List phoneFidMap); - List findFullPhoneFidMap(@Param("packageId") Long packageId, @Param("offset") int offset, @Param("limit") int limit); + List findFullPhoneFidMap(@Param("packageId") Long packageId, @Param("limit") int limit, @Param("offset") int offset); void updatePhoneFidMaps(@Param("subjects") Collection subjects); void updateOne(@Param("subject") PhoneFidMap subject); diff --git a/src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java b/src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java index d1b00b1..88bdcc3 100644 --- a/src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java +++ b/src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java @@ -37,15 +37,8 @@ public class PhoneRepository { SELECT pn.number, pn.fid, - pn.file_since_dt, - nf.first, - nf.middle, - nf.name_1, - nf.birth_dt, - nf.file_since_dt as file_since_dt_name + pn.file_since_dt FROM PHONE_NORM pn - JOIN NAME nf - ON nf.fid = pn.fid WHERE number in (:phones) WITH UR """; 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 dace278..37776b3 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java @@ -69,9 +69,9 @@ public class UpdatingFlagsService { SignalService.UpdateResult updateResult = signalService.updateSubjects(packageId, filter); if (updateResult.isUpdated()) { if (signalRestService.setupMonitoring(packageId)) { - LocalDateTime monitoringAt = userPackage.getMonitoringAt() != null ? userPackage.getMonitoringAt() : LocalDateTime.now(); - log.info("Flags were updated successfully. PackageId: {} PackageName: {} MonitoringAt: {}. Package status changed to ON_MONITORING", packageId, packageName, monitoringAt); - signalService.updatePackageOnMonitoringSuccess(packageId, monitoringAt); +// LocalDateTime monitoringAt = userPackage.getMonitoringAt() != null ? userPackage.getMonitoringAt() : LocalDateTime.now(); + log.info("Flags were updated successfully. PackageId: {} PackageName: {}. Package status changed to ON_MONITORING", packageId, packageName); + signalService.updatePackageOnMonitoringSuccess(packageId, LocalDateTime.now()); monitoringErrorProcesingService.findErrors(packageId); } else { log.info("Updated package registration was rejected by signal service. PackageId: {} PackageName: {}", packageId, packageName); diff --git a/src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService2.java b/src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService2.java index 8edd9ed..8132099 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService2.java +++ b/src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService2.java @@ -90,4 +90,8 @@ public class PhoneService2 { throw new RuntimeException(e); } } + + public List nextPage(Long packageId, Integer limit, Integer offset) { + return phoneFidMapMapper.findFullPhoneFidMap(packageId, limit, offset); + } } 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 0d480aa..354ca1d 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 @@ -1,6 +1,7 @@ package ru.nbch.credit_tracker.service.task; import java.nio.file.Path; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Semaphore; @@ -21,6 +22,7 @@ import ru.nbch.credit_tracker.service.task.impl.FidIdEnricher; import ru.nbch.credit_tracker.service.task.impl.FidNameEnricher; import ru.nbch.credit_tracker.service.task.impl.FlagsEnricher; import ru.nbch.credit_tracker.service.task.impl.SaveSubjectHandler; +import ru.nbch.credit_tracker.service.task.impl.Saver; import ru.nbch.credit_tracker.service.task.impl.XmlMergeCoordinator; import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter; import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator; @@ -48,8 +50,15 @@ public class PipelinePackageManager { t.setName("pipe-starter-%d".formatted(t.threadId())); 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 Semaphore semaphore; private final PipelineBackoff pipelineBackoff; + private final Saver saver; private final static int MAX_PARALLEL_REQUESTS = 3; public PipelinePackageManager(SignalRestService signalRestService, @@ -64,7 +73,7 @@ public class PipelinePackageManager { BiFunction> pipelineStarterFactory, FlagsEnricher flagsEnricherV2, FidNameEnricher fidNameEnricher, CreditTrackerSettings creditTrackerSettings, - PipelineBackoff pipelineBackoff) { + PipelineBackoff pipelineBackoff, Saver saver) { this.signalRestService = signalRestService; this.saveSubjectQueueFactory = saveSubjectQueueFactory; this.idEnricherQueueFactory = idEnricherQueueFactory; @@ -81,6 +90,7 @@ public class PipelinePackageManager { this.semaphore = new Semaphore(creditTrackerSettings.getPipeline().parallelRequests()); log.info("pipeline settings: {}", creditTrackerSettings.getPipeline()); this.pipelineBackoff = pipelineBackoff; + this.saver = saver; } public void buildAndRun(Long packageId, Path pkgPath) throws Exception { @@ -90,10 +100,7 @@ public class PipelinePackageManager { log.info("throttler permits after acquire: {}", semaphore.availablePermits()); long startTime = System.currentTimeMillis(); try { - Pipeline pipeline = PipeLineBuilderV2.first(saveSubjectQueueFactory.getObject(), idEnricherQueueFactory.getObject()) - .handler(saveSubjectHandler) - .andThen() - .stage(nameEnricherQueueFactory.getObject()) + Pipeline pipeline = PipeLineBuilderV2.first(nameEnricherQueueFactory.getObject(), idEnricherQueueFactory.getObject()) .handler(fidNameEnricher) .handlerResult(flagsEnricherV2) .andThen() @@ -106,7 +113,6 @@ public class PipelinePackageManager { .backoff(pipelineBackoff) .build(); - pipeline.start(executorService); pipeline.getTerminator() .doneFuture() .orTimeout(creditTrackerSettings.getPipeline().globalTimeout().toMillis(), TimeUnit.MILLISECONDS) @@ -133,6 +139,17 @@ public class PipelinePackageManager { log.info("semaphore amount: {}", semaphore.availablePermits()); } }); + + CompletableFuture + .runAsync(() -> { + try { + saver.readAndSave(packageId, pkgPath); + } catch (Throwable t) { + log.error(ExceptionUtils.getStackTrace(t)); + pipeline.close(); + } + }, subjectSaverExecutor) + .thenRun(() -> pipeline.start(executorService)); } catch (Exception e) { semaphore.release(); throw e; diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java index 83a04a0..077fd8b 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java @@ -11,6 +11,7 @@ import java.util.stream.Collectors; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.entities.app.PhoneFidMap; import ru.nbch.credit_tracker.entities.app.UserPackage; import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest; import ru.nbch.credit_tracker.exceptions.SignalClientException; @@ -56,7 +57,47 @@ public class KickPipeLineImpl implements IPipeLineStarter { @Override public void startFillingQueue(BlockingQueue firstQueue) { - log.info("Start registering subjects. PackageId: {}", packageId); + try { + log.info("Start registering subjects. PackageId: {};", packageId); + int offset = 0; + int limit = BATCH_SIZE; + UserPackage userPackage = userPackageMapper.findPackage(packageId); + List page = phoneService2.nextPage(packageId, limit, offset); + while (!page.isEmpty()) { + log.debug("Get page {}", page.size()); + List subjectProfiles = page.stream().map(phoneFidMap -> SubjectProfile.builder() + .identifier(phoneFidMap.getSubjectId()) + .phone(phoneFidMap.getPhone()) + .build()).toList(); + try { + firstQueue.put(new SubjectPayload(packageId, userPackage.getUserId(), userPackage.getPackageName(), subjectProfiles)); + } catch (InterruptedException e) { + isInterrupted = true; + log.error("Pipeline starter was interrupted!"); + throw new PipelineStarterException(); + } + offset += limit; + page = phoneService2.nextPage(packageId, limit, offset); + } + } catch (PipelineStarterException e) { + log.info("Pipeline starter shitting down..."); + } finally { + if (isInterrupted) { + //noinspection ReturnInsideFinallyBlock + return; + } + log.debug("Send poison pill"); + try { + SubjectPayload poison = new SubjectPayload(packageId, PoisonType.NORMALLY); + firstQueue.put(poison); + } catch (InterruptedException e) { + log.error("couldn't send poison: {}", ExceptionUtils.getStackTrace(e)); + } + } + } +// @Override + public void startFillingQueueXml(BlockingQueue firstQueue) { + log.info("Start registering subjects. PackageId: {}; pkg path: {}", packageId, pkgPath); UserPackage userPackage = userPackageMapper.findPackage(packageId); AtomicInteger counter = new AtomicInteger(0); try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(pkgPath), 8 * 1024 * 1024)) { diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/Saver.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/Saver.java new file mode 100644 index 0000000..381dcbf --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/Saver.java @@ -0,0 +1,72 @@ +package ru.nbch.credit_tracker.service.task.impl; + +import java.io.BufferedInputStream; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +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.entities.user_request.monitoring_request.MonitoringRequest; +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.service.indic.PhoneService2; +import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; +import ru.nbch.credit_tracker.utils.error.ExceptionUtils; + +@Service +public class Saver { + private final Logger log = LoggerFactory.getLogger(getClass()); + private final StaxXmlProcessor xmlProcessor; + private final PhoneService2 phoneService2; + private final UserPackageMapper userPackageMapper; + private final int BATCH_SIZE; + + + public Saver(StaxXmlProcessor xmlProcessor, + PhoneService2 phoneService2, + UserPackageMapper userPackageMapper, + CreditTrackerSettings creditTrackerSettings) { + this.xmlProcessor = xmlProcessor; + this.phoneService2 = phoneService2; + this.userPackageMapper = userPackageMapper; + this.BATCH_SIZE = creditTrackerSettings.getSaverSubjectBatchSize(); + } + + public void readAndSave(Long packageId, Path pkgPath) { + log.info("Start registering subjects. PackageId: {}; pkg path: {}", packageId, pkgPath); + try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(pkgPath), 8 * 1024 * 1024)) { + xmlProcessor.read(in, MonitoringRequest.class, + request -> { + if (!request.getSubjects().getSubject().isEmpty()) { + List subjects = request.getSubjects().getSubject().stream().map(subject -> + SubjectProfile.builder() + .identifier(subject.getId()) + .phone(subject.getClientId()) + .build() + + ).toList(); + log.trace("Reading subject's {} batch. PackageId: {}", request.getSubjects().getSubject().size(), packageId); + List subjectForSave = subjects.stream().map(subject -> PhoneFidMap.builder() + .subjectId(subject.getIdentifier()) + .packageId(packageId) + .phone(subject.getPhone()) + .build()).toList(); + request.getSubjects().getSubject().clear(); + phoneService2.registerSubjects(subjectForSave); + } + }, + request -> request.getSubjects().getSubject().isEmpty(), + BATCH_SIZE, "/MonitoringRequest/Subjects/Subject" + ); + } catch (IOException e) { + log.error("{}", ExceptionUtils.getStackTrace(e)); + } + userPackageMapper.updatePackageStatus(packageId, Status.PROCESSING.name()); + log.info("Complete registering subjects. PackageId: {}; pkg path: {}", packageId, pkgPath); + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/utils/LogUtil.java b/src/main/java/ru/nbch/credit_tracker/utils/LogUtil.java index de07d12..883bac7 100644 --- a/src/main/java/ru/nbch/credit_tracker/utils/LogUtil.java +++ b/src/main/java/ru/nbch/credit_tracker/utils/LogUtil.java @@ -68,6 +68,8 @@ public class LogUtil { } task.run(); + } catch (Throwable e) { + log.error(ExceptionUtils.getStackTrace(e)); } finally { // восстанавливаем предыдущий (может быть null) try { diff --git a/src/main/resources/mybatis/mapper/signal/UserPackageMapper.xml b/src/main/resources/mybatis/mapper/signal/UserPackageMapper.xml index fe58d65..e4f74d5 100644 --- a/src/main/resources/mybatis/mapper/signal/UserPackageMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/UserPackageMapper.xml @@ -13,7 +13,7 @@ - + - SELECT id, membercode, package_name, created_at, status, deactivated_at, last_downloaded_report, monitoring_at + SELECT id, membercode, package_name, created_at, status, deactivated_at, last_downloaded_report FROM signals_user_packages WHERE id = #{id} LIMIT 1