From a67cc4056a968011e209f7eabc15f2c7817e3599 Mon Sep 17 00:00:00 2001 From: ialbert Date: Tue, 11 Nov 2025 13:37:00 +0300 Subject: [PATCH] . --- .../BlockingRejectedExecutionHandler.java | 23 ++++ .../service/task/PipelinePackageManager.java | 61 +++------- .../credit_tracker/service/task/Stage.java | 6 +- .../service/task/impl/FlagsEnricherV2.java | 112 ++++++++++++++++++ 4 files changed, 156 insertions(+), 46 deletions(-) create mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/BlockingRejectedExecutionHandler.java create mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricherV2.java diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/BlockingRejectedExecutionHandler.java b/src/main/java/ru/nbch/credit_tracker/service/task/BlockingRejectedExecutionHandler.java new file mode 100644 index 0000000..d239c11 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/BlockingRejectedExecutionHandler.java @@ -0,0 +1,23 @@ +package ru.nbch.credit_tracker.service.task; + +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.RejectedExecutionHandler; +import java.util.concurrent.ThreadPoolExecutor; + +public class BlockingRejectedExecutionHandler implements RejectedExecutionHandler { + + public static RejectedExecutionHandler INSTANCE = new BlockingRejectedExecutionHandler(); + + @Override + public void rejectedExecution(Runnable r, ThreadPoolExecutor exec) { + try { + if (exec.isShutdown()) { + throw new RejectedExecutionException("Executor is shut down"); + } + exec.getQueue().put(r); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RejectedExecutionException("Interrupted while enqueueing task", e); + } + } +} 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 55d7013..01d4264 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,12 +1,10 @@ package ru.nbch.credit_tracker.service.task; -import java.util.concurrent.BlockingQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.function.BiFunction; -import java.util.function.Consumer; import java.util.function.Supplier; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Qualifier; @@ -19,6 +17,7 @@ import ru.nbch.credit_tracker.model.impl.SubjectPayload; import ru.nbch.credit_tracker.service.signal.SignalRestService; import ru.nbch.credit_tracker.service.task.impl.FidIdEnricher; import ru.nbch.credit_tracker.service.task.impl.FlagsEnricher; +import ru.nbch.credit_tracker.service.task.impl.FlagsEnricherV2; import ru.nbch.credit_tracker.service.task.impl.SaveSubjectHandler; import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter; import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator; @@ -41,6 +40,7 @@ public class PipelinePackageManager { private final Supplier> pipelineTerminatorFactory; private final Supplier flagEnricherFactory; private final BiFunction> pipelineStarterFactory; + private final FlagsEnricherV2 flagsEnricherV2; private final ExecutorService executorService = Executors.newFixedThreadPool(10, r -> { Thread t = new Thread(r); t.setName("pipe-starter-%d".formatted(t.threadId())); @@ -60,7 +60,7 @@ public class PipelinePackageManager { SaveSubjectHandler saveSubjectHandler, FidIdEnricher fidIdEnricherHandler, Supplier> pipelineTerminatorFactory, Supplier flagEnricherFactory, - BiFunction> pipelineStarterFactory) { + BiFunction> pipelineStarterFactory, FlagsEnricherV2 flagsEnricherV2) { this.applicationContext = applicationContext; this.xmlProcessor = xmlProcessor; this.signalRestService = signalRestService; @@ -73,6 +73,7 @@ public class PipelinePackageManager { this.pipelineTerminatorFactory = pipelineTerminatorFactory; this.flagEnricherFactory = flagEnricherFactory; this.pipelineStarterFactory = pipelineStarterFactory; + this.flagsEnricherV2 = flagsEnricherV2; this.semaphore = new Semaphore(MAX_PARALLEL_REQUESTS); } @@ -80,19 +81,12 @@ public class PipelinePackageManager { if (!semaphore.tryAcquire()) { throw new RuntimeException("Too many concurrent requests"); } + log.info("throttler permits after acquire: {}", semaphore.availablePermits()); try { - BlockingQueue flagQueue = flagQueueFactory.getObject(); - Consumer consumer = subjectPayload -> { - try { - flagQueue.put(subjectPayload); - } catch (InterruptedException e) { - throw new RuntimeException(e); - } - }; Pipeline pipeline = PipeLineBuilderV2.first(saveSubjectQueueFactory.getObject(), idEnricherQueueFactory.getObject()) .handler(saveSubjectHandler) .poison(SubjectPayload.POISON) - .handlerResult(consumer) + .handlerResult(flagsEnricherV2) .andThen() .stage(sendSignalQueueFactory.getObject()) .handler(fidIdEnricherHandler) @@ -102,10 +96,6 @@ public class PipelinePackageManager { .pollFinishingQueue(pipelineTerminatorFactory.get()) .build(); - FlagsEnricher flagEnricher = flagEnricherFactory.get(); - flagEnricher.setPollingQueue(flagQueue); - - executorService.submit(flagEnricher); executorService.submit(pipeline.getStarter()); executorService.submit(pipeline.getTerminator()); @@ -113,42 +103,25 @@ public class PipelinePackageManager { .doneFuture() .orTimeout(5, TimeUnit.MINUTES) .whenComplete((v, e) -> { - if (e != null) { - log.error(ExceptionUtils.getStackTrace(e)); - } - SignalXmlFilePayload res = (SignalXmlFilePayload) v; - log.trace("Signal file by path: {} is completed", res.getSignalXmlFilePath()); try { + if (e != null) { + log.error(ExceptionUtils.getStackTrace(e)); + return; + } + SignalXmlFilePayload res = (SignalXmlFilePayload) v; + log.trace("Signal file by path: {} is completed", res.getSignalXmlFilePath()); signalRestService.setupMonitoring(res.getSignalXmlFilePath().toString()); } catch (Exception ex) { - throw new RuntimeException(ex); + log.error("pipeline finished with error{}", ExceptionUtils.getStackTrace(ex)); + } finally { + log.info("packageId={}: pipeline finished", packageId); + pipeline.getStages().forEach(Stage::close); + semaphore.release(); } - log.info("packageId={}: pipeline finished", packageId); - pipeline.getStages().forEach(Stage::close); - semaphore.release(); }); - - flagEnricher.doneFuture() - .orTimeout(10, TimeUnit.MINUTES) - .whenComplete((v, e) -> { - if (e != null) { - log.error(ExceptionUtils.getStackTrace(e)); - } - log.debug("Calculation flags completed"); - }); } catch (Exception e) { semaphore.release(); throw e; } - -// try { -// pipeline.getLast().doneFuture().get(5, TimeUnit.SECONDS); -// log.info("Subjects were registered. PackageId: {}", packageId); -// } catch (InterruptedException e) { -// Thread.currentThread().interrupt(); -// return; -// } catch (ExecutionException|TimeoutException e) { -// log.error(ExceptionUtils.getStackTrace(e)); -// } } } 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 bf8d372..8274266 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 @@ -35,12 +35,13 @@ public class Stage implements Ru this.ioPool = new ThreadPoolExecutor(10, 10, 0L, TimeUnit.MINUTES, - new LinkedBlockingQueue<>(), + new LinkedBlockingQueue<>(1), r -> { Thread t = new Thread(r); t.setName("pipe-handler-%s-%d".formatted(dispatcher.getClass().getSimpleName(), t.threadId())); return t; - } + }, + BlockingRejectedExecutionHandler.INSTANCE ); this.POISON = POISON; } @@ -152,4 +153,5 @@ public class Stage implements Ru public void setConsumer(Consumer consumer) { this.consumer = consumer; } + } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricherV2.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricherV2.java new file mode 100644 index 0000000..4c7f4fe --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricherV2.java @@ -0,0 +1,112 @@ +package ru.nbch.credit_tracker.service.task.impl; + +import com.fasterxml.jackson.annotation.JsonInclude; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +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 com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.SerializationFeature; +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; +import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper; +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.task.BlockingRejectedExecutionHandler; + +@Slf4j +@Service +public class FlagsEnricherV2 implements Consumer { + + private final FlagsService2 flagsService; + private final PhoneFidMapMapper phoneFidMapMapper; + private final ExecutorService ioPool; + private final ObjectMapper mapper = new ObjectMapper() + .registerModule(new JavaTimeModule()) + .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS) + .setSerializationInclusion(JsonInclude.Include.NON_NULL); + + public FlagsEnricherV2(FlagsService2 flagsService, + PhoneFidMapMapper phoneFidMapMapper) { + this.flagsService = flagsService; + this.phoneFidMapMapper = phoneFidMapMapper; + this.ioPool = new ThreadPoolExecutor(10, + 10, + 0L, TimeUnit.MINUTES, + new LinkedBlockingQueue<>(100), + r -> { + Thread t = new Thread(r); + t.setName("flags-enricher-%d".formatted(t.threadId())); + return t; + }, + BlockingRejectedExecutionHandler.INSTANCE + ); + } + + public void process(SubjectPayload subjectPayload) { + List subjects = subjectPayload.getSubjects(); + log.debug("FlagsEnricher handle by packageId: {}, size: {}", subjectPayload.getPackageId(), subjects.size()); + + List fids = subjects.stream() + .map(SubjectProfile::getFid) + .filter(Objects::nonNull) + .toList(); + LocalDate yearAgo = LocalDate.now().minusYears(1); + + Map flagsDebt5000 = flagsService.findFlagsDebt5000(fids); + Map flag90ByPastDueArrear = flagsService.findFlag90ByPastDueArrear(fids, yearAgo); + Map flag90ByRuPaymt = flagsService.findFlag90ByRuPaymt(fids, yearAgo); + + String json; + try { + json = mapper.writeValueAsString(subjects.stream().map(subject -> { + Integer debt5000 = null; + Integer flagDays90 = null; + LocalDateTime calculatedAt = null; + + if (subject.getFid() != null) { + debt5000 = flagsDebt5000.getOrDefault(subject.getFid(), 0); + int pastDue90 = flag90ByPastDueArrear.getOrDefault(subject.getFid(), 0); + int paymt90 = flag90ByRuPaymt.getOrDefault(subject.getFid(), 0); + flagDays90 = Math.max(pastDue90, paymt90); + calculatedAt = LocalDateTime.now(); + } + + Integer finalDebt500 = debt5000; + Integer finalFlagDays9 = flagDays90; + LocalDateTime finalCalculatedAt = calculatedAt; + return new HashMap() {{ + put("subject_id", subject.getIdentifier()); + put("fid", subject.getFid()); + put("phone", subject.getPhone()); + put("package_id", subjectPayload.getPackageId()); + put("flag_debt5000", finalDebt500); + put("flag_90_12", finalFlagDays9); + put("calculated_at", finalCalculatedAt); + }}; + }).toList()); + } catch (JsonProcessingException e) { + throw new RuntimeException(e); + } + phoneFidMapMapper.registerJson(json); + } + + @Override + public void accept(SubjectPayload subjectPayload) { + ThreadPoolExecutor casted = (ThreadPoolExecutor) ioPool; + log.info("submitting payload. active tasks number: {}; queue size: {}", casted.getActiveCount(), casted.getQueue().size()); + this.ioPool.submit(() -> { + process(subjectPayload); + }); + } +}