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

update flags

Browse source
This commit is contained in:
ialbert 2026-01-16 17:47:41 +03:00
parent 477a515244
commit 9fce65ea4f
10 changed files with 77 additions and 91 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.gitignore vendored
View file

@ -31,3 +31,4 @@ build/
### VS Code ###
.vscode/
tmp/

7
credit-tracker.properties
View file

@ -90,9 +90,10 @@ ct.pipeline.id-enricher-stage-concurrency=10
ct.pipeline.id-enricher-queue-capacity=20
#Save part of signal xml file and merge; only one thread
ct.pipeline.merger-queue-capacity=50
#select and calc flags; instant run after name-enricher
ct.pipeline.flag-enricher-stage-concurrency=10
ct.pipeline.flag-enricher-queue-capacity=50
#select and calc flags
ct.flag-updater.concurrency=10
ct.flag-updater.queue-capacity=100
#Backoff settings
ct.pipeline.error-attempt-count=5

2
src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java
View file

@ -23,6 +23,8 @@ public class CreditTrackerSettings {
private XsdValidation xsdValidation;
@NestedConfigurationProperty
private PipelineSettings pipeline;
@NestedConfigurationProperty
private FlagUpdaterSettings flagUpdater;
private String signalsRootUri;
private String signalsToken;

9
src/main/java/ru/nbch/credit_tracker/config/settings/FlagUpdaterSettings.java Normal file
View file

@ -0,0 +1,9 @@
package ru.nbch.credit_tracker.config.settings;
import org.springframework.boot.context.properties.bind.DefaultValue;
public record FlagUpdaterSettings(
@DefaultValue("10") Integer concurrency,
@DefaultValue("100") Integer queueCapacity
) {
}

4
src/main/java/ru/nbch/credit_tracker/config/settings/PipelineSettings.java
View file

@ -4,14 +4,14 @@ import java.time.Duration;
import java.util.List;
import org.springframework.boot.context.properties.bind.DefaultValue;
//X это количество потоков Save
//тогда: Name должно быть ~3x, Fid должно быть ~2x чтобы они поспевали друг за другом?
public record PipelineSettings(
@DefaultValue("5000") Integer subjectBatchSize,
@DefaultValue("15") Integer nameEnricherStageConcurrency,
@DefaultValue("20") Integer nameEnricherQueueCapacity,
@DefaultValue("10") Integer idEnricherStageConcurrency,
@DefaultValue("15") Integer idEnricherQueueCapacity,
@DefaultValue("10") Integer flagEnricherStageConcurrency,
@DefaultValue("100") Integer flagEnricherQueueCapacity,
@DefaultValue("50") Integer mergerQueueCapacity,
@DefaultValue("5") Integer errorAttemptCount,
@DefaultValue("2s") Duration errorAttemptDelay,

2
src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java
View file

@ -10,6 +10,8 @@ public interface PhonesMapMapper {
List<PhonesMap> getCachedData(@Param("phones") List<Long> phones);
List<PhonesMap> searchedAtLessThan(@Param("daysCount") Integer daysCount, @Param("limit") int limit, @Param("offset") int offset);
PhonesMap findByPhone(@Param("phone") Long phone);
List<PhonesMap> findEligibleForFlagsUpdate(@Param("lastId") Long lastId,
@Param("batchSize") Integer batchSize);
void updateIdCachedData(@Param("json") String json);
void updateNameCachedData(@Param("json") String json);
void updateFlagCachedData(@Param("json") String json);

63
src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java
View file

@ -2,21 +2,12 @@ package ru.nbch.credit_tracker.service;
import java.util.List;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import ru.nbch.credit_tracker.entities.app.UserPackage;
import ru.nbch.credit_tracker.enums.Status;
import ru.nbch.credit_tracker.entities.app.PhonesMap;
import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper;
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
import ru.nbch.credit_tracker.service.task.Pipeline;
import ru.nbch.credit_tracker.service.task.PipelinePackageFabric;
import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator;
import ru.nbch.credit_tracker.service.task.PipelineRecalcPDOrchestrator;
import ru.nbch.credit_tracker.service.task.SetupParam;
import ru.nbch.credit_tracker.service.task.impl.PipelineMode;
import ru.nbch.credit_tracker.service.task.impl.FlagsEnricher;
/**
* Сервис периодического обновления фидов и флагов просрочек по пакетам находящимся в статусе ON_MONITORING
@ -26,47 +17,31 @@ import ru.nbch.credit_tracker.service.task.impl.PipelineMode;
@Service
@EnableScheduling
public class UpdatingFlagsService {
private final Long daysToUpdateFlags;
private final UserPackageMapper userPackageMapper;
private final static int BATCH_SIZE = 1000;
private final PhonesMapMapper phonesMapMapper;
private final PipelinePackageFabric pipelinePackageManager;
private final PipelinePackageOrchestrator pipelineOrchestrator;
private final PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator;
private final FlagsEnricher flagsEnricherV2;
public UpdatingFlagsService(CreditTrackerSettings config,
UserPackageMapper userPackageMapper,
PhonesMapMapper phonesMapMapper,
PipelinePackageFabric pipelinePackageManager,
PipelinePackageOrchestrator pipelineOrchestrator,
PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator) {
this.daysToUpdateFlags = config.getDaysToUpdateFlags() != null ? config.getDaysToUpdateFlags() : 1;
this.userPackageMapper = userPackageMapper;
public UpdatingFlagsService(PhonesMapMapper phonesMapMapper, FlagsEnricher flagsEnricherV2) {
this.phonesMapMapper = phonesMapMapper;
this.pipelinePackageManager = pipelinePackageManager;
this.pipelineOrchestrator = pipelineOrchestrator;
this.pipelineRecalcPDOrchestrator = pipelineRecalcPDOrchestrator;
this.flagsEnricherV2 = flagsEnricherV2;
}
@Scheduled(cron = "${ct.update-flags-cron-expression}")
private void update() {
log.info("Start updating flags");
List<UserPackage> packages = userPackageMapper.findAllPackagesByStatus(Status.ON_MONITORING.name());
for (UserPackage userPackage : packages) {
MDC.put("ru.nbch.credit_tracker.log.mdc.key", "[" + userPackage.getPackageName() + "]");
log.info("Check on updating flags for Package {}", userPackage.getPackageName());
boolean hasOldRow = userPackageMapper.hasOldRow(userPackage.getId(), daysToUpdateFlags.intValue());
log.info("There are {} old rows", hasOldRow ? "exist" : "doesn't exist");
if (hasOldRow) {
try {
userPackageMapper.updatePackageStatus(userPackage.getId(), Status.PROCESSING_REPEATED.name());
Pipeline pipeline = pipelinePackageManager.build(new SetupParam(userPackage.getId(), null, PipelineMode.DEFAULT));
pipelineOrchestrator.startOrDelay(pipeline, userPackage.getId());
} catch (Exception e) {
log.error("Failed to build pipeline", e);
log.info("Start filling flags");
long lastId = 0L;
int batchNumber = 1;
List<PhonesMap> phonesMaps;
while (true) {
phonesMaps = phonesMapMapper.findEligibleForFlagsUpdate(lastId, BATCH_SIZE);
log.trace("batch {}, signals_phones_map records loaded: {}", batchNumber, phonesMaps.size());
if (phonesMaps.isEmpty()) {
break;
}
flagsEnricherV2.accept(phonesMaps);
lastId = phonesMaps.getLast().getId();
log.trace("batch {}, lastId {}", batchNumber++, lastId);
}
MDC.remove("ru.nbch.credit_tracker.log.mdc.key");
}
log.info("Finish updating flags");
}
}

50
src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java
View file

@ -7,20 +7,18 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.Stream;
import lombok.extern.slf4j.Slf4j;
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.model.FlagData;
import ru.nbch.credit_tracker.model.SubjectProfile;
import ru.nbch.credit_tracker.model.impl.SubjectPayload;
import ru.nbch.credit_tracker.service.indic.FlagsService2;
import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
@Slf4j
@Service
public class FlagsEnricher implements Consumer<SubjectPayload> {
public class FlagsEnricher {
private final FlagsService2 flagsService;
private final PackageRecordsService packageRecordsService;
@ -31,10 +29,10 @@ public class FlagsEnricher implements Consumer<SubjectPayload> {
CreditTrackerSettings creditTrackerSettings) {
this.flagsService = flagsService;
this.packageRecordsService = packageRecordsService;
this.ioPool = new ThreadPoolExecutor(creditTrackerSettings.getPipeline().flagEnricherStageConcurrency(),
10,
this.ioPool = new ThreadPoolExecutor(creditTrackerSettings.getFlagUpdater().concurrency(),
creditTrackerSettings.getFlagUpdater().concurrency(),
0L, TimeUnit.MINUTES,
new LinkedBlockingQueue<>(creditTrackerSettings.getPipeline().flagEnricherQueueCapacity()),
new LinkedBlockingQueue<>(creditTrackerSettings.getFlagUpdater().queueCapacity()),
r -> {
Thread t = new Thread(r);
t.setName("flags-enricher-%d".formatted(t.threadId()));
@ -44,17 +42,11 @@ public class FlagsEnricher implements Consumer<SubjectPayload> {
);
}
public void process(SubjectPayload subjectPayload) {
process(subjectPayload, false);
}
public void process(List<PhonesMap> phonesMaps) {
log.debug("FlagsEnricher handle: size: {}", phonesMaps.size());
public void process(SubjectPayload subjectPayload, boolean update) {
List<SubjectProfile> subjects = subjectPayload.getSubjects();
log.debug("FlagsEnricher handle by packageId: {}, size: {}", subjectPayload.getPackageId(), subjects.size());
List<Integer> noFlagDataFids = subjects.stream()
.filter(subjectProfile -> subjectProfile.getFlagData() == null)
.map(SubjectProfile::getFid)
List<Integer> noFlagDataFids = phonesMaps.stream()
.map(PhonesMap::getFid)
.toList();
LocalDate yearAgo = LocalDate.now().minusYears(1);
@ -63,34 +55,28 @@ public class FlagsEnricher implements Consumer<SubjectPayload> {
Map<Integer, Integer> flag90ByPastDueArrear = flagsService.findFlag90ByPastDueArrear(noFlagDataFids, yearAgo);
Map<Integer, Integer> flag90ByRuPaymt = flagsService.findFlag90ByRuPaymt(noFlagDataFids, yearAgo);
List<FlagData> flagCacheOnUpdate = subjects.stream()
.filter(subjectProfile -> subjectProfile.getFid() != null)
.flatMap(subject -> {
if (subject.getFlagData() == null) {
Integer debt5000 = flagsDebt5000.getOrDefault(subject.getFid(), 0);
int pastDue90 = flag90ByPastDueArrear.getOrDefault(subject.getFid(), 0);
int paymt90 = flag90ByRuPaymt.getOrDefault(subject.getFid(), 0);
List<FlagData> flagCacheOnUpdate = phonesMaps.stream()
.flatMap(phonesMap -> {
Integer fid = phonesMap.getFid();
Integer debt5000 = flagsDebt5000.getOrDefault(fid, 0);
int pastDue90 = flag90ByPastDueArrear.getOrDefault(fid, 0);
int paymt90 = flag90ByRuPaymt.getOrDefault(fid, 0);
Integer flagDays90 = Math.max(pastDue90, paymt90);
FlagData flagData = FlagData.builder()
.flag_90_12(flagDays90)
.flag_debt5000(debt5000)
.phone(subject.getPhone())
.phone(phonesMap.getPhone())
.build();
subject.setFlagData(flagData);
return Stream.of(flagData);
}
return Stream.empty();
}).toList();
packageRecordsService.updateFlagCache(flagCacheOnUpdate);
}
@Override
public void accept(SubjectPayload subjectPayload) {
public void accept(List<PhonesMap> phonesMaps) {
ThreadPoolExecutor casted = (ThreadPoolExecutor) ioPool;
log.info("submitting payload. active tasks number: {}; queue size: {}", casted.getActiveCount(), casted.getQueue().size());
this.ioPool.submit(() -> process(subjectPayload));
this.ioPool.submit(() -> process(phonesMaps));
}
}

10
src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml
View file

@ -116,6 +116,16 @@
birth_dt = EXCLUDED.birth_dt
</select>
<select id="findEligibleForFlagsUpdate" resultMap="PhoneMap">
select id, fid, phone
from signals_phones_map
where fid is not null
and (calculated_at is null or calculated_at &lt; searched_at)
and id > #{lastId}
order by id
limit #{batchSize}
</select>
<update id="updateIdCachedData" parameterType="string">
update signals_phones_map dst
set ser_num = src.ser_num,