From d3041a48bae0b8f25ab184117113e12207578857 Mon Sep 17 00:00:00 2001 From: Mikhail Trofimov Date: Thu, 18 Jun 2026 14:56:19 +0300 Subject: [PATCH] Get rid of MonitoringTimerService --- .../config/kafka/KafkaConfig.java | 2 +- .../kafka/PackageMessage.java | 15 +++ .../monitoring/KafkaMonitoringService.java | 53 +-------- .../mode/monitoring/MonitoringService.java | 40 ++----- .../monitoring/MonitoringTimerService.java | 112 ------------------ .../mode/monitoring/PackageMessage.java | 9 -- 6 files changed, 30 insertions(+), 201 deletions(-) create mode 100644 src/main/java/ru/nbch/credit_tracker/entities/external_request/kafka/PackageMessage.java delete mode 100644 src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/MonitoringTimerService.java delete mode 100644 src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/PackageMessage.java 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 4f26a00..57a0036 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,7 +11,7 @@ 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.service.mode.monitoring.PackageMessage; +import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage; import java.util.HashMap; import java.util.Map; diff --git a/src/main/java/ru/nbch/credit_tracker/entities/external_request/kafka/PackageMessage.java b/src/main/java/ru/nbch/credit_tracker/entities/external_request/kafka/PackageMessage.java new file mode 100644 index 0000000..db40aa3 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/entities/external_request/kafka/PackageMessage.java @@ -0,0 +1,15 @@ +package ru.nbch.credit_tracker.entities.external_request.kafka; + +import lombok.AllArgsConstructor; +import lombok.Getter; +import lombok.NoArgsConstructor; +import lombok.Setter; + +@AllArgsConstructor +@NoArgsConstructor +@Getter +@Setter +public class PackageMessage { + private Integer fid; + private String memberCode; +} 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 b0bca0a..3473597 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,22 +1,12 @@ package ru.nbch.credit_tracker.service.mode.monitoring; -import com.fasterxml.jackson.core.JsonProcessingException; -import com.fasterxml.jackson.databind.ObjectMapper; -import com.fasterxml.jackson.databind.json.JsonMapper; import lombok.extern.slf4j.Slf4j; -import org.apache.kafka.clients.consumer.ConsumerRecord; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.support.SendResult; +import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage; -import java.io.IOException; -import java.nio.file.Files; -import java.nio.file.Path; -import java.nio.file.Paths; -import java.nio.file.StandardOpenOption; import java.util.List; -import java.util.Random; -import java.util.UUID; import java.util.concurrent.CompletableFuture; @Slf4j @@ -24,59 +14,28 @@ import java.util.concurrent.CompletableFuture; public class KafkaMonitoringService { private final KafkaTemplate kafkaTemplate; - private final MonitoringTimerService monitoringTimerService; // @Value("${ct.tmp.dir}") private String tmpDir; - public KafkaMonitoringService(KafkaTemplate kafkaTemplate, - MonitoringTimerService monitoringTimerService) { + public KafkaMonitoringService(KafkaTemplate kafkaTemplate) { this.kafkaTemplate = kafkaTemplate; - this.monitoringTimerService = monitoringTimerService; } // @KafkaListener( // topics = "my-topic", // concurrency = "3" // ) - public void listen(List> records, + public void listen(List records, Acknowledgment acknowledgment) { - for (ConsumerRecord record : records) { - try { - processRecord(record.value()); - } catch (Exception e) { - log.error("Failed to process record: {}", record, e); - } - } + process(records); + // insert to db acknowledgment.acknowledge(); } - private void processRecord(PackageMessage msg) throws IOException { - Path packageDir = Paths.get(tmpDir, String.valueOf(msg.packageId())); - Files.createDirectories(packageDir); + private void process(List messages) { - Path filePath = packageDir.resolve(msg.uuid() + ".xml"); - Files.writeString(filePath, msg.message(), StandardOpenOption.CREATE_NEW); - - monitoringTimerService.registerFile(msg.packageId(), filePath); - } - - public void generateMessages() throws JsonProcessingException { - ObjectMapper mapper = JsonMapper.builder() - .build(); - Random rand = new Random(); - int max = 10; - int min = 1; - for (int i = 0; i < 500; i++) { - PackageMessage data = new PackageMessage( - rand.nextInt((max - min) + 1) + min, - "Message " + i, - UUID.randomUUID() - ); - - send(data.uuid().toString(), mapper.writeValueAsString(data)); - } } private void send(String id, String data) { diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/MonitoringService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/MonitoringService.java index 732d4ba..7e55e61 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/MonitoringService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/MonitoringService.java @@ -1,7 +1,5 @@ package ru.nbch.credit_tracker.service.mode.monitoring; -import lombok.Getter; -import lombok.Setter; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.slf4j.MDC; @@ -11,6 +9,7 @@ import ru.nbch.credit_tracker.entities.app.Hit; import ru.nbch.credit_tracker.entities.app.PackagesRecordsMap; import ru.nbch.credit_tracker.entities.app.PhonesMap; import ru.nbch.credit_tracker.entities.app.UserPackage; +import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage; import ru.nbch.credit_tracker.entities.external_response.online_download.Inquiry; import ru.nbch.credit_tracker.entities.external_response.online_download.Person; import ru.nbch.credit_tracker.entities.external_response.online_download.Persons; @@ -75,30 +74,15 @@ public class MonitoringService { } public void run() { - List messages = new ArrayList<>(); + List messages = new ArrayList<>(); for (int i = 1000; i < 2000; i++) { - MessageData messageData = new MessageData(); + PackageMessage messageData = new PackageMessage(); messageData.setFid(i); messageData.setMemberCode("ZX01LL"); messages.add(messageData); } - MessageData messageData1 = new MessageData(); - messageData1.setFid(114885613); - messageData1.setMemberCode("ZX01LL"); - messages.add(messageData1); - - MessageData messageData2 = new MessageData(); - messageData2.setFid(16445000); - messageData2.setMemberCode("ZX01LL"); - messages.add(messageData2); - - MessageData messageData3 = new MessageData(); - messageData3.setFid(109837221); - messageData3.setMemberCode("ZX01LL"); - messages.add(messageData3); - long start = System.currentTimeMillis(); log.info("Start selecting by one"); @@ -111,8 +95,8 @@ public class MonitoringService { log.info("Finish selecting by bacth. Time: {} ms", System.currentTimeMillis() - start); } - private void monitorByOne(List messages) { - for (MessageData message : messages) { + private void monitorByOne(List messages) { + for (PackageMessage message : messages) { Integer fid = message.getFid(); List phonesMaps = packageRecordsService.findCachedByFid(fid); for (PhonesMap phonesMap : phonesMaps) { @@ -136,8 +120,8 @@ public class MonitoringService { } } - private void monitorbyBatch(List messages) { - List fids = messages.stream().map(MessageData::getFid).collect(Collectors.toList()); + private void monitorbyBatch(List messages) { + List fids = messages.stream().map(PackageMessage::getFid).collect(Collectors.toList()); List memberCodes = messages.stream() .map(message -> message.getMemberCode().substring(0,6)) .toList(); @@ -161,7 +145,7 @@ public class MonitoringService { Map> userPackagePhoneDataMap = packagePhoneData.stream() .collect(Collectors.groupingBy(UserPackagePhoneData::getPhone)); - for (MessageData message : messages) { + for (PackageMessage message : messages) { Integer fid = message.getFid(); String memberCode = message.getMemberCode(); @@ -194,14 +178,6 @@ public class MonitoringService { } } - @Getter - @Setter - public static class MessageData { - private Integer fid; - private String memberCode; - // other data - } - private void monitor() { try { log.info("Start monitoring"); diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/MonitoringTimerService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/MonitoringTimerService.java deleted file mode 100644 index 2ea09ac..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/MonitoringTimerService.java +++ /dev/null @@ -1,112 +0,0 @@ -package ru.nbch.credit_tracker.service.mode.monitoring; - -import lombok.extern.slf4j.Slf4j; - -import java.io.BufferedWriter; -import java.io.IOException; -import java.nio.file.Files; -import java.nio.file.Path; -import java.nio.file.Paths; -import java.nio.file.StandardCopyOption; -import java.util.ArrayList; -import java.util.List; -import java.util.Queue; -import java.util.concurrent.*; - -//@Service -@Slf4j -public class MonitoringTimerService { - - private static final long DELAY_MINUTES = 5; - -// @Value("${ct.target.dir}") - private String targetDir; - - private final ScheduledExecutorService scheduler = - Executors.newScheduledThreadPool(4); - - private final ConcurrentHashMap bundles = - new ConcurrentHashMap<>(); - - public void registerFile(Integer packageId, Path filePath) { - bundles.compute(packageId, (id, existing) -> { - if (existing == null) { - PackageBundle bundle = new PackageBundle(); - bundle.addFile(filePath); - scheduleFlush(packageId, bundle); - log.info("Started new timer for packageId={}", packageId); - return bundle; - } - existing.addFile(filePath); - return existing; - }); - } - - private void scheduleFlush(Integer packageId, PackageBundle bundle) { - scheduler.schedule(() -> flush(packageId, bundle), DELAY_MINUTES, TimeUnit.MINUTES); - } - - private void flush(Integer packageId, PackageBundle bundle) { - // Атомарно забираем snapshot и удаляем из мапы - List snapshot = new ArrayList<>(); - bundles.compute(packageId, (id, current) -> { - snapshot.addAll(bundle.drainFiles()); - return null; // удаляем этот пакет - }); - - try { - if (!snapshot.isEmpty()) { - Path destDir = Paths.get(targetDir, String.valueOf(packageId)); - Files.createDirectories(destDir); - - List movedFiles = new ArrayList<>(); - for (Path src : snapshot) { - if (!Files.exists(src)) { - log.warn("File missing, skipping: {}", src); - continue; - } - Path dest = destDir.resolve(src.getFileName()); - Files.move(src, dest, StandardCopyOption.ATOMIC_MOVE); - movedFiles.add(dest); - } - mergeXmlFiles(movedFiles, destDir, packageId); - } - } catch (IOException e) { - log.error("Flush failed for packageId={}", packageId, e); - } finally { - // Файлы, добавленные во время мержа — забираем и стартуем новый цикл - bundle.drainFiles().forEach(leftover -> registerFile(packageId, leftover)); - } - } - - private void mergeXmlFiles(List files, Path destDir, Integer packageId) throws IOException { - if (files.isEmpty()) return; - Path merged = destDir.resolve("merged_" + packageId + ".xml"); - try (BufferedWriter writer = Files.newBufferedWriter(merged)) { - writer.write(""); - writer.newLine(); - for (Path file : files) { - writer.write(Files.readString(file)); - writer.newLine(); - } - writer.write(""); - } - log.info("Merged {} files → {} for packageId={}", files.size(), merged, packageId); - } - - - private static class PackageBundle { - private final Queue files = new ConcurrentLinkedQueue<>(); - - void addFile(Path path) { - files.add(path); - } - - List drainFiles() { - List result = new ArrayList<>(); - Path p; - while ((p = files.poll()) != null) result.add(p); - return result; - } - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/PackageMessage.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/PackageMessage.java deleted file mode 100644 index 21183f1..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/PackageMessage.java +++ /dev/null @@ -1,9 +0,0 @@ -package ru.nbch.credit_tracker.service.mode.monitoring; - -import java.util.UUID; - -public record PackageMessage( - Integer packageId, - String message, - UUID uuid -) {}