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 c38ce62..517c872 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 @@ -25,7 +25,7 @@ public class StageConfig { @Bean("partFileQueueFactory") public StageQueueFactory partFileQueueFactory() { - return new StageQueueFactory<>(100); + return new StageQueueFactory<>(500); } } 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 88bdcc3..d1b00b1 100644 --- a/src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java +++ b/src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java @@ -37,8 +37,15 @@ public class PhoneRepository { SELECT pn.number, pn.fid, - pn.file_since_dt + pn.file_since_dt, + nf.first, + nf.middle, + nf.name_1, + nf.birth_dt, + nf.file_since_dt as file_since_dt_name 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/task/Stage.java b/src/main/java/ru/nbch/credit_tracker/service/task/Stage.java index b78e508..8fa190c 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/Stage.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/Stage.java @@ -67,7 +67,7 @@ public class Stage implements Ru @Override public void run() { - List> inflight = new ArrayList<>(); + List> inflight = new ArrayList<>(); log.info("this is main thread for {}", dispatcherName); try { int counter = 0; @@ -79,7 +79,7 @@ public class Stage implements Ru if (b.isPoison()) { log.info("Got poison"); // дождаться всех in-flight задач - for (Future f : inflight) { + for (Future f : inflight) { if (wasInterrupted()) { log.info("stage {} was interrupted while waiting for tasks to finish", dispatcherName); break; @@ -94,44 +94,44 @@ public class Stage implements Ru counter++; int finalCounter = counter; AtomicInteger retryForSingleTask = new AtomicInteger(0); - Future f = ioPool.submit(() -> { - while (retryForSingleTask.incrementAndGet() < pipelineBackoff.getRetryCount()) { - try { - if (Thread.currentThread().isInterrupted()) { - log.info("stage {} was interrupted. skipping task {}...", dispatcherName, finalCounter); - return null; - } - log.info("stage {} task {} started.", dispatcher.getClass().getSimpleName(), finalCounter); - Z p; - try (var m = stopWatch.start(dispatcherName + "-handler")){ - p = dispatcher.handle(b); - } - if (Thread.currentThread().isInterrupted()) { - log.info("after {} task {} finished, worker has been interrupted. not putting result in queue.", - dispatcherName, finalCounter); - return null; - } - putQuiet(out, p); - log.info("stage {} task {} ended.", dispatcher.getClass().getSimpleName(), finalCounter); - return p; - } catch (Exception e) { - boolean isRetryable = pipelineBackoff.isRetryable(e); - if (!isRetryable) { - log.error("{} ERROR: {}", dispatcherName, ExceptionUtils.getStackTrace(e)); - pipeline.close(); - return null; - } - log.warn("{} RETRY AFTER ERROR: {}", dispatcherName, ExceptionUtils.getStackTrace(e)); - pipelineBackoff.backoff(); + Future f = ioPool.submit(() -> { + while (retryForSingleTask.incrementAndGet() < pipelineBackoff.getRetryCount()) { + try { + if (Thread.currentThread().isInterrupted()) { + log.info("stage {} was interrupted. skipping task {}...", dispatcherName, finalCounter); + return null; } + log.info("stage {} task {} started.", dispatcher.getClass().getSimpleName(), finalCounter); + Z p; + try (var m = stopWatch.start(dispatcherName + "-handler")) { + p = dispatcher.handle(b); + } + if (Thread.currentThread().isInterrupted()) { + log.info("after {} task {} finished, worker has been interrupted. not putting result in queue.", + dispatcherName, finalCounter); + return null; + } + putQuiet(out, p); + log.info("stage {} task {} ended.", dispatcher.getClass().getSimpleName(), finalCounter); + return null; + } catch (Exception e) { + boolean isRetryable = pipelineBackoff.isRetryable(e); + if (!isRetryable) { + log.error("{} ERROR: {}", dispatcherName, ExceptionUtils.getStackTrace(e)); + pipeline.close(); + return null; + } + log.warn("{} RETRY AFTER ERROR: {}", dispatcherName, ExceptionUtils.getStackTrace(e)); + pipelineBackoff.backoff(); } - log.error("tried for {} times, failed batch...", retryForSingleTask.get()); - pipeline.close(); - return null; } - ); + log.error("tried for {} times, failed batch...", retryForSingleTask.get()); + pipeline.close(); + return null; + }); inflight.add(f); if ((counter % 20) == 0) { + log.debug("Clean inflight future"); inflight.removeIf(Future::isDone); } } 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 077fd8b..a76b344 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 @@ -4,10 +4,10 @@ import java.io.BufferedInputStream; import java.io.IOException; import java.nio.file.Files; import java.nio.file.Path; +import java.util.ArrayList; import java.util.List; import java.util.concurrent.BlockingQueue; import java.util.concurrent.atomic.AtomicInteger; -import java.util.stream.Collectors; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; @@ -66,6 +66,7 @@ public class KickPipeLineImpl implements IPipeLineStarter { while (!page.isEmpty()) { log.debug("Get page {}", page.size()); List subjectProfiles = page.stream().map(phoneFidMap -> SubjectProfile.builder() + .mapId(phoneFidMap.getId()) .identifier(phoneFidMap.getSubjectId()) .phone(phoneFidMap.getPhone()) .build()).toList(); @@ -104,17 +105,27 @@ public class KickPipeLineImpl implements IPipeLineStarter { xmlProcessor.read(in, MonitoringRequest.class, request -> { if (!request.getSubjects().getSubject().isEmpty()) { - List subjectProfileBatch = request.getSubjects().getSubject().stream().map(subject -> - SubjectProfile.builder() - .identifier(subject.getId()) - .phone(subject.getClientId()) - .build() - ).collect(Collectors.toList()); + List subjectForSave = request.getSubjects().getSubject().stream().map(subject -> PhoneFidMap.builder() + .subjectId(subject.getId()) + .packageId(packageId) + .phone(subject.getClientId()) + .build()).toList(); + List ids = phoneService2.registerSubjects(subjectForSave); + List subjectProfiles = new ArrayList<>(); + for (int i = 0; i < ids.size(); i++) { + PhoneFidMap phoneFidMap = subjectForSave.get(i); + subjectProfiles.add(SubjectProfile.builder() + .identifier(phoneFidMap.getSubjectId()) + .phone(phoneFidMap.getPhone()) + .mapId(ids.get(i)) + .build()); + } + int count = counter.incrementAndGet(); log.trace("Reading subject's {} batch. Total count: {}. PackageId: {}", count, request.getSubjects().getSubject().size(), packageId); try { - firstQueue.put(new SubjectPayload(packageId,userPackage.getUserId(), userPackage.getPackageName(), subjectProfileBatch)); + firstQueue.put(new SubjectPayload(packageId,userPackage.getUserId(), userPackage.getPackageName(), subjectProfiles)); } catch (InterruptedException e) { isInterrupted = true; log.error("Pipeline starter was interrupted!"); 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 index 381dcbf..c8ea7fb 100644 --- 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 @@ -13,7 +13,6 @@ 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; @@ -43,21 +42,14 @@ public class Saver { 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()) + List subjectForSave = request.getSubjects().getSubject().stream().map(subject -> PhoneFidMap.builder() + .subjectId(subject.getId()) .packageId(packageId) - .phone(subject.getPhone()) + .phone(subject.getClientId()) .build()).toList(); - request.getSubjects().getSubject().clear(); phoneService2.registerSubjects(subjectForSave); + request.getSubjects().getSubject().clear(); } }, request -> request.getSubjects().getSubject().isEmpty(), diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilderV2.java b/src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilderV2.java index b3f671d..749b564 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilderV2.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilderV2.java @@ -78,8 +78,8 @@ public class PipeLineBuilderV2