diff --git a/pom.xml b/pom.xml index 276e164..dc2df27 100644 --- a/pom.xml +++ b/pom.xml @@ -300,25 +300,6 @@ - - build-monitoring-test - package - - repackage - - - ru.nbch.credit_tracker.service.mode.test_monitoring.CTMonitoringTestApplication - monitoring-test - true - - - org.projectlombok - lombok - - - - - build-migration package 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 6e5257a..564ee0d 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,8 +26,8 @@ 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(); } 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 f6c7940..b0bca0a 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 @@ -5,12 +5,9 @@ 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.beans.factory.annotation.Value; -import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.support.SendResult; -import org.springframework.stereotype.Service; import java.io.IOException; import java.nio.file.Files; @@ -23,13 +20,13 @@ import java.util.UUID; import java.util.concurrent.CompletableFuture; @Slf4j -@Service +//@Service public class KafkaMonitoringService { private final KafkaTemplate kafkaTemplate; private final MonitoringTimerService monitoringTimerService; - @Value("${ct.tmp.dir}") +// @Value("${ct.tmp.dir}") private String tmpDir; public KafkaMonitoringService(KafkaTemplate kafkaTemplate, @@ -38,10 +35,10 @@ public class KafkaMonitoringService { this.monitoringTimerService = monitoringTimerService; } - @KafkaListener( - topics = "my-topic", - concurrency = "3" - ) +// @KafkaListener( +// topics = "my-topic", +// concurrency = "3" +// ) public void listen(List> records, Acknowledgment acknowledgment) { for (ConsumerRecord record : records) { 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 b0361a7..b6c32cc 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 @@ -20,6 +20,7 @@ import ru.nbch.credit_tracker.entities.user_response.monitoring_signal_response. import ru.nbch.credit_tracker.entities.user_response.monitoring_signal_response.Subject; import ru.nbch.credit_tracker.enums.Status; import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; +import ru.nbch.credit_tracker.model.UserPackagePhoneData; import ru.nbch.credit_tracker.service.signal.HitsService; import ru.nbch.credit_tracker.service.signal.PackageRecordsService; import ru.nbch.credit_tracker.service.signal.SignalRestService; @@ -38,6 +39,8 @@ import java.time.LocalDateTime; import java.util.*; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Function; +import java.util.stream.Collectors; /** * Сервис периодического опроса официального сервиса сигналов на наличие новых отчетов по пакетам находящимся в статусе ON_MONITORING @@ -68,11 +71,73 @@ public class MonitoringService { } public void run() { -// ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(new CustomizableThreadFactory("monitoring-")); -// long period = config.getMonitoringInterval() != null ? config.getMonitoringInterval() : 15; -// scheduler.scheduleAtFixedRate(this::monitor, 0, period, TimeUnit.MINUTES); + List fids = new ArrayList<>(); + + for (int i = 1000; i < 2000; i++) { + fids.add(i); + } + + long start = System.currentTimeMillis(); + log.info("Start selecting by one"); + monitorByOne(fids); + log.info("Finish selecting by one. Time: {} ms", System.currentTimeMillis() - start); + + start = System.currentTimeMillis(); + log.info("Start selecting by batch"); + monitorbyBatch(fids); + log.info("Finish selecting by bacth. Time: {} ms", System.currentTimeMillis() - start); } + private void monitorByOne(List fids) { + for (Integer fid : fids) { + List phonesMaps = packageRecordsService.findCachedByFid(fid); + for (PhonesMap phonesMap : phonesMaps) { + if (phonesMap != null) { + Long clientId = phonesMap.getPhone(); + Integer flagDebt5000 = phonesMap.getFlagDebt5000(); + Integer flag90days = phonesMap.getFlag90days(); + List packageRecords = packageRecordsService.findByPhone(clientId); + for (PackagesRecordsMap packageRecord : packageRecords) { + Long packageId = packageRecord.getPackageId(); + String subjectId = packageRecord.getSubjectId(); + UserPackage userPackage = userPackageMapper.findPackage(packageId); + String status = userPackage.getStatus(); + String signalTypes = userPackage.getSignalTypes(); + + // проверить статус ON_MPNITORING + разрешенные signalTypes + } + } + } + } + } + + private void monitorbyBatch(List fids) { + List phonesMaps = packageRecordsService.findCachedByFids(fids); + // key: phone; value: флаги просрочек + Map phonesDataMap = phonesMaps.stream() + .collect( + Collectors.toMap(PhonesMap::getPhone, Function.identity()) + ); + + // subjectId, packageId, signalTypes, phone только для статуса ON_MONITORING + List userPackagePhoneDataList = packageRecordsService + .selectUserPackagePhoneData(phonesDataMap.keySet()); + + // остались только нужные номера, те что не вошли в userPackagePhoneDataList игнорируются + for (UserPackagePhoneData userPackagePhoneData : userPackagePhoneDataList) { + Long clientId = userPackagePhoneData.getPhone(); + String subjectId = userPackagePhoneData.getSubjectId(); + String signalTypes = userPackagePhoneData.getSignalTypes(); + + PhonesMap phonesMap = phonesDataMap.get(clientId); + Integer flagDebt5000 = phonesMap.getFlagDebt5000(); + Integer flag90days = phonesMap.getFlag90days(); + + // проверить разрешенные signalTypes + } + } + + 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 index 6f108f0..2ea09ac 100644 --- 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 @@ -1,8 +1,6 @@ package ru.nbch.credit_tracker.service.mode.monitoring; import lombok.extern.slf4j.Slf4j; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.stereotype.Service; import java.io.BufferedWriter; import java.io.IOException; @@ -15,13 +13,13 @@ import java.util.List; import java.util.Queue; import java.util.concurrent.*; -@Service +//@Service @Slf4j public class MonitoringTimerService { private static final long DELAY_MINUTES = 5; - @Value("${ct.target.dir}") +// @Value("${ct.target.dir}") private String targetDir; private final ScheduledExecutorService scheduler = diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/test_monitoring/CTMonitoringTestApplication.java b/src/main/java/ru/nbch/credit_tracker/service/mode/test_monitoring/CTMonitoringTestApplication.java deleted file mode 100644 index 8c554b5..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/test_monitoring/CTMonitoringTestApplication.java +++ /dev/null @@ -1,33 +0,0 @@ -package ru.nbch.credit_tracker.service.mode.test_monitoring; - -import com.fasterxml.jackson.core.JsonProcessingException; -import org.springframework.boot.SpringApplication; -import org.springframework.boot.WebApplicationType; -import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.context.annotation.ComponentScan; -import org.springframework.context.annotation.FilterType; - -@SpringBootApplication -@ComponentScan( - value = {"ru.nbch.credit_tracker"}, - excludeFilters = { - @ComponentScan.Filter( - type = FilterType.REGEX, - pattern = { - "ru\\.nbch\\.credit_tracker\\.service\\.mode\\.common..*", - "ru\\.nbch\\.credit_tracker\\.service\\.mode\\.migration..*", - "ru\\.nbch\\.credit_tracker\\.service\\.mode\\.monitoring..*" - } - ) - } -) -public class CTMonitoringTestApplication { - public static void main(String[] args) throws JsonProcessingException { - SpringApplication application = new SpringApplication(CTMonitoringTestApplication.class); - application.setWebApplicationType(WebApplicationType.NONE); - ConfigurableApplicationContext ctx = application.run(args); - MonitoringService monitoringService = ctx.getBean(MonitoringService.class); - monitoringService.run(); - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/test_monitoring/MonitoringService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/test_monitoring/MonitoringService.java deleted file mode 100644 index 12b8f35..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/test_monitoring/MonitoringService.java +++ /dev/null @@ -1,110 +0,0 @@ -package ru.nbch.credit_tracker.service.mode.test_monitoring; - -import lombok.extern.slf4j.Slf4j; -import org.springframework.stereotype.Service; -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.mapper.signal.UserPackageMapper; -import ru.nbch.credit_tracker.model.UserPackagePhoneData; -import ru.nbch.credit_tracker.service.signal.PackageRecordsService; - -import java.util.ArrayList; -import java.util.List; -import java.util.Map; -import java.util.function.Function; -import java.util.stream.Collectors; - -/** - * Сервис периодического опроса официального сервиса сигналов на наличие новых отчетов по пакетам находящимся в статусе ON_MONITORING - */ -@Slf4j -@Service -public class MonitoringService { - - private final PackageRecordsService packageRecordsService; - private final UserPackageMapper userPackageMapper; - - public MonitoringService(PackageRecordsService packageRecordsService, - UserPackageMapper userPackageMapper) { - this.packageRecordsService = packageRecordsService; - this.userPackageMapper = userPackageMapper; - } - - public void run() { - List fids = new ArrayList<>(); - - for (int i = 1000; i < 2000; i++) { - fids.add(i); - } - - fids.add(114885613); - fids.add(16445000); - fids.add(114829813); - fids.add(109837221); - fids.add(108550768); - fids.add(108546145); - fids.add(111098278); - - - long start = System.currentTimeMillis(); - log.info("Start selecting by one"); - monitorByOne(fids); - log.info("Finish selecting by one. Time: {} ms", System.currentTimeMillis() - start); - - start = System.currentTimeMillis(); - log.info("Start selecting by batch"); - monitorbyBatch(fids); - log.info("Finish selecting by bacth. Time: {} ms", System.currentTimeMillis() - start); - } - - private void monitorByOne(List fids) { - for (Integer fid : fids) { - List phonesMaps = packageRecordsService.findCachedByFid(fid); - for (PhonesMap phonesMap : phonesMaps) { - if (phonesMap != null) { - Long clientId = phonesMap.getPhone(); - Integer flagDebt5000 = phonesMap.getFlagDebt5000(); - Integer flag90days = phonesMap.getFlag90days(); - List packageRecords = packageRecordsService.findByPhone(clientId); - for (PackagesRecordsMap packageRecord : packageRecords) { - Long packageId = packageRecord.getPackageId(); - String subjectId = packageRecord.getSubjectId(); - UserPackage userPackage = userPackageMapper.findPackage(packageId); - String status = userPackage.getStatus(); - String signalTypes = userPackage.getSignalTypes(); - - // проверить статус ON_MPNITORING + разрешенные signalTypes - } - } - } - } - } - - private void monitorbyBatch(List fids) { - List phonesMaps = packageRecordsService.findCachedByFids(fids); - // key: phone; value: флаги просрочек - Map phonesDataMap = phonesMaps.stream() - .collect( - Collectors.toMap(PhonesMap::getPhone, Function.identity()) - ); - - // subjectId, packageId, signalTypes, phone только для статуса ON_MONITORING - List userPackagePhoneDataList = packageRecordsService - .selectUserPackagePhoneData(phonesDataMap.keySet()); - - // остались только нужные номера, те что не вошли в userPackagePhoneDataList игнорируются - for (UserPackagePhoneData userPackagePhoneData : userPackagePhoneDataList) { - Long clientId = userPackagePhoneData.getPhone(); - String subjectId = userPackagePhoneData.getSubjectId(); - String signalTypes = userPackagePhoneData.getSignalTypes(); - - PhonesMap phonesMap = phonesDataMap.get(clientId); - Integer flagDebt5000 = phonesMap.getFlagDebt5000(); - Integer flag90days = phonesMap.getFlag90days(); - - // проверить разрешенные signalTypes - } - } - -}