From 525de912854c7b184ddc682f3d50c044ffa6ab96 Mon Sep 17 00:00:00 2001 From: ialbert Date: Mon, 10 Nov 2025 16:37:30 +0300 Subject: [PATCH] interrupt --- .../config/task/StageConfig.java | 2 +- .../service/task/PipelinePackageManager.java | 14 ++- .../credit_tracker/service/task/Stage.java | 96 +++++++++---------- .../service/task/impl/KickPipeLineImpl.java | 11 ++- 4 files changed, 63 insertions(+), 60 deletions(-) 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 872a783..0c5e988 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 @@ -20,7 +20,7 @@ public class StageConfig { @Bean("partFileQueueFactory") public StageQueueFactory partFileQueueFactory() { - return new StageQueueFactory<>(10); + return new StageQueueFactory<>(10000); } } 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 bf9ae73..32f3178 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 @@ -30,7 +30,11 @@ public class PipelinePackageManager { private final SaveSubjectHandler saveSubjectHandler; private final FidIdEnricher fidIdEnricherHandler; private final BiFunction> pipelineStarterFactory; - private final ExecutorService executorService = Executors.newFixedThreadPool(10); + private final ExecutorService executorService = Executors.newFixedThreadPool(10, r -> { + Thread t = new Thread(r); + t.setName("pipe-starter-%d".formatted(t.threadId())); + return t; + }); private final Semaphore semaphore; private final static int BATCH_SIZE = 10_000; private final static int MAX_PARALLEL_REQUESTS = 3; @@ -57,9 +61,8 @@ public class PipelinePackageManager { if (!semaphore.tryAcquire()) { throw new RuntimeException("Too many concurrent requests"); } - Pipeline pipeline = null; try { - pipeline = PipeLineBuilderV2.first(saveSubjectQueueFactory.getObject(), idEnricherQueueFactory.getObject()) + Pipeline pipeline = PipeLineBuilderV2.first(saveSubjectQueueFactory.getObject(), idEnricherQueueFactory.getObject()) .handler(saveSubjectHandler) .poison(SubjectPayload.POISON) .andThen() @@ -71,11 +74,14 @@ public class PipelinePackageManager { executorService.submit(pipeline.getStarter()); pipeline.getLast() .doneFuture() - .orTimeout(1, TimeUnit.MINUTES) + .orTimeout(5, TimeUnit.SECONDS) .whenComplete((v, e) -> { if (e != null) { log.error(ExceptionUtils.getStackTrace(e)); + } else { + log.info("packageId={}: pipeline finished", packageId); } + pipeline.getStages().forEach(Stage::close); semaphore.release(); }); } catch (Exception 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 bf4a2b2..32e1ce8 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 @@ -2,15 +2,14 @@ package ru.nbch.credit_tracker.service.task; import java.time.Duration; 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; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.exception.ExceptionUtils; import ru.nbch.credit_tracker.model.StagePayload; @@ -31,7 +30,16 @@ public class Stage implements Ru this.dispatcher = dispatcher; this.in = in; this.out = out; - this.ioPool = Executors.newFixedThreadPool(10); + this.ioPool = new ThreadPoolExecutor(10, + 10, + 0L, TimeUnit.MINUTES, + new LinkedBlockingQueue<>(), + r -> { + Thread t = new Thread(r); + t.setName("pipe-handler-%s-%d".formatted(dispatcher.getClass().getSimpleName(), t.threadId())); + return t; + } + ); this.POISON = POISON; } @@ -46,68 +54,68 @@ public class Stage implements Ru @Override public void run() { List> inflight = new ArrayList<>(); - AtomicInteger inFlightCount = new AtomicInteger(); try { + int counter = 0; while (true) { X b = takeQuiet(in, Duration.ofSeconds(5)); + if (wasInterrupted()) { + break; + } if (b == null) continue; if (b.isPoison()) { // дождаться всех in-flight задач for (Future f : inflight) { + if (wasInterrupted()) { + log.info("stage {} was interrupted while waiting for tasks to finish", + dispatcher.getClass().getSimpleName()); + break; + } f.get(); } putQuiet(out, POISON); + log.info("stage {} finished.", dispatcher.getClass().getSimpleName()); done.complete(null); return; } - + counter++; + int finalCounter = counter; Future f = ioPool.submit(() -> { + if (wasInterrupted()) { + log.info("stage {} task {} interrupted...", dispatcher.getClass().getSimpleName(), finalCounter); + return null; + } + log.info("stage {} task {} started.", dispatcher.getClass().getSimpleName(), finalCounter); Z p = dispatcher.handle(b); putQuiet(out, p); + log.info("stage {} task {} finished.", dispatcher.getClass().getSimpleName(), finalCounter); return p; } ); inflight.add(f); - - // чтобы не накапливать слишком много future в памяти, периодически сливаем - if (inFlightCount.incrementAndGet() % 8 == 0) { - drainDone(inflight, out); - } } } catch (Exception e) { - log.error("Error: {}", ExceptionUtils.getStackTrace(e)); + log.error("error: {}", ExceptionUtils.getStackTrace(e)); putQuiet(out, POISON); done.complete(null); - } finally { - drainAll(inflight, out); } } - void drainDone(List> fs, BlockingQueue out) throws Exception { - Iterator> it = fs.iterator(); - while (it.hasNext()) { - Future f = it.next(); - if (f.isDone()) { - putQuiet(out, f.get()); - it.remove(); - } + private boolean wasInterrupted() { + if (Thread.currentThread().isInterrupted()) { + log.warn("pipeline was interrupted..."); + putQuiet(out, POISON); + done.complete(null); + return true; } + return false; } - void drainAll(List> fs, BlockingQueue out) { - for (Future f : fs) { - try { - putQuiet(out, f.get()); - } catch (Exception e) { - log.error("Error: {}", ExceptionUtils.getStackTrace(e)); - } + public void close() { + log.info("Stage {} is getting shutdown", dispatcher.getClass().getSimpleName()); + ioPool.shutdownNow(); // прерываем обработчики + if (!done.isDone()) { + done.completeExceptionally(new java.util.concurrent.CancellationException("Stage cancelled")); } - fs.clear(); - } - - private ExecutorService newIoExecutor() { - // один carrier на ядро, задачи — виртуальные - return Executors.newThreadPerTaskExecutor(Thread.ofVirtual().name("vt-io-", 0).factory()); } void putQuiet(BlockingQueue q, Z b) { @@ -130,22 +138,4 @@ public class Stage implements Ru public BlockingQueue getIn() { return in; } - - static void sleep(long ms) { - try { - Thread.sleep(ms); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - } - -// @Override - public void stop() { - - } - -// @Override - public boolean isRunning() { - return true; - } } 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 c040e29..e0d8f2d 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 @@ -6,6 +6,7 @@ import java.nio.file.Files; import java.nio.file.Path; 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; @@ -33,6 +34,7 @@ public class KickPipeLineImpl implements IPipeLineStarter { @Override public void startFillingQueue(BlockingQueue firstQueue) { log.info("Start registering subjects. PackageId: {}", packageId); + AtomicInteger counter = new AtomicInteger(0); try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(pkgPath)), 8 * 1024 * 1024)) { xmlProcessor.read(in, MonitoringRequest.class, request -> { @@ -44,8 +46,13 @@ public class KickPipeLineImpl implements IPipeLineStarter { .build() ).collect(Collectors.toList()); - log.trace("Reading subject's batch. Total count: {}. PackageId: {}", request.getSubjects().getSubject().size(), packageId); - firstQueue.add(new SubjectPayload(packageId, subjectProfileBatch)); + 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, subjectProfileBatch)); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } request.getSubjects().getSubject().clear(); } },