From 51e6bb05c9923aaff911eb90fb7f83358c8a06b4 Mon Sep 17 00:00:00 2001 From: ialbert Date: Wed, 3 Dec 2025 17:12:25 +0300 Subject: [PATCH] pipeline context example --- .../credit_tracker/model/PipelineContext.java | 21 +++++++++++++++++++ .../credit_tracker/model/StagePayload.java | 2 ++ .../model/impl/PackageVariable.java | 5 +++++ .../model/impl/SubjectPayload.java | 12 +++++++++++ .../credit_tracker/service/task/Handler.java | 12 ++++++++++- .../service/task/PipelinePackageFabric.java | 2 ++ .../credit_tracker/service/task/Stage.java | 14 +++++++++++++ .../service/task/impl/FidNameEnricher.java | 5 ++++- .../service/task/impl/KickPipeLine.java | 14 ++++++++++++- .../service/task/setup/IPipeLineStarter.java | 3 +++ .../service/task/setup/PipeLineBuilderV2.java | 10 +++++++++ 11 files changed, 97 insertions(+), 3 deletions(-) create mode 100644 src/main/java/ru/nbch/credit_tracker/model/PipelineContext.java create mode 100644 src/main/java/ru/nbch/credit_tracker/model/impl/PackageVariable.java diff --git a/src/main/java/ru/nbch/credit_tracker/model/PipelineContext.java b/src/main/java/ru/nbch/credit_tracker/model/PipelineContext.java new file mode 100644 index 0000000..286619b --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/model/PipelineContext.java @@ -0,0 +1,21 @@ +package ru.nbch.credit_tracker.model; + +import java.util.EnumMap; + +public class PipelineContext> { + private final EnumMap storage; + + public PipelineContext(Class keyClass) { + this.storage = new EnumMap<>(keyClass); + } + + public void store(T key, Object value) { + storage.put(key, value); + } + + @SuppressWarnings("unchecked") + public S get(T key) { + Object o = storage.get(key); + return (S) o; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/model/StagePayload.java b/src/main/java/ru/nbch/credit_tracker/model/StagePayload.java index 5fbd1e1..0dc7b42 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/StagePayload.java +++ b/src/main/java/ru/nbch/credit_tracker/model/StagePayload.java @@ -2,4 +2,6 @@ package ru.nbch.credit_tracker.model; public interface StagePayload { boolean isPoison(); + default void setContext(PipelineContext context) {} + default PipelineContext getContext() {return null;} } diff --git a/src/main/java/ru/nbch/credit_tracker/model/impl/PackageVariable.java b/src/main/java/ru/nbch/credit_tracker/model/impl/PackageVariable.java new file mode 100644 index 0000000..9482096 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/model/impl/PackageVariable.java @@ -0,0 +1,5 @@ +package ru.nbch.credit_tracker.model.impl; + +public enum PackageVariable { + packageId, +} diff --git a/src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java b/src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java index bd68f1b..4eaf567 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java +++ b/src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java @@ -2,6 +2,7 @@ package ru.nbch.credit_tracker.model.impl; import java.util.List; import ru.nbch.credit_tracker.enums.PoisonType; +import ru.nbch.credit_tracker.model.PipelineContext; import ru.nbch.credit_tracker.model.StagePayload; import ru.nbch.credit_tracker.model.SubjectProfile; @@ -11,6 +12,7 @@ public class SubjectPayload implements StagePayload { private final String memberCode; private final String packageName; private final List subjects; + private PipelineContext ctx; public static final SubjectPayload POISON = new SubjectPayload(null, PoisonType.NORMALLY); private final PoisonType poison; @@ -30,6 +32,16 @@ public class SubjectPayload implements StagePayload { this.poison = PoisonType.NONE; } + @Override + public void setContext(PipelineContext context) { + this.ctx = (PipelineContext) context; + } + + @Override + public PipelineContext getContext() { + return this.ctx; + } + public Long getPackageId() { return packageId; } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/Handler.java b/src/main/java/ru/nbch/credit_tracker/service/task/Handler.java index 4b7d615..2ba0562 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/Handler.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/Handler.java @@ -1,7 +1,17 @@ package ru.nbch.credit_tracker.service.task; -public interface Handler { +import ru.nbch.credit_tracker.model.PipelineContext; +import ru.nbch.credit_tracker.model.StagePayload; + +public interface Handler { Z handle(X taskPayload); + + @SuppressWarnings("unchecked") + default > PipelineContext obtainContext(Class keyClass, X input) { + PipelineContext ctx = input.getContext(); + return (PipelineContext) ctx; + } + default void handleOnPoison(X poison){ //do nothing } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java index 1bba46f..cd3d596 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java @@ -8,6 +8,7 @@ import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.stereotype.Component; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.config.task.StageQueueFactory; +import ru.nbch.credit_tracker.model.impl.PackageVariable; import ru.nbch.credit_tracker.model.impl.PartFilePayload; import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload; import ru.nbch.credit_tracker.model.impl.SubjectPayload; @@ -64,6 +65,7 @@ public class PipelinePackageFabric implements InitializingBean { .fillInitialQueue(pipelineStarterFactory.apply(setupParam)) .pollFinishingQueue(pipelineTerminatorFactory.get()) .backoff(pipelineBackoff) + .context(PackageVariable.class) .build(); } 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 44d1f40..be6155e 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 @@ -12,6 +12,7 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.exception.ExceptionUtils; +import ru.nbch.credit_tracker.model.PipelineContext; import ru.nbch.credit_tracker.model.StagePayload; import ru.nbch.credit_tracker.utils.LogUtil; import ru.nbch.credit_tracker.utils.TaskTimer; @@ -30,6 +31,7 @@ public class Stage implements Ru private static final ThreadPoolExecutor.CallerRunsPolicy callerRunsPolicy = new ThreadPoolExecutor.CallerRunsPolicy(); + private PipelineContext context; public Stage(Handler dispatcher, BlockingQueue in, @@ -106,7 +108,11 @@ public class Stage implements Ru log.info("stage {} task {} started.", dispatcher.getClass().getSimpleName(), finalCounter); Z p; try (var m = stopWatch.start(dispatcherName + "-handler")) { + if (b.getContext() == null) { + b.setContext(context); + } p = dispatcher.handle(b); + p.setContext(b.getContext()); // нужно т.к. в конце придет в терминатор } if (Thread.currentThread().isInterrupted()) { log.info("after {} task {} finished, worker has been interrupted. not putting result in queue.", @@ -210,4 +216,12 @@ public class Stage implements Ru public void setBackoff(PipelineBackoff pipelineBackoff) { this.pipelineBackoff = pipelineBackoff; } + + public void setContext(PipelineContext context) { + this.context = context; + } + + public PipelineContext getContext() { + return context; + } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameEnricher.java index 54b1029..be9ce06 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameEnricher.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameEnricher.java @@ -10,6 +10,7 @@ import ru.nbch.credit_tracker.entities.app.PhoneFidMap; import ru.nbch.credit_tracker.entities.indic.Name; import ru.nbch.credit_tracker.model.NameData; import ru.nbch.credit_tracker.model.SubjectProfile; +import ru.nbch.credit_tracker.model.impl.PackageVariable; import ru.nbch.credit_tracker.model.impl.SubjectPayload; import ru.nbch.credit_tracker.service.indic.PhoneService2; import ru.nbch.credit_tracker.service.task.Handler; @@ -35,7 +36,9 @@ public class FidNameEnricher implements Handler @Override public SubjectPayload handle(SubjectPayload taskPayload) { List subjects = taskPayload.getSubjects(); - log.debug("NameEnricherStage handle by packageId: {}, size of bucket for save: {}", taskPayload.getPackageId(), subjects.size()); + var ctx = obtainContext(PackageVariable.class, taskPayload); + Long packageId = ctx.get(PackageVariable.packageId); + log.debug("NameEnricherStage handle by packageId: {}, size of bucket for save: {}", packageId, subjects.size()); if (subjects.getFirst().getFid() == null) { processByNumber(subjects); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLine.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLine.java index cba791f..da367ed 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLine.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLine.java @@ -12,7 +12,9 @@ import ru.nbch.credit_tracker.enums.PoisonType; import ru.nbch.credit_tracker.exceptions.SignalClientException; import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper; import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; +import ru.nbch.credit_tracker.model.PipelineContext; import ru.nbch.credit_tracker.model.SubjectProfile; +import ru.nbch.credit_tracker.model.impl.PackageVariable; import ru.nbch.credit_tracker.model.impl.SubjectPayload; import ru.nbch.credit_tracker.service.task.SetupParam; import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter; @@ -30,6 +32,7 @@ public abstract class KickPipeLine implements IPipeLineStarter { private final int BATCH_SIZE; private boolean isInterrupted = false; protected int offset = 0; + private PipelineContext context; public KickPipeLine(StaxXmlProcessor xmlProcessor, @@ -52,10 +55,17 @@ public abstract class KickPipeLine implements IPipeLineStarter { } } + @SuppressWarnings("unchecked") + @Override + public void setContext(PipelineContext context) { + this.context = (PipelineContext) context; + } + @Override public void startFillingQueue(BlockingQueue firstQueue) { try { log.info("Start kicker: mode:{}, packageId: {};", mode(), packageId); + context.store(PackageVariable.packageId, packageId); int limit = BATCH_SIZE; UserPackage userPackage = userPackageMapper.findPackage(packageId); List page = nextPage(limit); @@ -69,7 +79,9 @@ public abstract class KickPipeLine implements IPipeLineStarter { .phone(phoneFidMap.getPhone()) .build()).toList(); try { - firstQueue.put(new SubjectPayload(packageId, userPackage.getUserId(), userPackage.getPackageName(), subjectProfiles)); + SubjectPayload payload = new SubjectPayload(packageId, userPackage.getUserId(), userPackage.getPackageName(), subjectProfiles); + payload.setContext(context); + firstQueue.put(payload); } catch (InterruptedException e) { isInterrupted = true; log.error("Pipeline starter was interrupted!"); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/setup/IPipeLineStarter.java b/src/main/java/ru/nbch/credit_tracker/service/task/setup/IPipeLineStarter.java index a4b48fa..48a96e7 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/setup/IPipeLineStarter.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/setup/IPipeLineStarter.java @@ -1,8 +1,11 @@ package ru.nbch.credit_tracker.service.task.setup; import java.util.concurrent.BlockingQueue; +import ru.nbch.credit_tracker.model.PipelineContext; import ru.nbch.credit_tracker.model.StagePayload; public interface IPipeLineStarter { void startFillingQueue(BlockingQueue queue); + + default void setContext(PipelineContext context) {} } 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 cd03e2f..2fdff5a 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 @@ -4,6 +4,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Objects; import java.util.concurrent.BlockingQueue; +import ru.nbch.credit_tracker.model.PipelineContext; import ru.nbch.credit_tracker.model.StagePayload; import ru.nbch.credit_tracker.service.task.Pipeline; import ru.nbch.credit_tracker.service.task.PipelineBackoff; @@ -17,6 +18,7 @@ public class PipeLineBuilderV2 pipeLineStarter; private IPipeLineTerminator pipeLineTerminator; private PipelineBackoff backoff; + private PipelineContext context; private PipeLineBuilderV2() { this.stages = new ArrayList<>(); @@ -60,14 +62,22 @@ public class PipeLineBuilderV2> + PipeLineBuilderV2 context(Class contextKeyClass) { + this.context = new PipelineContext<>(contextKeyClass); + return this; + } + public Pipeline build() { Pipeline p = new Pipeline(); p.setStarter(() -> { stages.forEach(stage -> { stage.setPipeline(p); stage.setBackoff(backoff); + stage.setContext(context); stage.startRunning(); }); + pipeLineStarter.setContext(context); pipeLineStarter.startFillingQueue(firstStage.getIn()); }); pipeLineTerminator.setPollingQueue(lastQueue);