treshenkov/credit_tracker
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

.

Browse source
This commit is contained in:
ialbert 2025-11-25 15:44:08 +03:00
parent c65d88f645
commit 015c7ca24b
5 changed files with 25 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
src/main/java/ru/nbch/credit_tracker/config/settings/PipelineSettings.java
View file

@ -6,6 +6,7 @@ import org.springframework.boot.context.properties.bind.DefaultValue;
public record PipelineSettings(
@DefaultValue("5000") Integer subjectBatchSize,
@DefaultValue("3") Integer saveSubjectStageConcurrency,
@DefaultValue("15") Integer nameEnricherStageConcurrency,
@DefaultValue("10") Integer idEnricherStageConcurrency,
@DefaultValue("5") Integer errorAttemptCount,

4
src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java
View file

@ -79,6 +79,7 @@ public class PipelinePackageManager {
this.fidNameEnricher = fidNameEnricher;
this.creditTrackerSettings = creditTrackerSettings;
this.semaphore = new Semaphore(creditTrackerSettings.getPipeline().parallelRequests());
log.info("pipeline settings: {}", creditTrackerSettings.getPipeline());
this.pipelineBackoff = pipelineBackoff;
}
@ -87,6 +88,7 @@ public class PipelinePackageManager {
throw new RuntimeException("Too many concurrent requests");
}
log.info("throttler permits after acquire: {}", semaphore.availablePermits());
long startTime = System.currentTimeMillis();
try {
Pipeline pipeline = PipeLineBuilderV2.first(saveSubjectQueueFactory.getObject(), idEnricherQueueFactory.getObject())
.handler(saveSubjectHandler)
@ -110,6 +112,8 @@ public class PipelinePackageManager {
.orTimeout(creditTrackerSettings.getPipeline().globalTimeout().toMillis(), TimeUnit.MILLISECONDS)
.whenComplete((v, e) -> {
try {
long finishTime = System.currentTimeMillis();
log.info("pipeline time passed: {} ms", finishTime - startTime);
if (e instanceof XmlMergeCoordinator.PipelineTerminatorException) {
log.info("when complete: terminator was interrupted.");
return;

19
src/main/java/ru/nbch/credit_tracker/service/task/Stage.java
View file

@ -5,7 +5,7 @@ import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@ -13,8 +13,8 @@ import java.util.function.Consumer;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import ru.nbch.credit_tracker.model.StagePayload;
import ru.nbch.credit_tracker.utils.TaskTimer;
import ru.nbch.credit_tracker.utils.LogUtil;
import ru.nbch.credit_tracker.utils.TaskTimer;
@Slf4j
public class Stage<X extends StagePayload, Z extends StagePayload> implements Runnable {
@ -42,10 +42,11 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
this.out = out;
this.dispatcherName = dispatcher.getClass().getSimpleName();
var mdcMap = LogUtil.extractMdcMap();
BlockingQueue<Runnable> handoffQueue = new SynchronousQueue<>();
this.ioPool = new ThreadPoolExecutor(dispatcher.preferredConcurrency(),
dispatcher.preferredConcurrency(),
0L, TimeUnit.MINUTES,
new LinkedBlockingQueue<>(1),
handoffQueue,
r -> {
Thread t = new Thread(LogUtil.wrap(mdcMap, r));
t.setName("pipe-handler-%s-%d".formatted(dispatcherName, t.threadId()));
@ -53,7 +54,13 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
},
(r, executor) -> {
log.info("stage {}: all threads are working; switch to CallerRunsPolicy", dispatcherName);
callerRunsPolicy.rejectedExecution(r, executor);
try {
//блокировать отправителя, пока какой‑то рабочий поток не возьмёт задачу
handoffQueue.put(r);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
// callerRunsPolicy.rejectedExecution(r, executor);
}
);
}
@ -79,7 +86,6 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
}
f.get();
}
log.trace("Stage time stats: {}", stopWatch.info());
dispatcher.handleOnPoison(b);
putQuiet(out, dispatcher.cookPoison());
log.info("stage {} finished.", dispatcherName);
@ -97,7 +103,7 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
}
log.info("stage {} task {} started.", dispatcher.getClass().getSimpleName(), finalCounter);
Z p;
try (var m = stopWatch.start(dispatcherName)){
try (var m = stopWatch.start(dispatcherName + "-handler")){
p = dispatcher.handle(b);
}
if (Thread.currentThread().isInterrupted()) {
@ -159,6 +165,7 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
public void close() {
log.info("Stage {} is getting shutdown", dispatcher.getClass().getSimpleName());
log.trace("Stage time stats: {}", stopWatch.info());
ioPool.shutdownNow(); // прерываем обработчики
}

2
src/main/java/ru/nbch/credit_tracker/service/task/impl/SaveSubjectHandler.java
View file

@ -57,7 +57,7 @@ public class SaveSubjectHandler implements Handler<SubjectPayload, SubjectPayloa
@Override
public int preferredConcurrency() {
return 20;
return creditTrackerSettings.getPipeline().saveSubjectStageConcurrency();
}
@Override

5
src/main/java/ru/nbch/credit_tracker/service/task/impl/XmlMergeCoordinator.java
View file

@ -21,6 +21,7 @@ import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import ru.nbch.credit_tracker.model.impl.PartFilePayload;
import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload;
import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator;
import ru.nbch.credit_tracker.utils.TaskTimer;
@Slf4j
public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload> {
@ -29,6 +30,7 @@ public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePa
private Path out;
private final CreditTrackerSettings config;
private BlockingQueue<PartFilePayload> inQueue;
private final TaskTimer stopWatch = new TaskTimer();
public XmlMergeCoordinator(CreditTrackerSettings config) {
this.config = config;
@ -45,9 +47,12 @@ public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePa
if (part.isPoison()) {
log.debug("Got poison");
done.complete(new SignalXmlFilePayload(out));
log.info("Stage time stats: {}", stopWatch.info());
return;
}
try (var m = stopWatch.start("XmlMergeCoordinator")) {
mergeBatch(part);
}
} catch (InterruptedException ie) {
log.info("pipeline terminator was interrupted!");
done.completeExceptionally(new PipelineTerminatorException());