From 9813fe40538e99142076ac2dd7b3fdf250647073 Mon Sep 17 00:00:00 2001 From: ialbert Date: Wed, 5 Nov 2025 13:58:28 +0300 Subject: [PATCH] pipeline futures --- .../credit_tracker/service/task/Pipeline.java | 31 +++++++++++ .../service/task/PipelinePackageManager.java | 55 +++++++++++++++---- .../credit_tracker/service/task/Stage.java | 8 +++ .../service/task/setup/PipeLineBuilderV2.java | 10 +++- .../utils/error/ExceptionUtils.java | 9 +++ 5 files changed, 99 insertions(+), 14 deletions(-) create mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/Pipeline.java diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/Pipeline.java b/src/main/java/ru/nbch/credit_tracker/service/task/Pipeline.java new file mode 100644 index 0000000..f90cd13 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/Pipeline.java @@ -0,0 +1,31 @@ +package ru.nbch.credit_tracker.service.task; + +import java.util.List; + +public class Pipeline { + private Runnable starter; + private List> stages; + + public Pipeline() { + } + + public Stage getLast() { + return stages.getLast(); + } + + public Runnable getStarter() { + return starter; + } + + public void setStarter(Runnable starter) { + this.starter = starter; + } + + public List> getStages() { + return stages; + } + + public void setStages(List> stages) { + this.stages = stages; + } +} 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 4d5edf8..634478e 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,8 @@ import java.util.concurrent.BlockingQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; import java.util.function.BiFunction; import java.util.stream.Collectors; import lombok.extern.slf4j.Slf4j; @@ -25,6 +27,7 @@ import ru.nbch.credit_tracker.service.task.impl.dprctd.FidIdEnricher; import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter; import ru.nbch.credit_tracker.service.task.setup.PipeLineBuilderV2; import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; +import ru.nbch.credit_tracker.utils.error.ExceptionUtils; @Component @Slf4j @@ -38,7 +41,9 @@ public class PipelinePackageManager { private final FidIdEnricher fidIdEnricherHandler; private final BiFunction> pipelineStarterFactory; private final ExecutorService executorService = Executors.newFixedThreadPool(10); + private final Semaphore semaphore; private final static int BATCH_SIZE = 10_000; + private final static int MAX_PARALLEL_REQUESTS = 3; public PipelinePackageManager(ApplicationContext applicationContext, StaxXmlProcessor xmlProcessor, @@ -55,6 +60,7 @@ public class PipelinePackageManager { this.saveSubjectHandler = saveSubjectHandler; this.fidIdEnricherHandler = fidIdEnricherHandler; this.pipelineStarterFactory = pipelineStarterFactory; + this.semaphore = new Semaphore(MAX_PARALLEL_REQUESTS); } public void buildAndRun(Long packageId, String pkgPath) throws Exception { @@ -100,16 +106,43 @@ public class PipelinePackageManager { } public void buildAndRun2(Long packageId, String pkgPath) throws Exception { - var startPipeLine = PipeLineBuilderV2.first(saveSubjectQueueFactory.getObject(), idEnricherQueueFactory.getObject()) - .handler(saveSubjectHandler) - .poison(SubjectPayload.POISON) - .andThen() - .stage(sendSignalQueueFactory.getObject()) - .handler(fidIdEnricherHandler) - .poison(SendSignalsPayload.POISON) - .andThen() - .fillInitialQueue(pipelineStarterFactory.apply(packageId, pkgPath)); - executorService.submit(startPipeLine); - log.info("Subjects were registered. PackageId: {}", packageId); + if (!semaphore.tryAcquire()) { + throw new RuntimeException("Too many concurrent requests"); + } + Pipeline pipeline = null; + try { + pipeline = PipeLineBuilderV2.first(saveSubjectQueueFactory.getObject(), idEnricherQueueFactory.getObject()) + .handler(saveSubjectHandler) + .poison(SubjectPayload.POISON) + .andThen() + .stage(sendSignalQueueFactory.getObject()) + .handler(fidIdEnricherHandler) + .poison(SendSignalsPayload.POISON) + .andThen() + .fillInitialQueue(pipelineStarterFactory.apply(packageId, pkgPath)); + executorService.submit(pipeline.getStarter()); + pipeline.getLast() + .doneFuture() + .orTimeout(1, TimeUnit.MINUTES) + .whenComplete((v, e) -> { + if (e != null) { + log.error(ExceptionUtils.getStackTrace(e)); + } + semaphore.release(); + }); + } 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 ed6df42..bf4a2b2 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 @@ -5,6 +5,7 @@ import java.util.ArrayList; import java.util.Iterator; import java.util.List; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; @@ -21,6 +22,7 @@ public class Stage implements Ru private final BlockingQueue out; private final Z POISON; private final ExecutorService ioPool; + private final CompletableFuture done = new CompletableFuture<>(); public Stage(Handler dispatcher, BlockingQueue in, @@ -33,6 +35,10 @@ public class Stage implements Ru this.POISON = POISON; } + public CompletableFuture doneFuture() { + return done; + } + public void startRunning() { ioPool.submit(this); } @@ -51,6 +57,7 @@ public class Stage implements Ru f.get(); } putQuiet(out, POISON); + done.complete(null); return; } @@ -70,6 +77,7 @@ public class Stage implements Ru } catch (Exception e) { log.error("Error: {}", ExceptionUtils.getStackTrace(e)); putQuiet(out, POISON); + done.complete(null); } finally { drainAll(inflight, out); } 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 daec97f..dfb6d33 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 @@ -5,6 +5,7 @@ import java.util.List; import java.util.Objects; import java.util.concurrent.BlockingQueue; import ru.nbch.credit_tracker.model.StagePayload; +import ru.nbch.credit_tracker.service.task.Pipeline; import ru.nbch.credit_tracker.service.task.Stage; import ru.nbch.credit_tracker.service.task.setup.step.HandlerStep; @@ -34,15 +35,18 @@ public class PipeLineBuilderV2 pipeLineStarter) { + public Pipeline fillInitialQueue(IPipeLineStarter pipeLineStarter) { Objects.requireNonNull(pipeLineStarter, "pipeLineStarter is required"); if (firstStage == null) { throw new IllegalStateException("Pipeline has no stages"); } - return () -> { + Pipeline p = new Pipeline(); + p.setStarter(() -> { stages.forEach(Stage::startRunning); pipeLineStarter.startFillingQueue(firstStage.getIn()); - }; + }); + p.setStages(stages); + return p; } @SuppressWarnings("unchecked") diff --git a/src/main/java/ru/nbch/credit_tracker/utils/error/ExceptionUtils.java b/src/main/java/ru/nbch/credit_tracker/utils/error/ExceptionUtils.java index af7969e..05e71b6 100644 --- a/src/main/java/ru/nbch/credit_tracker/utils/error/ExceptionUtils.java +++ b/src/main/java/ru/nbch/credit_tracker/utils/error/ExceptionUtils.java @@ -1,5 +1,7 @@ package ru.nbch.credit_tracker.utils.error; +import java.io.PrintWriter; +import java.io.StringWriter; import ru.nbch.credit_tracker.enums.EnumKeyMessage; import ru.nbch.credit_tracker.enums.IEnumKeyCode; import ru.nbch.credit_tracker.enums.XmlErrorFormat; @@ -8,6 +10,13 @@ import ru.nbch.credit_tracker.exceptions.SignalServerException; public class ExceptionUtils { + public static String getStackTrace(Throwable throwable) { + StringWriter sw = new StringWriter(); + PrintWriter pw = new PrintWriter(sw, true); + throwable.printStackTrace(pw); + return sw.getBuffer().toString(); + } + public static SignalClientException clientExcp(IEnumKeyCode enumKeyCode, XmlErrorFormat xmlErrorFormat, Object... args) { return clientExcp(enumKeyCode, null, xmlErrorFormat, args); }