конвейер очередей, тестовый вариант

This commit is contained in:
etreschenkov 2025-10-08 17:59:43 +03:00
parent 194828d0b7
commit 7fc7bacbc8
24 changed files with 912 additions and 108 deletions

View file

@ -0,0 +1,126 @@
package ru.nbch.credit_tracker.builder;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.time.LocalDate;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import ru.nbch.credit_tracker.entities.dto.IdProfile;
import ru.nbch.credit_tracker.entities.dto.NameProfile;
import ru.nbch.credit_tracker.entities.dto.SubjectProfile;
import ru.nbch.credit_tracker.entities.external_request.signals.Person;
import ru.nbch.credit_tracker.utils.CTUtil;
@Slf4j
public class PersonFactoryBuilderV2 {
private final static DateFormat DATE_FORMAT = new SimpleDateFormat("dd.MM.yyyy");
private final static String DEFAULT_DATE = "01.01.2000";
private final static String IP = "0";
private final static String OKSM = "999";
private final static String CONSENT_PURPOSE = "16";
private final static String REPORT_USER = "АО «НБКИ»";
private final static String LIABILITY = "1";
private final static String CONSENT_PERIOD = "3";
private final static String REPORT_USER_REG_NUM = "1057746710713";
private final static String REPORT_USER_TAX_ID = "7703548386";
private final static String CONSENT_HASH = "hash";
private final static String CONSENT_DIV_CODE = "000-000";
private final static String INQ_PURPOSE = "16";
private final static String ODT = "Иной документ";
private final static String CONSENT_ISSUE_AUTHORITY = "Кем выдан";
private SubjectProfile personData;
private Integer uid;
public static PersonFactoryBuilderV2 builder() {
return new PersonFactoryBuilderV2();
}
public PersonFactoryBuilderV2 personData(SubjectProfile personData) {
this.personData = personData;
return this;
}
public PersonFactoryBuilderV2 uid(Integer uid) {
this.uid = uid;
return this;
}
public Person build() {
if (personData == null) {
return null;
}
NameProfile nameData = personData.getNameProfile();
IdProfile passportData = personData.getIdProfile();
Person person = new Person();
person.setS(nameData.getLastName());
person.setN(nameData.getFirstName());
person.setM(StringUtils.isNotBlank(nameData.getMiddleName()) ? nameData.getMiddleName() : null);
person.setD(DATE_FORMAT.format(CTUtil.localDatetoDate(nameData.getBirthDate())));
person.setPn(getPn(passportData.getSerNum(), passportData.getIdNum()));
person.setPd(DATE_FORMAT.format(CTUtil.localDatetoDate(LocalDate.now()))); // Do not use passport iss_date. Signal service could reject the package
person.setDt(getDt(passportData.getIdType(), passportData.getIdType2()));
if (person.getDt() != null && person.getDt().equals("999")) {
person.setOdt(ODT);
person.setConsentOdt(ODT);
}
person.setIp(IP);
person.setUid(uid);
person.setOksm(OKSM);
person.setConsentDate(DATE_FORMAT.format(CTUtil.localDatetoDate(LocalDate.now())));
person.setConsentPurpose(CONSENT_PURPOSE);
person.setReportUser(REPORT_USER);
person.setLiability(LIABILITY);
person.setConsentPeriod(CONSENT_PERIOD);
person.setAgrDate(DATE_FORMAT.format(CTUtil.localDatetoDate(LocalDate.now())));
person.setReportUserRegNum(REPORT_USER_REG_NUM);
person.setReportUserTaxID(REPORT_USER_TAX_ID);
person.setConsentHash(CONSENT_HASH);
person.setCs(person.getS());
person.setConsentN(person.getN());
person.setConsentM(person.getM());
person.setConsentBd(person.getD());
person.setConsentPn(person.getPn());
person.setConsentPd(person.getPd());
person.setConsentDivCode(CONSENT_DIV_CODE);
person.setConsentDt(person.getDt());
person.setConsentIssueAuthority(CONSENT_ISSUE_AUTHORITY);
person.setInqPurpose(INQ_PURPOSE);
return person;
}
private String getPn(String serNum, String idNum) {
String normSerNum = CTUtil.normalize(serNum);
String normIdNum = CTUtil.normalize(idNum);
return StringUtils.isNotBlank(normSerNum) ? normSerNum + normIdNum : normIdNum;
}
private String getDt(String idType, String idType2) {
if (StringUtils.isNotBlank(idType2)) {
return idType2;
} else {
return switch (idType) {
case "21" -> "21";
case "22" -> "22.1";
case "09" -> "22.2";
case "26" -> "23";
case "04" -> "24";
case "07" -> "25";
case "14", "03" -> "27";
case "10" -> "31";
case "15" -> "32";
case "13" -> "37";
case "06", "27" -> "28";
case "01", "02" -> "999";
default -> {
log.warn("ID_TYPE2 is blank and no matching found for ID_TYPE {}", idType);
yield "";
}
};
}
}
}

View file

@ -1,15 +0,0 @@
package ru.nbch.credit_tracker.config;
import java.util.ArrayDeque;
import java.util.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class AppConfig {
@Bean
public Queue<Integer> registeredPackages() {
return new ArrayDeque<>();
}
}

View file

@ -5,7 +5,12 @@ import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import ru.nbch.credit_tracker.model.PackageSavePayload;
import ru.nbch.credit_tracker.model.PackageToSavePayload;
import ru.nbch.credit_tracker.model.PartFilePayload;
import ru.nbch.credit_tracker.model.PhoneFidPayload;
import ru.nbch.credit_tracker.model.SubjectsProfilePayload;
import ru.nbch.credit_tracker.service.task.FidIdEnricher;
import ru.nbch.credit_tracker.service.task.FidNameEnricher;
import ru.nbch.credit_tracker.service.task.PackageSaver;
import ru.nbch.credit_tracker.service.task.TaskConsumer;
@ -13,12 +18,39 @@ import ru.nbch.credit_tracker.service.task.TaskConsumer;
public class ConsumerConfig {
@Bean
public BlockingQueue<PackageSavePayload> savingRequest() {
public BlockingQueue<PackageToSavePayload> savingRequest() {
return new LinkedBlockingQueue<>();
}
@Bean
public TaskConsumer<PackageSavePayload> saverTask(PackageSaver packageSaver,
BlockingQueue<PackageSavePayload> savingRequest) {
return new TaskConsumer<>(packageSaver, savingRequest, Executors.newFixedThreadPool(5));
public TaskConsumer<PackageToSavePayload> savingRequestTask(PackageSaver packageSaver,
BlockingQueue<PackageToSavePayload> savingRequest) {
return new TaskConsumer<>(packageSaver, savingRequest, Executors.newSingleThreadScheduledExecutor());
}
@Bean
public BlockingQueue<PhoneFidPayload> fidNameEnricherQueue() {
return new LinkedBlockingQueue<>();
}
@Bean
public TaskConsumer<PhoneFidPayload> fidNameEnricherTask(FidNameEnricher fidNameEnricher,
BlockingQueue<PhoneFidPayload> fidNameEnricherQueue) {
return new TaskConsumer<>(fidNameEnricher, fidNameEnricherQueue, Executors.newFixedThreadPool(10));
}
@Bean
public BlockingQueue<SubjectsProfilePayload> fidIdEnricherQueue() {
return new LinkedBlockingQueue<>();
}
@Bean
public TaskConsumer<SubjectsProfilePayload> fidIdEnricherTask(FidIdEnricher fidIdEnricher,
BlockingQueue<SubjectsProfilePayload> fidIdEnricherQueue) {
return new TaskConsumer<>(fidIdEnricher, fidIdEnricherQueue, Executors.newFixedThreadPool(10));
}
@Bean
public BlockingQueue<PartFilePayload> mergerQueue() {
return new LinkedBlockingQueue<>();
}
}

View file

@ -0,0 +1,20 @@
package ru.nbch.credit_tracker.entities.dto;
import java.time.LocalDate;
import lombok.Builder;
import lombok.Getter;
import lombok.Setter;
@Getter
@Setter
@Builder
public class IdProfile {
private Integer fid;
private String serNum;
private String idNum;
private LocalDate issDate;
private String idType;
private String idType2;
private LocalDate fileSinceDate;
}

View file

@ -0,0 +1,19 @@
package ru.nbch.credit_tracker.entities.dto;
import java.time.LocalDate;
import lombok.Builder;
import lombok.Getter;
import lombok.Setter;
@Getter
@Setter
@Builder
public class NameProfile {
private Integer fid;
private String firstName;
private String lastName;
private String middleName;
private LocalDate birthDate;
private LocalDate fileSinceDate;
}

View file

@ -27,4 +27,13 @@ public class PhoneFidMap {
private String middleName;
private LocalDate birthDate;
private LocalDate fileSinceDate;
//ID data
private String serNum;
private String idNum;
private LocalDate issDate;
private String idType;
private String idType2;
private LocalDate IdfileSinceDate;
boolean isFullFilled = false;
}

View file

@ -0,0 +1,13 @@
package ru.nbch.credit_tracker.entities.dto;
import lombok.Getter;
import lombok.Setter;
@Getter
@Setter
public class SubjectProfile {
private Integer fid;
private IdProfile idProfile;
private NameProfile nameProfile;
private boolean isFilled = false;
}

View file

@ -15,7 +15,7 @@ import ru.nbch.credit_tracker.enums.Status;
import ru.nbch.credit_tracker.enums.XmlErrorFormat;
import ru.nbch.credit_tracker.exceptions.SignalClientException;
import ru.nbch.credit_tracker.model.MonitoringValidationError;
import ru.nbch.credit_tracker.model.PackageSavePayload;
import ru.nbch.credit_tracker.model.PackageToSavePayload;
import ru.nbch.credit_tracker.model.ValidationResult;
import ru.nbch.credit_tracker.service.SignalService;
import ru.nbch.credit_tracker.service.scoring.AuthorizationService;
@ -32,12 +32,12 @@ public class SignalFacadeV2 {
private final CreditTrackerSettings config;
private final SignalService signalService;
private final AuthorizationService authorizationService;
private final BlockingQueue<PackageSavePayload> savingRequest;
private final BlockingQueue<PackageToSavePayload> savingRequest;
public SignalFacadeV2(MonitoringRequestValidator monitoringRequestValidator,
CreditTrackerSettings config,
SignalService signalService,
AuthorizationService authorizationService,
BlockingQueue<PackageSavePayload> savingRequest) {
BlockingQueue<PackageToSavePayload> savingRequest) {
this.monitoringRequestValidator = monitoringRequestValidator;
this.config = config;
this.signalService = signalService;
@ -89,14 +89,14 @@ public class SignalFacadeV2 {
}
// проверка на уникальность пакета
if (signalService.isPackageNotUnique(userId, packageName)) {
signalService.rejectPackage(userId, packageName);
throw clientExcp(SignalError.ERROR_007, XmlErrorFormat.MonitoringError);
}
// if (signalService.isPackageNotUnique(userId, packageName)) {
// signalService.rejectPackage(userId, packageName);
// throw clientExcp(SignalError.ERROR_007, XmlErrorFormat.MonitoringError);
// }
Integer packageId = signalService.registerPackage(userId, packageName);
log.info("Package registered. PackageId {}. Status: {}", packageId, Status.PROCESSING.name());
savingRequest.add(new PackageSavePayload(packageId, filePath));
savingRequest.add(new PackageToSavePayload(packageId, filePath));
} catch (Exception e) {
CTUtil.removeFile(filePath);
if (e instanceof SignalClientException) {

View file

@ -1,20 +1,23 @@
package ru.nbch.credit_tracker.mapper.signal;
import java.util.Collection;
import java.util.List;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
import ru.nbch.credit_tracker.entities.user_request.monitoring_request.Subject;
import java.util.List;
@Mapper
public interface PhoneFidMapMapper {
void registerSubjects(@Param("subjects") List<Subject> subjects, @Param("packageId") Integer packageId);
void register(@Param("phoneFidMap") List<PhoneFidMap> phoneFidMap, @Param("packageId") Integer packageId);
List<PhoneFidMap> findFullPhoneFidMap(@Param("packageId") Integer packageId, @Param("offset") int offset, @Param("limit") int limit);
void updatePhoneFidMaps(@Param("subjects") List<PhoneFidMap> subjects);
void updatePhoneFidMaps(@Param("subjects") Collection<PhoneFidMap> subjects);
void updateOne(@Param("subject") PhoneFidMap subject);
void updateFid(@Param("subjects") Collection<PhoneFidMap> subjects);
void updatePhoneFidMapsErrors(@Param("subjects") List<PhoneFidMap> subjects);

View file

@ -2,11 +2,11 @@ package ru.nbch.credit_tracker.model;
import ru.nbch.credit_tracker.service.task.TaskPayload;
public class PackageSavePayload implements TaskPayload {
public class PackageToSavePayload implements TaskPayload {
private final Integer packageId;
private final String filePath;
public PackageSavePayload(Integer packageId, String filePath) {
public PackageToSavePayload(Integer packageId, String filePath) {
this.packageId = packageId;
this.filePath = filePath;
}

View file

@ -0,0 +1,24 @@
package ru.nbch.credit_tracker.model;
import java.nio.file.Path;
import ru.nbch.credit_tracker.entities.external_request.signals.Signals;
public class PartFilePayload {
private final Path partPath;
private final Integer packageId;
private final Signals signals;
public PartFilePayload(Path partPath, Integer packageId, Signals signals) {
this.partPath = partPath;
this.packageId = packageId;
this.signals = signals;
}
public Path getPartPath() {
return partPath;
}
public Integer getPackageId() {
return packageId;
}
}

View file

@ -0,0 +1,23 @@
package ru.nbch.credit_tracker.model;
import java.util.List;
import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
import ru.nbch.credit_tracker.service.task.TaskPayload;
public class PhoneFidPayload implements TaskPayload {
private final Integer packageId;
private final List<PhoneFidMap> phoneFidsBatch;
public PhoneFidPayload(Integer packageId, List<PhoneFidMap> phoneFidsBatch) {
this.packageId = packageId;
this.phoneFidsBatch = phoneFidsBatch;
}
public Integer getPackageId() {
return packageId;
}
public List<PhoneFidMap> getPhoneFidsBatch() {
return phoneFidsBatch;
}
}

View file

@ -0,0 +1,23 @@
package ru.nbch.credit_tracker.model;
import java.util.List;
import ru.nbch.credit_tracker.entities.dto.SubjectProfile;
import ru.nbch.credit_tracker.service.task.TaskPayload;
public class SubjectsProfilePayload implements TaskPayload {
private final Integer packageId;
private final List<SubjectProfile> subjectProfiles;
public SubjectsProfilePayload(Integer packageId, List<SubjectProfile> subjectProfiles) {
this.packageId = packageId;
this.subjectProfiles = subjectProfiles;
}
public Integer getPackageId() {
return packageId;
}
public List<SubjectProfile> getSubjectProfiles() {
return subjectProfiles;
}
}

View file

@ -271,7 +271,7 @@ public class SignalService {
return filePath;
}
private Signals buildSignals(Integer packageId) {
public Signals buildSignals(Integer packageId) {
UserPackage userPackage = userPackageService.findPackage(packageId);
Signals signals = new Signals();
signals.setOnline("2"); // Всегда=2 (только онлайн-сигналы)

View file

@ -1,19 +1,20 @@
package ru.nbch.credit_tracker.service.indic;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import ru.nbch.credit_tracker.mapper.indic.PersonMapper;
import java.time.LocalDate;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import ru.nbch.credit_tracker.entities.dto.IdProfile;
import ru.nbch.credit_tracker.entities.dto.SubjectProfile;
import ru.nbch.credit_tracker.mapper.indic.PersonMapper;
@Slf4j
@Service
@ -51,22 +52,22 @@ public class PersonService {
private Map<Integer, NameData> findNameData(List<Integer> fids) {
List<NameData> persons = mapper.findNameData(fids);
return persons.stream()
.collect(Collectors.toMap(NameData::getFid, nameData -> nameData,
(existing, replacement) -> {
LocalDate existingDate = existing.getFileSinceDate();
LocalDate replacementDate = replacement.getFileSinceDate();
if (existingDate != null && replacementDate != null) {
if (existingDate.isAfter(replacementDate)) {
return existing;
} else {
return replacement;
}
} else if (existingDate != null) {
return existing;
} else {
return replacement;
}
}));
.collect(Collectors.toMap(NameData::getFid, nameData -> nameData,
(existing, replacement) -> {
LocalDate existingDate = existing.getFileSinceDate();
LocalDate replacementDate = replacement.getFileSinceDate();
if (existingDate != null && replacementDate != null) {
if (existingDate.isAfter(replacementDate)) {
return existing;
} else {
return replacement;
}
} else if (existingDate != null) {
return existing;
} else {
return replacement;
}
}));
}
private Map<Integer, PassportData> findPassportData(List<Integer> fids) {
@ -75,7 +76,7 @@ public class PersonService {
List<PassportData> passports = mapper.findPassportData(fids);
Map<Integer, List<PassportData>> passportsMap = passports.stream()
.collect(Collectors.groupingBy(PassportData::getFid));
.collect(Collectors.groupingBy(PassportData::getFid));
fids.forEach(fid -> {
PassportData passport = findPassport(passportsMap.get(fid));
@ -87,13 +88,37 @@ public class PersonService {
return resultMap;
}
public void filledPassportsData(List<SubjectProfile> subjectProfiles) {
List<Integer> fids = subjectProfiles.stream().map(SubjectProfile::getFid).collect(Collectors.toList());
List<PassportData> passports = mapper.findPassportData(fids);
Map<Integer, List<PassportData>> passportsMap = passports.stream()
.collect(Collectors.groupingBy(PassportData::getFid));
subjectProfiles.forEach(subjectProfile -> {
PassportData passport = findPassport(passportsMap.get(subjectProfile.getFid()));
if (passport != null) {
IdProfile idProfile = IdProfile.builder()
.serNum(passport.getSerNum())
.idNum(passport.getIdNum())
.issDate(passport.getIssDate())
.idType(passport.getIdType())
.idType2(passport.getIdType2())
.fileSinceDate(passport.getFileSinceDate())
.build();
subjectProfile.setIdProfile(idProfile);
subjectProfile.setFilled(true);
}
});
}
private PassportData findPassport(List<PassportData> passports) {
if (passports == null || passports.isEmpty()) {
return null;
}
List<PassportData> documents = passports.stream()
.filter(this::isDUL)
.toList(); // берем только ДУЛ
.filter(this::isDUL)
.toList(); // берем только ДУЛ
if (documents.isEmpty()) {
return null;
@ -101,16 +126,16 @@ public class PersonService {
// Отдаем приоритет паспортам РФ
PassportData passport = documents.stream()
.filter(p -> "21".equals(p.getIdType2()) || "21".equals(p.getIdType()))
.sorted(
Comparator.comparing(
PassportData::getIssDate,
Comparator.nullsLast(Comparator.reverseOrder())
).thenComparing(
PassportData::getFileSinceDate,
Comparator.nullsLast(Comparator.reverseOrder())
)
).findFirst().orElse(null);
.filter(p -> "21".equals(p.getIdType2()) || "21".equals(p.getIdType()))
.sorted(
Comparator.comparing(
PassportData::getIssDate,
Comparator.nullsLast(Comparator.reverseOrder())
).thenComparing(
PassportData::getFileSinceDate,
Comparator.nullsLast(Comparator.reverseOrder())
)
).findFirst().orElse(null);
if (passport != null) {
return passport;
@ -118,15 +143,15 @@ public class PersonService {
// Если паспорт РФ не найден, смотри остальные ДУЛ'ы и берем первый
passport = documents.stream()
.sorted(
Comparator.comparing(
PassportData::getIssDate,
Comparator.nullsLast(Comparator.reverseOrder())
).thenComparing(
PassportData::getFileSinceDate,
Comparator.nullsLast(Comparator.reverseOrder())
)
).findFirst().orElse(null);
.sorted(
Comparator.comparing(
PassportData::getIssDate,
Comparator.nullsLast(Comparator.reverseOrder())
).thenComparing(
PassportData::getFileSinceDate,
Comparator.nullsLast(Comparator.reverseOrder())
)
).findFirst().orElse(null);
return passport;
}

View file

@ -99,19 +99,22 @@ public class PhoneService2 {
if (fidsData != null) {
phoneFid.setFid(fidsData.getFid());
PhoneRepository.FidsDataWithName nameData = defineNameData(fidFidsDataMap.get(fidsData.getFid()));
List<PhoneRepository.FidsDataWithName> fidsDataWithNames = fidFidsDataMap.get(fidsData.getFid());
if (fidsDataWithNames != null) {
PhoneRepository.FidsDataWithName nameData = defineNameData(fidsDataWithNames);
if (nameData != null) {
phoneFid.setFirstName(nameData.getFirstName());
phoneFid.setMiddleName(nameData.getMiddleName());
phoneFid.setLastName(nameData.getLastName());
phoneFid.setBirthDate(nameData.getBirthDate());
phoneFid.setFileSinceDate(nameData.getFileSinceDateName());
if (nameData != null) {
phoneFid.setFirstName(nameData.getFirstName());
phoneFid.setMiddleName(nameData.getMiddleName());
phoneFid.setLastName(nameData.getLastName());
phoneFid.setBirthDate(nameData.getBirthDate());
phoneFid.setFileSinceDate(nameData.getFileSinceDateName());
}
phoneFidMap.put(fidsData.getFid(), phoneFid);
phoneToFidsMap.remove(phone);
fidFidsDataMap.remove(fidsData.getFid());
}
phoneFidMap.put(fidsData.getFid(), phoneFid);
phoneToFidsMap.remove(phone);
fidFidsDataMap.remove(fidsData.getFid());
}
}
log.debug("End merging fids and names");

View file

@ -1,33 +1,56 @@
package ru.nbch.credit_tracker.service.signal;
import java.util.Collection;
import java.util.List;
import org.apache.ibatis.session.ExecutorType;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
import ru.nbch.credit_tracker.entities.user_request.monitoring_request.Subject;
import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
import java.util.List;
@Service
public class PhoneFidMapService {
private final PhoneFidMapMapper mapper;
public PhoneFidMapService(PhoneFidMapMapper mapper) {
private final SqlSessionFactory sqlSessionFactory;
public PhoneFidMapService(PhoneFidMapMapper mapper,
@Qualifier("sqlSessionSignal") SqlSessionFactory sqlSessionFactory) {
this.mapper = mapper;
this.sqlSessionFactory = sqlSessionFactory;
}
public void registerSubjects(List<Subject> subjects, Integer packageId) {
mapper.registerSubjects(subjects, packageId);
}
public void register(List<PhoneFidMap> phoneFidMap, Integer packageId) {
mapper.register(phoneFidMap, packageId);
}
public List<PhoneFidMap> findFullPhoneFidMap(Integer packageId, int offset, int limit) {
return mapper.findFullPhoneFidMap(packageId, offset, limit);
}
public void updatePhoneFidMaps(List<PhoneFidMap> subjects) {
public void updatePhoneFidMaps(Collection<PhoneFidMap> subjects) {
mapper.updatePhoneFidMaps(subjects);
}
public void updatePhoneFidMaps2(Collection<PhoneFidMap> subjects) {
final int batchSize = 1000;
try (SqlSession session = sqlSessionFactory.openSession(ExecutorType.BATCH)) {
PhoneFidMapMapper mapper = session.getMapper(PhoneFidMapMapper.class);
int i = 0;
for (PhoneFidMap s : subjects) {
mapper.updateOne(s);
if (++i % batchSize == 0) session.flushStatements();
}
session.flushStatements();
session.commit();
}
}
public void updatePhoneFidMapsErrors(List<PhoneFidMap> subjects) {
mapper.updatePhoneFidMapsErrors(subjects);
}

View file

@ -0,0 +1,87 @@
package ru.nbch.credit_tracker.service.task;
import java.io.IOException;
import java.io.OutputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardCopyOption;
import java.nio.file.StandardOpenOption;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import ru.nbch.credit_tracker.builder.PersonFactoryBuilderV2;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import ru.nbch.credit_tracker.entities.dto.SubjectProfile;
import ru.nbch.credit_tracker.entities.external_request.signals.Person;
import ru.nbch.credit_tracker.entities.external_request.signals.Signals;
import ru.nbch.credit_tracker.model.PartFilePayload;
import ru.nbch.credit_tracker.model.SubjectsProfilePayload;
import ru.nbch.credit_tracker.service.SignalService;
import ru.nbch.credit_tracker.service.indic.PersonService;
import ru.nbch.credit_tracker.service.xml.stax.SignalsStaxXmlWriterV2;
@Service
@Slf4j
public class FidIdEnricher implements Dispatcher<SubjectsProfilePayload>{
private final PersonService personService;
private final SignalService signalService;
private final CreditTrackerSettings config;
private final BlockingQueue<PartFilePayload> mergerQueue;
public FidIdEnricher(PersonService personService,
SignalService signalService,
CreditTrackerSettings config,
BlockingQueue<PartFilePayload> mergerQueue) {
this.personService = personService;
this.signalService = signalService;
this.config = config;
this.mergerQueue = mergerQueue;
}
@Override
public void handle(SubjectsProfilePayload taskPayload) {
log.debug("FidIdEnricher handle by packageId: {}", taskPayload.getPackageId());
List<SubjectProfile> subjectsProfile = taskPayload.getSubjectProfiles();
personService.filledPassportsData(subjectsProfile);
List<Person> personOnProcessing = subjectsProfile.stream()
.filter(SubjectProfile::isFilled)
.map(subjectProfile -> PersonFactoryBuilderV2.builder().personData(subjectProfile).build())
.toList();
writeSignals(personOnProcessing, taskPayload.getPackageId());
}
private void writeSignals(List<Person> persons, Integer packageId){
Path dir = Paths.get(config.getLoadDir(), "tmp_folder");
try {
Files.createDirectories(dir);
} catch (IOException e) {
throw new RuntimeException(e);
}
String base = UUID.randomUUID() + "_output.xml";
log.debug("Write tmp xml by packageId: {}", packageId);
Path tmp = dir.resolve(base + ".part");
Path done = dir.resolve(base);
Signals signals = signalService.buildSignals(packageId);
try (OutputStream os = Files.newOutputStream(tmp, StandardOpenOption.CREATE_NEW)) {
SignalsStaxXmlWriterV2 writer = new SignalsStaxXmlWriterV2(persons, signals, config.getEncoding());
writer.writeWithStax(os); // перегрузи на OutputStream, если нужно
} catch (IOException e) {
throw new RuntimeException(e);
}
try {
Files.move(tmp, done, StandardCopyOption.ATOMIC_MOVE);
} catch (IOException e) {
throw new RuntimeException(e);
}
try {
mergerQueue.put(new PartFilePayload(done, packageId, signals));
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}

View file

@ -0,0 +1,61 @@
package ru.nbch.credit_tracker.service.task;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import ru.nbch.credit_tracker.entities.dto.NameProfile;
import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
import ru.nbch.credit_tracker.entities.dto.SubjectProfile;
import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
import ru.nbch.credit_tracker.model.PhoneFidPayload;
import ru.nbch.credit_tracker.model.SubjectsProfilePayload;
import ru.nbch.credit_tracker.service.indic.PhoneService2;
import ru.nbch.credit_tracker.service.signal.PhoneFidMapService;
@Service
@Slf4j
public class FidNameEnricher implements Dispatcher<PhoneFidPayload> {
private final PhoneService2 phoneService2;
private final PhoneFidMapService phoneFidMapService;
private final PhoneFidMapMapper mapper;
private final Queue<SubjectsProfilePayload> fidIdEnricherQueue;
public FidNameEnricher(PhoneService2 phoneService2,
PhoneFidMapService phoneFidMapService,
PhoneFidMapMapper mapper,
Queue<SubjectsProfilePayload> fidIdEnricherQueue) {
this.phoneService2 = phoneService2;
this.phoneFidMapService = phoneFidMapService;
this.mapper = mapper;
this.fidIdEnricherQueue = fidIdEnricherQueue;
}
@Override
public void handle(PhoneFidPayload taskPayload) {
log.debug("FidNameEnricher handle by packageId: {}", taskPayload.getPackageId());
List<PhoneFidMap> phoneFidMap = taskPayload.getPhoneFidsBatch();
phoneFidMapService.register(phoneFidMap, taskPayload.getPackageId());
Map<Integer, PhoneFidMap> fidsMapByBatch = phoneService2.defineFidsAndNameByPhone(phoneFidMap);
log.debug("Found {} fids by batch", fidsMapByBatch.size());
//todo по phoneFidMap может быть найден не полный набор фидов, поэтому стоит это контролировать
mapper.updateFid(fidsMapByBatch.values());
List<SubjectProfile> subjectProfiles = fidsMapByBatch.values().stream().map(pf -> {
SubjectProfile subjectProfile = new SubjectProfile();
subjectProfile.setFid(pf.getFid());
subjectProfile.setNameProfile(NameProfile.builder()
.firstName(pf.getFirstName())
.middleName(pf.getMiddleName())
.lastName(pf.getLastName())
.birthDate(pf.getBirthDate())
.fileSinceDate(pf.getFileSinceDate())
.build());
return subjectProfile;
}).toList();
SubjectsProfilePayload subjectsProfilePayload = new SubjectsProfilePayload(taskPayload.getPackageId(), subjectProfiles);
fidIdEnricherQueue.add(subjectsProfilePayload);
}
}

View file

@ -5,31 +5,35 @@ import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.List;
import java.util.Queue;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest;
import ru.nbch.credit_tracker.model.PackageSavePayload;
import ru.nbch.credit_tracker.model.PhoneFidPayload;
import ru.nbch.credit_tracker.model.PackageToSavePayload;
import ru.nbch.credit_tracker.service.signal.PhoneFidMapService;
import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
@Service
@Slf4j
public class PackageSaver implements Dispatcher<PackageSavePayload> {
public class PackageSaver implements Dispatcher<PackageToSavePayload> {
private final StaxXmlProcessor xmlProcessor;
private final PhoneFidMapService phoneFidMapService;
private final Queue<Integer> registeredPackage;
private final Queue<PhoneFidPayload> fidNameEnricherQueue;
private final static int BATCH_SIZE = 10_000;
public PackageSaver(StaxXmlProcessor xmlProcessor,
PhoneFidMapService phoneFidMapService,
Queue<Integer> registeredPackage) {
Queue<PhoneFidPayload> fidNameEnricherQueue) {
this.xmlProcessor = xmlProcessor;
this.phoneFidMapService = phoneFidMapService;
this.registeredPackage = registeredPackage;
this.fidNameEnricherQueue = fidNameEnricherQueue;
}
@Override
public void handle(PackageSavePayload taskPayload) {
public void handle(PackageToSavePayload taskPayload) {
Integer packageId = taskPayload.getPackageId();
Path pkgPath = Paths.get(taskPayload.getFilePath());
log.info("Start registering subjects. PackageId: {}", taskPayload.getPackageId());
@ -37,8 +41,13 @@ public class PackageSaver implements Dispatcher<PackageSavePayload> {
xmlProcessor.read(in, MonitoringRequest.class,
request -> {
if (!request.getSubjects().getSubject().isEmpty()) {
phoneFidMapService.registerSubjects(request.getSubjects().getSubject(), packageId);
log.trace("Registered subject's batch. Total count: {}. PackageId: {}", request.getSubjects().getSubject().size(), packageId);
List<PhoneFidMap> phoneFidMap = request.getSubjects().getSubject().stream().map(subject -> PhoneFidMap.builder()
.subjectId(subject.getId())
.phone(subject.getClientId())
.packageId(packageId)
.build()).collect(Collectors.toList());
log.trace("Reading subject's batch. Total count: {}. PackageId: {}", request.getSubjects().getSubject().size(), packageId);
fidNameEnricherQueue.add(new PhoneFidPayload(packageId, phoneFidMap));
request.getSubjects().getSubject().clear();
}
},
@ -49,6 +58,5 @@ public class PackageSaver implements Dispatcher<PackageSavePayload> {
throw new RuntimeException("Error while parsing monitoring request", e);
}
log.info("Subjects were registered. PackageId: {}", packageId);
registeredPackage.add(packageId);
}
}

View file

@ -2,8 +2,12 @@ package ru.nbch.credit_tracker.service.task;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.springframework.context.SmartLifecycle;
@Slf4j
public class TaskConsumer<X extends TaskPayload> implements SmartLifecycle {
private final Dispatcher<X> dispatcher;
private final BlockingQueue<X> taskQueue;
@ -18,18 +22,19 @@ public class TaskConsumer<X extends TaskPayload> implements SmartLifecycle {
this.worker = worker;
}
@Override
public void start() {
running = true;
worker.submit(() -> {
ExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
executorService.submit(() -> {
while (running && !Thread.currentThread().isInterrupted()) {
try {
X job = taskQueue.take();
dispatcher.handle(job);
worker.submit(() -> dispatcher.handle(job));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} catch (Exception e) {
log.error("Error: {}", ExceptionUtils.getStackTrace(e));
}
}
});

View file

@ -0,0 +1,166 @@
package ru.nbch.credit_tracker.service.task;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.charset.Charset;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import javax.xml.stream.XMLInputFactory;
import javax.xml.stream.XMLOutputFactory;
import javax.xml.stream.XMLStreamConstants;
import javax.xml.stream.XMLStreamException;
import javax.xml.stream.XMLStreamReader;
import javax.xml.stream.XMLStreamWriter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import ru.nbch.credit_tracker.model.PartFilePayload;
@Slf4j
@Service
public final class XmlMergeCoordinator implements InitializingBean {
private final BlockingQueue<PartFilePayload> mergerQueue;
private final Path mergedDir;
private final Charset encoding = Charset.forName("windows-1251");
private final int batchSize = 10;
private final CreditTrackerSettings config;
private final ExecutorService mergerExec = Executors.newSingleThreadExecutor();
public XmlMergeCoordinator(BlockingQueue<PartFilePayload> mergerQueue,
CreditTrackerSettings config) throws IOException {
this.mergerQueue = mergerQueue;
this.mergedDir = Path.of(config.getLoadDir() + File.separator + "tmp_folder" + File.separator + "merged" + File.separator);
this.config = config;
Files.createDirectories(mergedDir);
}
public void start() {
mergerExec.submit(this::runLoop);
}
public void shutdown() {
mergerExec.shutdownNow();
}
private void runLoop() {
List<PartFilePayload> batch = new ArrayList<>(batchSize);
while (!Thread.currentThread().isInterrupted()) {
try {
PartFilePayload first = mergerQueue.poll(5, TimeUnit.SECONDS);
if (first != null) batch.add(first);
// добираем до batchSize, но не ждём бесконечно
while (batch.size() < batchSize) {
PartFilePayload p = mergerQueue.poll(1000, TimeUnit.MILLISECONDS);
if (p == null) {
if (first != null) {
log.debug("Timeout poll mergerQueue, break");
}
break;
}
batch.add(p);
}
if (!batch.isEmpty()) {
log.debug("Starting merge batch; size: {}", batch.size());
mergeBatch(batch);
// удаляем исходники после успешного мерджа
for (PartFilePayload p : batch) Files.deleteIfExists(p.getPartPath());
batch.clear();
}
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
} catch (Exception e) {
log.error("Error: {}", ExceptionUtils.getStackTrace(e));
batch.clear();
}
}
}
private void mergeBatch(List<PartFilePayload> inputs) throws Exception {
Integer packageId = inputs.stream().findFirst().get().getPackageId();
String outName = "merged_" + UUID.randomUUID() + ".xml";
Path tmp = mergedDir.resolve(outName + ".part");
Path out = mergedDir.resolve(outName);
XMLOutputFactory xof = XMLOutputFactory.newFactory();
try (OutputStream os = Files.newOutputStream(tmp, StandardOpenOption.CREATE_NEW)) {
XMLStreamWriter w = xof.createXMLStreamWriter(os, encoding.name());
w.writeStartDocument(encoding.name(), "1.0");
w.writeStartElement("Persons");
for (PartFilePayload in : inputs) {
copyPersonsChildren(in.getPartPath(), w);
}
w.writeEndElement();
w.writeEndDocument();
w.flush();
w.close();
}
Files.move(tmp, out, StandardCopyOption.ATOMIC_MOVE);
}
/**
* Копирует всё содержимое внутри <Persons>...</Persons> входного файла
* в текущий writer, но без повторного открытия/закрытия корневого тега.
*/
private void copyPersonsChildren(Path input, XMLStreamWriter w) throws Exception {
XMLInputFactory xif = XMLInputFactory.newFactory();
try (InputStream is = Files.newInputStream(input)) {
XMLStreamReader r = xif.createXMLStreamReader(is, encoding.name());
while (r.hasNext()) {
int t = r.next();
if (t == XMLStreamConstants.START_ELEMENT && "Persons".equals(r.getLocalName())) {
break;
}
}
int depth = 0; // 0 внутри корневого контейнера
while (r.hasNext()) {
int t = r.next();
if (t == XMLStreamConstants.START_ELEMENT) {
depth++;
writeStart(w, r);
} else if (t == XMLStreamConstants.CHARACTERS) {
if (!r.isWhiteSpace()) w.writeCharacters(r.getText());
} else if (t == XMLStreamConstants.CDATA) {
w.writeCData(r.getText());
} else if (t == XMLStreamConstants.END_ELEMENT) {
if (depth > 0) {
w.writeEndElement();
depth--;
} else if ("Persons".equals(r.getLocalName())) {
break;
}
}
}
r.close();
}
}
private void writeStart(XMLStreamWriter w, XMLStreamReader r) throws XMLStreamException {
w.writeStartElement(r.getLocalName());
for (int i = 0; i < r.getAttributeCount(); i++) {
w.writeAttribute(r.getAttributeLocalName(i), r.getAttributeValue(i));
}
}
@Override
public void afterPropertiesSet() throws Exception {
mergerExec.submit(this::runLoop);
}
}

View file

@ -0,0 +1,114 @@
package ru.nbch.credit_tracker.service.xml.stax;
import java.io.OutputStream;
import java.util.List;
import javax.xml.stream.XMLEventFactory;
import javax.xml.stream.XMLEventWriter;
import javax.xml.stream.XMLOutputFactory;
import lombok.extern.slf4j.Slf4j;
import ru.nbch.credit_tracker.entities.external_request.signals.Auth;
import ru.nbch.credit_tracker.entities.external_request.signals.Person;
import ru.nbch.credit_tracker.entities.external_request.signals.Signals;
@Slf4j
//todo сделать с этим что-то!!!
public class SignalsStaxXmlWriterV2 {
private final List<Person> personList;
private final Signals signals;
private final String charsetName;
public SignalsStaxXmlWriterV2(List<Person> personList,
Signals signals,
String charsetName) {
this.personList = personList;
this.signals = signals;
this.charsetName = charsetName;
}
public void writeWithStax(OutputStream ou) {
try {
XMLOutputFactory output = XMLOutputFactory.newInstance();
XMLEventFactory xmlEventFactory = XMLEventFactory.newInstance();
XMLEventWriter writer = output.createXMLEventWriter(ou, charsetName);
writer = new IndentingXMLEventWriter(writer);
WriterWrapper writerWrapper = new WriterWrapper(xmlEventFactory, writer);
// writerWrapper.addStartDocument(charsetName);
//
// writerWrapper.addStartElement("Signals");
// writerWrapper.addAttribute("online", signals.getOnline());
// writerWrapper.addAttribute("processWithErrors", signals.getProcessWithErrors());
// writerWrapper.addSimplelement("Id", signals.getId());
// writeAuth(writerWrapper, signals.getAuth());
//
// writerWrapper.addStartElement("Conditions");
writerWrapper.addStartElement("Persons");
writePersons(writerWrapper, personList);
writerWrapper.addEndElement("Persons");
// writerWrapper.addEndElement("Conditions");
// writerWrapper.addEndElement("Signals");
writerWrapper.addEndDocument();
writerWrapper.flush();
System.gc();
} catch (Throwable e) {
log.error("Error serializing object to XML string: {}", e.getMessage(), e);
throw new IllegalStateException("cannot serialize object to XML");
}
}
private void writePersons(WriterWrapper writerWrapper, List<Person> persons) throws Exception {
writerWrapper.writeCollection("p", persons, (obj, name) -> writePerson(writerWrapper, obj, name));
}
private void writeAuth(WriterWrapper writerWrapper, Auth auth) throws Exception {
writerWrapper.addStartElement("Auth");
writerWrapper.addSimplelement("User", auth.getUser());
writerWrapper.addEndElement("Auth");
}
public void writePerson(WriterWrapper writerWrapper, Person person, String name) throws Exception {
if (person == null) {
return;
}
writerWrapper.addStartElement(name);
writerWrapper.addAttribute("uid", String.valueOf(person.getUid()));
writerWrapper.addAttribute("oksm", person.getOksm());
writerWrapper.addAttribute("s", person.getS());
writerWrapper.addAttribute("n", person.getN());
writerWrapper.addAttribute("m", person.getM());
writerWrapper.addAttribute("d", person.getD());
writerWrapper.addAttribute("pn", person.getPn());
writerWrapper.addAttribute("pd", person.getPd());
writerWrapper.addAttribute("dt", person.getDt());
writerWrapper.addAttribute("odt", person.getOdt());
writerWrapper.addAttribute("ip", person.getIp());
writerWrapper.addAttribute("consentDate", person.getConsentDate());
writerWrapper.addAttribute("consentPurpose", person.getConsentPurpose());
writerWrapper.addAttribute("reportUser", person.getReportUser());
writerWrapper.addAttribute("liability", person.getLiability());
writerWrapper.addAttribute("consentPeriod", person.getConsentPeriod());
writerWrapper.addAttribute("agrDate", person.getAgrDate());
writerWrapper.addAttribute("reportUserRegNum", person.getReportUserRegNum());
writerWrapper.addAttribute("reportUserTaxID", person.getReportUserTaxID());
writerWrapper.addAttribute("consentHash", person.getConsentHash());
writerWrapper.addAttribute("cS", person.getCs());
writerWrapper.addAttribute("consentN", person.getConsentN());
writerWrapper.addAttribute("consentM", person.getConsentM());
writerWrapper.addAttribute("consentBd", person.getConsentBd());
writerWrapper.addAttribute("consentPn", person.getConsentPn());
writerWrapper.addAttribute("consentPd", person.getConsentPd());
writerWrapper.addAttribute("consentDivCode", person.getConsentDivCode());
writerWrapper.addAttribute("consentDt", person.getConsentDt());
writerWrapper.addAttribute("consentOdt", person.getConsentOdt());
writerWrapper.addAttribute("consentIssueAuthority", person.getConsentIssueAuthority());
writerWrapper.addAttribute("inqPurpose", person.getInqPurpose());
writerWrapper.addEndElement(name);
}
}

View file

@ -25,6 +25,18 @@
</foreach>
</insert>
<insert id="register" parameterType="java.util.List"
useGeneratedKeys="true"
keyProperty="phoneFidMap.id"
keyColumn="id">
INSERT INTO signals_phone_fid_map (subject_id, phone, package_id)
values
<foreach collection="phoneFidMap" item="phoneFid" separator=",">
(#{phoneFid.subjectId}, #{phoneFid.phone}, #{packageId})
</foreach>
RETURNING id
</insert>
<update id="updatePhoneFidMaps" parameterType="java.util.List">
<foreach collection="subjects" item="subject" separator=";">
@ -42,6 +54,29 @@
</foreach>
</update>
<update id="updateOne" parameterType="ru.nbch.credit_tracker.entities.dto.PhoneFidMap">
UPDATE signals_phone_fid_map
SET
package_id = #{subject.packageId},
subject_id = #{subject.subjectId},
phone = #{subject.phone},
fid = #{subject.fid2},
flag_debt5000 = #{subject.flagDebt5000},
flag_90_12 = #{subject.flag90days},
calculated_at = #{subject.calculatedAt},
error_text = #{subject.errorText}
WHERE id = #{subject.id}
</update>
<update id="updateFid" parameterType="java.util.List">
<foreach collection="subjects" item="subject" separator=";">
UPDATE signals_phone_fid_map
SET
fid = #{subject.fid}
WHERE id = #{subject.id}
</foreach>
</update>
<update id="updatePhoneFidMapsErrors" parameterType="java.util.List">
<foreach collection="subjects" item="subject" separator=";">
UPDATE signals_phone_fid_map