diff --git a/src/main/java/ru/nbch/credit_tracker/config/kafka/KafkaConfig.java b/src/main/java/ru/nbch/credit_tracker/config/kafka/KafkaConfig.java index 57a0036..3630601 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/kafka/KafkaConfig.java +++ b/src/main/java/ru/nbch/credit_tracker/config/kafka/KafkaConfig.java @@ -11,15 +11,25 @@ import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; import org.springframework.kafka.core.*; import org.springframework.kafka.listener.ContainerProperties; import org.springframework.kafka.support.serializer.JsonDeserializer; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage; +import ru.nbch.credit_tracker.model.SignalKafkaEvent; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; @EnableKafka @Configuration public class KafkaConfig { + @Bean("signalKafkaEventQueue") + public BlockingQueue signalKafkaEventQueue(CreditTrackerSettings settings) { + // TODO add capacity + return new LinkedBlockingQueue<>(); + } + @Bean public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory(ConsumerFactory consumerFactory) { ConcurrentKafkaListenerContainerFactory factory = diff --git a/src/main/java/ru/nbch/credit_tracker/enums/EventType.java b/src/main/java/ru/nbch/credit_tracker/enums/EventType.java new file mode 100644 index 0000000..41c6b4c --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/enums/EventType.java @@ -0,0 +1,7 @@ +package ru.nbch.credit_tracker.enums; + +public enum EventType { + WRITE, + ROTATE, + STOP // используется для остановки +} diff --git a/src/main/java/ru/nbch/credit_tracker/model/SignalKafkaEvent.java b/src/main/java/ru/nbch/credit_tracker/model/SignalKafkaEvent.java new file mode 100644 index 0000000..5f49721 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/model/SignalKafkaEvent.java @@ -0,0 +1,16 @@ +package ru.nbch.credit_tracker.model; + +import lombok.*; +import ru.nbch.credit_tracker.enums.EventType; + +@Getter +@Setter +@Builder +@NoArgsConstructor +@AllArgsConstructor +public class SignalKafkaEvent { + private EventType eventType; + + private Long packageId; + // other data for signal +} 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 564ee0d..d3e8c1a 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 @@ -26,9 +26,10 @@ public class CTMonitoringApplication { SpringApplication application = new SpringApplication(CTMonitoringApplication.class); application.setWebApplicationType(WebApplicationType.NONE); ConfigurableApplicationContext ctx = application.run(args); - MonitoringService monitoringService = ctx.getBean(MonitoringService.class); - monitoringService.run(); +// MonitoringService monitoringService = ctx.getBean(MonitoringService.class); +// monitoringService.run(); -// ctx.getBean(KafkaMonitoringService.class).generateMessages(); + 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 3284bb5..690980f 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,12 +1,19 @@ 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.kafka.core.KafkaTemplate; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.support.Acknowledgment; -import org.springframework.kafka.support.SendResult; +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; import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage; +import ru.nbch.credit_tracker.enums.EventType; +import ru.nbch.credit_tracker.model.SignalKafkaEvent; import ru.nbch.credit_tracker.model.UserPackagePhoneData; import ru.nbch.credit_tracker.service.nbch.SourceService; import ru.nbch.credit_tracker.service.signal.PackageRecordsService; @@ -15,41 +22,64 @@ import ru.nbch.credit_tracker.utils.CTUtil; import java.util.List; import java.util.Map; import java.util.Set; -import java.util.concurrent.CompletableFuture; +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 +@Service public class KafkaMonitoringService { - private final KafkaTemplate kafkaTemplate; + private final ScheduledExecutorService rotateScheduler = Executors.newSingleThreadScheduledExecutor( + new CustomizableThreadFactory("monitoring-rotate-") + ); + private volatile boolean isShuttingdDown = false; + private final AtomicInteger activeProcessCount = new AtomicInteger(0); + + private final BlockingQueue signalEventsQueue; + private final PackageRecordsService packageRecordsService; private final SourceService sourceService; - public KafkaMonitoringService(KafkaTemplate kafkaTemplate, + private final CreditTrackerSettings config; + + public KafkaMonitoringService(@Qualifier("signalKafkaEventQueue") BlockingQueue signalEventsQueue, PackageRecordsService packageRecordsService, - SourceService sourceService) { - this.kafkaTemplate = kafkaTemplate; + SourceService sourceService, + CreditTrackerSettings config) { + this.signalEventsQueue = signalEventsQueue; this.packageRecordsService = packageRecordsService; this.sourceService = sourceService; + this.config = config; } -// @KafkaListener( -// topics = "my-topic", -// concurrency = "3" -// ) + @KafkaListener( + topics = "my-topic", + concurrency = "8" + ) public void listen(List records, Acknowledgment acknowledgment) { - process(records); - - // insert to db - acknowledgment.acknowledge(); + if (isShuttingdDown) { + log.warn("KafkaMonitoringService is shutting down, rejecting new messages"); + return; // без ака, завершаем обработку сообщений, которые уже попали в очередь, новые не обрабатываем + } + try { + acknowledgment.acknowledge(); + activeProcessCount.incrementAndGet(); + process(records); + log.info("Processed {} messages from Kafka", records.size()); + } finally { + activeProcessCount.decrementAndGet(); + } } private void process(List messages) { List fids = messages.stream().map(PackageMessage::getFid).collect(Collectors.toList()); List memberCodes = messages.stream() - .map(message -> message.getMemberCode().substring(0,6)) + .map(message -> message.getMemberCode().substring(0, 6)) .toList(); List phonesMaps = packageRecordsService.findCachedByFids(fids); @@ -102,30 +132,88 @@ public class KafkaMonitoringService { } for (UserPackagePhoneData userPackagePhoneData : userPackagePhoneDataList) { + Long packageId = userPackagePhoneData.getPackageId(); String subjectId = userPackagePhoneData.getSubjectId(); String signalTypes = userPackagePhoneData.getSignalTypes(); if (signalTypes.contains(signalType.toString())) { - // TODO put to queue for process + SignalKafkaEvent event = new SignalKafkaEvent(); + event.setEventType(EventType.WRITE); + event.setPackageId(packageId); + putToQueue(event); } } } } + + // TODO insert hits } - private void send(String id, String data) { - CompletableFuture> future = - kafkaTemplate.send("my-topic", id, data); + private void putToQueue(SignalKafkaEvent event) { + try { + signalEventsQueue.put(event); + } catch (InterruptedException e) { + log.error("Error while putting signal event", e); + Thread.currentThread().interrupt(); + } + } - future.whenComplete((result, exception) -> { - if (exception == null) { - log.info("Sent user event=[{}] with offset=[{}]", - data, result.getRecordMetadata().offset()); - } else { - log.error("Unable to send user event=[{}] due to: {}", - data, exception.getMessage()); + /** + * Таймер, который отсчитывает {@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() { + log.info("Stopping KafkaMonitoringService..."); + isShuttingdDown = true; + + while (activeProcessCount.get() > 0) { + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } + } + + try { + SignalKafkaEvent rotateEvent = new SignalKafkaEvent(); + rotateEvent.setEventType(EventType.ROTATE); + signalEventsQueue.put(rotateEvent); + log.info("ROTATE event sent to queue"); + + SignalKafkaEvent stopEvent = new SignalKafkaEvent(); + 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(); + } + + 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/SignalKafkaEventWriterService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java new file mode 100644 index 0000000..272b9ef --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java @@ -0,0 +1,67 @@ +package ru.nbch.credit_tracker.service.mode.monitoring; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.stereotype.Service; +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; + +@Slf4j +@Service +public class SignalKafkaEventWriterService { + + private final BlockingQueue signalEventsQueue; + private final CreditTrackerSettings config; + + public SignalKafkaEventWriterService(@Qualifier("signalKafkaEventQueue") BlockingQueue signalEventsQueue, + CreditTrackerSettings config) { + this.signalEventsQueue = signalEventsQueue; + this.config = config; + } + + // TODO вызвать в main + public void startWriting() { + while (true) { + SignalKafkaEvent event = take(); + if (event.getEventType() == EventType.STOP) { + break; + } + + if (event.getEventType() == EventType.WRITE) { + writeSignal(event); + } + + if (event.getEventType() == EventType.ROTATE) { + rotate(); + } + } + } + + /** + * Записывает сигнал в xml для конкретного пакета + */ + private void writeSignal(SignalKafkaEvent event) { + // TODO записать сигнал в существующий xml/создать новый и записать + // TODO добавить пакет в сет + } + + /** + * Упаковывает все результирующие xml в архив + */ + private void rotate() { + // TODO пройтись по сету пакетов и финализировать запись. Очистить сет + } + + private SignalKafkaEvent take() { + try { + return signalEventsQueue.take(); + } catch (InterruptedException e) { + log.error("Error while taking signal event", e); + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + } +}