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

Use semaphore instead of AtomicInteger counter

Browse source
This commit is contained in:
Mikhail Trofimov 2026-06-23 15:35:35 +03:00
parent d9f23cdff9
commit ec93c4b726
1 changed files with 16 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

33
src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java
View file

@ -21,14 +21,14 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.Semaphore;
import java.util.stream.Collectors;
@Slf4j
@Service
public class KafkaMonitoringService {
private volatile boolean isShuttingdDown = false;
private final AtomicInteger activeProcessCount = new AtomicInteger(0);
private final Semaphore processingPermits = new Semaphore(8); // количество равно concurency листенера
private final BlockingQueue<SignalKafkaEvent> signalEventsQueue;
@ -53,17 +53,22 @@ public class KafkaMonitoringService {
)
public void listen(List<PackageMessage> records,
Acknowledgment acknowledgment) {
if (isShuttingdDown) {
if (!processingPermits.tryAcquire()) {
log.warn("KafkaMonitoringService is shutting down, rejecting new messages");
return; // без ака, завершаем обработку сообщений, которые уже попали в очередь, новые не обрабатываем
return;
}
try {
if (isShuttingdDown) {
log.warn("KafkaMonitoringService is shutting down, rejecting new messages");
return; // без ака, завершаем обработку сообщений, которые уже попали в очередь, новые не обрабатываем
}
acknowledgment.acknowledge();
activeProcessCount.incrementAndGet();
process(records);
log.info("Processed {} messages from Kafka", records.size());
} finally {
activeProcessCount.decrementAndGet();
processingPermits.release();
}
}
@ -149,17 +154,12 @@ public class KafkaMonitoringService {
}
@PreDestroy
public void stopKafkaMonitoringService() {
public void stopKafkaMonitoringService() throws InterruptedException {
log.info("Stopping KafkaMonitoringService...");
isShuttingdDown = true;
while (activeProcessCount.get() > 0) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
for (int i = 0; i < 8; i++) {
processingPermits.acquire();
}
try {
@ -172,9 +172,8 @@ public class KafkaMonitoringService {
stopEvent.setEventType(EventType.STOP);
signalEventsQueue.put(stopEvent);
log.info("STOP event sent to queue");
} catch (InterruptedException e) {
log.error("Error while sending ROTATE and STOP event", e);
Thread.currentThread().interrupt();
} finally {
processingPermits.release();
}
log.info("KafkaMonitoringService stopped");