From 4633a83a6485fc38e05a8f8fe45b92248afcf870 Mon Sep 17 00:00:00 2001 From: Mikhail Trofimov Date: Mon, 22 Jun 2026 16:22:22 +0300 Subject: [PATCH] Moved RotateSheduler as separate component --- .../monitoring/CTMonitoringApplication.java | 3 - .../monitoring/KafkaMonitoringService.java | 38 +--------- .../mode/monitoring/RotateSheduler.java | 74 +++++++++++++++++++ .../SignalKafkaEventWriterService.java | 31 ++++++++ 4 files changed, 106 insertions(+), 40 deletions(-) create mode 100644 src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/RotateSheduler.java diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/CTMonitoringApplication.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/CTMonitoringApplication.java index d3e8c1a..772bf76 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/CTMonitoringApplication.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/CTMonitoringApplication.java @@ -28,8 +28,5 @@ public class CTMonitoringApplication { ConfigurableApplicationContext ctx = application.run(args); // MonitoringService monitoringService = ctx.getBean(MonitoringService.class); // monitoringService.run(); - - SignalKafkaEventWriterService signalKafkaEventWriterService = ctx.getBean(SignalKafkaEventWriterService.class); - signalKafkaEventWriterService.startWriting(); } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java index d9b9d7a..6418209 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java @@ -1,13 +1,11 @@ package ru.nbch.credit_tracker.service.mode.monitoring; -import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.support.Acknowledgment; -import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.entities.app.PhonesMap; @@ -23,19 +21,12 @@ import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.BlockingQueue; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; @Slf4j @Service public class KafkaMonitoringService { - - private final ScheduledExecutorService rotateScheduler = Executors.newSingleThreadScheduledExecutor( - new CustomizableThreadFactory("monitoring-rotate-") - ); private volatile boolean isShuttingdDown = false; private final AtomicInteger activeProcessCount = new AtomicInteger(0); @@ -157,25 +148,8 @@ public class KafkaMonitoringService { } } - /** - * Таймер, который отсчитывает {@code monitoring-interval:5} минут с момента начала накопления сигналов. - * По истечении таймера отправляет Event на ROTATE файлов - */ - @PostConstruct - public void startRotateTimer() { - long period = config.getMonitoringInterval() != null ? config.getMonitoringInterval() : 5; - rotateScheduler.scheduleAtFixedRate(() -> { - if (isShuttingdDown) { - return; - } - SignalKafkaEvent rotateEvent = new SignalKafkaEvent(); - rotateEvent.setEventType(EventType.ROTATE); - putToQueue(rotateEvent); - }, 0, period, TimeUnit.MINUTES); - } - @PreDestroy - public void stopRotateTimer() { + public void stopKafkaMonitoringService() { log.info("Stopping KafkaMonitoringService..."); isShuttingdDown = true; @@ -203,16 +177,6 @@ public class KafkaMonitoringService { Thread.currentThread().interrupt(); } - rotateScheduler.shutdown(); - try { - if (!rotateScheduler.awaitTermination(5, TimeUnit.SECONDS)) { - rotateScheduler.shutdownNow(); - } - } catch (InterruptedException e) { - rotateScheduler.shutdownNow(); - Thread.currentThread().interrupt(); - } - log.info("KafkaMonitoringService stopped"); } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/RotateSheduler.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/RotateSheduler.java new file mode 100644 index 0000000..7ec5f73 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/RotateSheduler.java @@ -0,0 +1,74 @@ +package ru.nbch.credit_tracker.service.mode.monitoring; + +import jakarta.annotation.PostConstruct; +import jakarta.annotation.PreDestroy; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.scheduling.concurrent.CustomizableThreadFactory; +import org.springframework.stereotype.Component; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.enums.EventType; +import ru.nbch.credit_tracker.model.SignalKafkaEvent; + +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +@Slf4j +@Component +public class RotateSheduler { + private final ScheduledExecutorService rotateScheduler = Executors.newSingleThreadScheduledExecutor( + new CustomizableThreadFactory("monitoring-rotate-") + ); + private volatile boolean isShuttingdDown = false; + + private final BlockingQueue signalEventsQueue; + private final CreditTrackerSettings config; + + public RotateSheduler(@Qualifier("signalKafkaEventQueue") BlockingQueue signalEventsQueue, + CreditTrackerSettings config) { + this.signalEventsQueue = signalEventsQueue; + this.config = config; + } + + /** + * Таймер, который отсчитывает {@code monitoring-interval:5} минут с момента начала накопления сигналов. + * По истечении таймера отправляет Event на ROTATE файлов + */ + @PostConstruct + public void startRotateTimer() { + long period = config.getMonitoringInterval() != null ? config.getMonitoringInterval() : 5; + rotateScheduler.scheduleAtFixedRate(() -> { + if (isShuttingdDown) { + return; + } + try { + SignalKafkaEvent rotateEvent = new SignalKafkaEvent(); + rotateEvent.setEventType(EventType.ROTATE); + signalEventsQueue.put(rotateEvent); + } catch (InterruptedException e) { + log.error("Error while putting signal event", e); + Thread.currentThread().interrupt(); + } + }, 0, period, TimeUnit.MINUTES); + } + + @PreDestroy + public void stopRotateTimer() { + log.info("Stopping RotateSheduler..."); + isShuttingdDown = true; + + rotateScheduler.shutdown(); + try { + if (!rotateScheduler.awaitTermination(5, TimeUnit.SECONDS)) { + rotateScheduler.shutdownNow(); + } + } catch (InterruptedException e) { + rotateScheduler.shutdownNow(); + Thread.currentThread().interrupt(); + } + + log.info("RotateSheduler stopped"); + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java index 9f75b0e..cd7073c 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java @@ -1,7 +1,10 @@ package ru.nbch.credit_tracker.service.mode.monitoring; +import jakarta.annotation.PostConstruct; +import jakarta.annotation.PreDestroy; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.entities.app.UserPackage; @@ -16,10 +19,16 @@ import java.nio.file.Files; import java.nio.file.Path; import java.util.List; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; @Slf4j @Service public class SignalKafkaEventWriterService { + private final ExecutorService executorService = Executors.newSingleThreadExecutor( + new CustomizableThreadFactory("signal-kafka-event-writer-") + ); private final BlockingQueue signalEventsQueue; private final UserPackageMapper userPackageMapper; @@ -33,6 +42,11 @@ public class SignalKafkaEventWriterService { this.config = config; } + @PostConstruct + public void start() { + executorService.submit(this::startWriting); + } + public void startWriting() { while (true) { SignalKafkaEvent event = take(); @@ -103,4 +117,21 @@ public class SignalKafkaEventWriterService { return null; } } + + @PreDestroy + public void stop() { + log.info("Stopping SignalKafkaEventWriterService..."); + + executorService.shutdown(); + try { + if (!executorService.awaitTermination(5, TimeUnit.MINUTES)) { + executorService.shutdownNow(); + } + } catch (InterruptedException e) { + executorService.shutdownNow(); + Thread.currentThread().interrupt(); + } + + log.info("SignalKafkaEventWriterService stopped"); + } }