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

Get rid of test module; Moved code to MonitoringService

Browse source
This commit is contained in:
Mikhail Trofimov 2026-06-10 12:14:48 +03:00
parent 5b6f3ce63c
commit c3abd1ef6c
7 changed files with 78 additions and 180 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
pom.xml
View file

@ -300,25 +300,6 @@
</configuration>
</execution>
<execution>
<id>build-monitoring-test</id>
<phase>package</phase>
<goals>
<goal>repackage</goal>
</goals>
<configuration>
<mainClass>ru.nbch.credit_tracker.service.mode.test_monitoring.CTMonitoringTestApplication</mainClass>
<classifier>monitoring-test</classifier>
<attach>true</attach>
<excludes>
<exclude>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</exclude>
</excludes>
</configuration>
</execution>
<execution>
<id>build-migration</id>
<phase>package</phase>

4
src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/CTMonitoringApplication.java
View file

@ -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();
}

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

@ -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<String, String> kafkaTemplate;
private final MonitoringTimerService monitoringTimerService;
@Value("${ct.tmp.dir}")
// @Value("${ct.tmp.dir}")
private String tmpDir;
public KafkaMonitoringService(KafkaTemplate<String, String> 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<ConsumerRecord<String, PackageMessage>> records,
Acknowledgment acknowledgment) {
for (ConsumerRecord<String, PackageMessage> record : records) {

71
src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/MonitoringService.java
View file

@ -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<Integer> 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<Integer> fids) {
for (Integer fid : fids) {
List<PhonesMap> phonesMaps = packageRecordsService.findCachedByFid(fid);
for (PhonesMap phonesMap : phonesMaps) {
if (phonesMap != null) {
Long clientId = phonesMap.getPhone();
Integer flagDebt5000 = phonesMap.getFlagDebt5000();
Integer flag90days = phonesMap.getFlag90days();
List<PackagesRecordsMap> 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<Integer> fids) {
List<PhonesMap> phonesMaps = packageRecordsService.findCachedByFids(fids);
// key: phone; value: флаги просрочек
Map<Long, PhonesMap> phonesDataMap = phonesMaps.stream()
.collect(
Collectors.toMap(PhonesMap::getPhone, Function.identity())
);
// subjectId, packageId, signalTypes, phone только для статуса ON_MONITORING
List<UserPackagePhoneData> 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");

6
src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/MonitoringTimerService.java
View file

@ -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 =

33
src/main/java/ru/nbch/credit_tracker/service/mode/test_monitoring/CTMonitoringTestApplication.java
View file

@ -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();
}
}

110
src/main/java/ru/nbch/credit_tracker/service/mode/test_monitoring/MonitoringService.java
View file

@ -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<Integer> 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<Integer> fids) {
for (Integer fid : fids) {
List<PhonesMap> phonesMaps = packageRecordsService.findCachedByFid(fid);
for (PhonesMap phonesMap : phonesMaps) {
if (phonesMap != null) {
Long clientId = phonesMap.getPhone();
Integer flagDebt5000 = phonesMap.getFlagDebt5000();
Integer flag90days = phonesMap.getFlag90days();
List<PackagesRecordsMap> 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<Integer> fids) {
List<PhonesMap> phonesMaps = packageRecordsService.findCachedByFids(fids);
// key: phone; value: флаги просрочек
Map<Long, PhonesMap> phonesDataMap = phonesMaps.stream()
.collect(
Collectors.toMap(PhonesMap::getPhone, Function.identity())
);
// subjectId, packageId, signalTypes, phone только для статуса ON_MONITORING
List<UserPackagePhoneData> 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
}
}
}