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

http://jira.mfd.msk:8088/browse/BKI-3217

Browse source
This commit is contained in:
etreschenkov 2026-01-14 19:07:23 +03:00
parent 35bc835a0d
commit 5c1d9ccd2a
7 changed files with 39 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java
View file

@ -13,6 +13,8 @@ public class CreditTrackerApplication {
ConfigurableApplicationContext ctx = SpringApplication.run(CreditTrackerApplication.class, args);
UpdateProcessingPackagesService updateProcessingPackagesService = ctx.getBean(UpdateProcessingPackagesService.class);
MonitoringService monitoringService = ctx.getBean(MonitoringService.class);
// UpdatingFlagsService updatingFlagsService = ctx.getBean(UpdatingFlagsService.class);
// updatingFlagsService.update();
updateProcessingPackagesService.run();
monitoringService.run();
}

8
src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java
View file

@ -5,6 +5,7 @@ import java.util.function.Supplier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper;
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
import ru.nbch.credit_tracker.model.impl.EmptyTerminatePayload;
import ru.nbch.credit_tracker.model.impl.PartFilePayload;
@ -15,6 +16,7 @@ import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
import ru.nbch.credit_tracker.service.task.SetupParam;
import ru.nbch.credit_tracker.service.task.impl.KickPipeLineDefaultImpl;
import ru.nbch.credit_tracker.service.task.impl.KickPipeLineRepeatedRestoreImpl;
import ru.nbch.credit_tracker.service.task.impl.KickRecalcPipeLine;
import ru.nbch.credit_tracker.service.task.impl.NoOpTerminator;
import ru.nbch.credit_tracker.service.task.impl.XmlMergeCoordinator;
import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
@ -25,7 +27,8 @@ public class PipelineStarterConfig {
@Bean
public Function<SetupParam, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory(CreditTrackerSettings creditTrackerSettings,
UserPackageMapper userPackageMapper,
PackageRecordsService packageRecordsService) {
PackageRecordsService packageRecordsService,
PhonesMapMapper phonesMapMapper) {
return (setupParam) -> switch (setupParam.mode()) {
case DEFAULT ->
new KickPipeLineDefaultImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings);
@ -33,7 +36,8 @@ public class PipelineStarterConfig {
new KickPipeLineDefaultImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings);
case REPEATED_RESTORE ->
new KickPipeLineRepeatedRestoreImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings);
case RECALC_PERSONAL_DATA -> null;
case RECALC_PERSONAL_DATA ->
new KickRecalcPipeLine(setupParam, userPackageMapper, phonesMapMapper, packageRecordsService, creditTrackerSettings);
};
}

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

@ -70,7 +70,7 @@ public class UpdatingFlagsService {
}
// @Scheduled(cron = "${ct.update-flags-cron-expression}")
private void update() {
public void update() {
log.info("Start updating flags");
try {
Pipeline pipeline = pipelinePackageManager.buildRecalculationPD(new SetupParam(null, null, PipelineMode.RECALC_PERSONAL_DATA));

2
src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java
View file

@ -62,7 +62,7 @@ public class FidIdUpdater implements Handler<SubjectPayload, VoidPayload> {
})
.toList();
packageRecordsService.updateIdCache(idCacheOnUpdate);
packageRecordsService.updateIdCache(personOnProcessing);
log.debug("personOnProcessing size: {}", personOnProcessing.size());
return new VoidPayload();

2
src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java
View file

@ -43,7 +43,7 @@ public class FidNameUpdater implements Handler<SubjectPayload, SubjectPayload> {
Name name = nameByPhone.get(CTUtil.normalizePhoneNumber(subjectProfile.getPhone().toString()));
if (name == null) {
notFoundFidPhones.add(subjectProfile.getPhone());
} else if (subjectProfile.getFid().equals(name.getFid())) {
} else if (!subjectProfile.getFid().equals(name.getFid())) {
changingFidPhones.add(subjectProfile);
}
});

38
src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java
View file

@ -2,16 +2,16 @@ package ru.nbch.credit_tracker.service.task.impl;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import ru.nbch.credit_tracker.entities.app.PhonesMap;
import ru.nbch.credit_tracker.entities.app.UserPackage;
import ru.nbch.credit_tracker.enums.PoisonType;
import ru.nbch.credit_tracker.exceptions.SignalClientException;
import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper;
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
import ru.nbch.credit_tracker.model.IdData;
import ru.nbch.credit_tracker.model.NameData;
import ru.nbch.credit_tracker.model.PipelineContext;
import ru.nbch.credit_tracker.model.SubjectProfile;
import ru.nbch.credit_tracker.model.impl.PackageVariable;
@ -21,7 +21,7 @@ import ru.nbch.credit_tracker.service.task.SetupParam;
import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
public abstract class KickRecalcPipeLine implements IPipeLineStarter<SubjectPayload> {
public class KickRecalcPipeLine implements IPipeLineStarter<SubjectPayload> {
private final Logger log = LoggerFactory.getLogger(getClass());
protected final Long packageId;
protected final UserPackageMapper userPackageMapper;
@ -31,7 +31,6 @@ public abstract class KickRecalcPipeLine implements IPipeLineStarter<SubjectPayl
private final int BATCH_SIZE;
private final Long daysToUpdateFlags;
private boolean isInterrupted = false;
protected int offset = 0;
private PipelineContext<PackageVariable> context;
@ -64,21 +63,28 @@ public abstract class KickRecalcPipeLine implements IPipeLineStarter<SubjectPayl
@Override
public void startFillingQueue(BlockingQueue<SubjectPayload> firstQueue) {
try {
log.info("Start kicker: mode:{}, packageId: {};", mode(), packageId);
context.store(PackageVariable.packageId, packageId);
context.store(PackageVariable.cacheMisses, new AtomicInteger());
context.store(PackageVariable.cacheHits, new AtomicInteger());
int limit = BATCH_SIZE;
UserPackage userPackage = userPackageMapper.findPackage(packageId);
List<PhonesMap> page = nextPage(limit);
log.info("Start kicker: mode:{};", mode());
int offset = 0;
List<PhonesMap> page = nextPage(offset);
while (!page.isEmpty()) {
log.debug("Get page {}", page.size());
List<SubjectProfile> subjectProfiles = page.stream().map(phoneFidMap -> SubjectProfile.builder()
.mapId(phoneFidMap.getId())
.phone(phoneFidMap.getPhone())
.fid(phoneFidMap.getFid())
.nameData(NameData.builder()
.middleName(phoneFidMap.getMiddle())
.firstName(phoneFidMap.getFirst())
.lastName(phoneFidMap.getName1())
.birthDate(phoneFidMap.getBirthDt())
.build())
.idData(IdData.builder()
.idNum(phoneFidMap.getSerNum())
.idType(phoneFidMap.getIdType())
.build())
.build()).toList();
try {
SubjectPayload payload = new SubjectPayload(packageId, userPackage.getUserId(), userPackage.getPackageName(), subjectProfiles);
SubjectPayload payload = new SubjectPayload(null, null, null, subjectProfiles);
payload.setContext(context);
firstQueue.put(payload);
} catch (InterruptedException e) {
@ -86,8 +92,8 @@ public abstract class KickRecalcPipeLine implements IPipeLineStarter<SubjectPayl
log.error("Pipeline starter was interrupted!");
throw new PipelineStarterException();
}
offset += limit;
page = nextPage(limit);
offset += BATCH_SIZE;
page = nextPage(offset);
}
} catch (PipelineStarterException e) {
log.info("Pipeline starter shutting down...");
@ -109,7 +115,7 @@ public abstract class KickRecalcPipeLine implements IPipeLineStarter<SubjectPayl
protected PipelineMode mode() {
return PipelineMode.REPEATED_RESTORE;
}
protected List<PhonesMap> nextPage(Integer limit) {
return phonesMapMapper.searchedAtLessThan(daysToUpdateFlags.intValue(), limit, BATCH_SIZE);
protected List<PhonesMap> nextPage(int offset) {
return phonesMapMapper.searchedAtLessThan(daysToUpdateFlags.intValue(), BATCH_SIZE, offset);
}
}

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

@ -6,6 +6,7 @@
<mapper namespace="ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper">
<resultMap id="PhoneMap" type="ru.nbch.credit_tracker.entities.app.PhonesMap">
<result property="id" column="id"/>
<result property="fid" column="fid"/>
<result property="phone" column="phone"/>
<result property="name1" column="name_1"/>
@ -21,7 +22,8 @@
</resultMap>
<select id="findByPhone" resultMap="PhoneMap">
select fid,
select id,
fid,
phone,
name_1,
first,
@ -38,7 +40,8 @@
</select>
<select id="searchedAtLessThan" resultMap="PhoneMap">
select fid,
select id,
fid,
phone,
name_1,
first,
@ -51,7 +54,7 @@
searched_at,
calculated_at
FROM signals_phones_map
WHERE searched_at &gt; now() - make_interval(days => #{daysCount})
WHERE searched_at &lt; now() - make_interval(days => #{daysCount})
ORDER BY id
LIMIT #{limit} OFFSET #{offset}
</select>