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 6a32645..7f48c21 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,18 @@ package ru.nbch.credit_tracker.service.mode.monitoring; import jakarta.annotation.PreDestroy; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.Semaphore; +import java.util.stream.Collectors; 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.stereotype.Service; -import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.entities.app.PhonesMap; import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage; import ru.nbch.credit_tracker.enums.EventType; @@ -18,18 +23,13 @@ import ru.nbch.credit_tracker.service.signal.PackageRecordsService; import ru.nbch.credit_tracker.service.signal.SegmentMapService; import ru.nbch.credit_tracker.utils.CTUtil; -import java.util.List; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.Semaphore; -import java.util.stream.Collectors; - @Slf4j @Service public class KafkaMonitoringService { private volatile boolean isShuttingdDown = false; - private final Semaphore processingPermits = new Semaphore(8); // количество равно concurency листенера + private final static String CONCURRENCY = "8"; + private final static int PERMITS = Integer.parseInt(CONCURRENCY); + private final Semaphore processingPermits = new Semaphore(PERMITS); // количество равно concurency листенера private final BlockingQueue signalEventsQueue; @@ -37,37 +37,33 @@ public class KafkaMonitoringService { private final SourceService sourceService; private final SegmentMapService segmentMapService; - private final CreditTrackerSettings config; - public KafkaMonitoringService(@Qualifier("signalKafkaEventQueue") BlockingQueue signalEventsQueue, PackageRecordsService packageRecordsService, SourceService sourceService, - SegmentMapService segmentMapService, - CreditTrackerSettings config) { + SegmentMapService segmentMapService) { this.signalEventsQueue = signalEventsQueue; this.packageRecordsService = packageRecordsService; this.sourceService = sourceService; this.segmentMapService = segmentMapService; - this.config = config; } @KafkaListener( topics = "my-topic", - concurrency = "8" + concurrency = CONCURRENCY ) public void listen(List records, Acknowledgment acknowledgment) { + if (isShuttingdDown) { + log.debug("KafkaMonitoringService is shutting down, rejecting new messages"); + return; // без ака, завершаем обработку сообщений, которые уже попали в очередь, новые не обрабатываем + } + if (!processingPermits.tryAcquire()) { - log.warn("KafkaMonitoringService is shutting down, rejecting new messages"); + log.warn("cannot acquire processing permits"); return; } try { - if (isShuttingdDown) { - log.warn("KafkaMonitoringService is shutting down, rejecting new messages"); - return; // без ака, завершаем обработку сообщений, которые уже попали в очередь, новые не обрабатываем - } - acknowledgment.acknowledge(); process(records); log.info("Processed {} messages from Kafka", records.size()); @@ -170,9 +166,7 @@ public class KafkaMonitoringService { log.info("Stopping KafkaMonitoringService..."); isShuttingdDown = true; - for (int i = 0; i < 8; i++) { - processingPermits.acquire(); - } + processingPermits.acquire(PERMITS); try { SignalKafkaEvent rotateEvent = new SignalKafkaEvent(); @@ -185,7 +179,7 @@ public class KafkaMonitoringService { signalEventsQueue.put(stopEvent); log.info("STOP event sent to queue"); } finally { - processingPermits.release(); + processingPermits.release(PERMITS); } log.info("KafkaMonitoringService stopped");