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

add batch size as parameter

Browse source
This commit is contained in:
etreschenkov 2025-11-13 13:37:24 +03:00
parent 945694e50d
commit 8ce1948b07
5 changed files with 22 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
credit-tracker.properties
View file

@ -70,3 +70,8 @@ ct.update-flags-interval=${UPDATE_FLAGS_INTERVAL:1}
ct.days-to-update-flags=${DAYS_TO_UPDATE_FLAGS:30}
ct.encoding=windows-1251
#Batch size and global pipeline timeout
ct.batchSize=1000
#in minute
ct.timeout=30

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

@ -33,4 +33,7 @@ public class CreditTrackerSettings {
private Long daysToUpdateFlags;
private String encoding;
private Integer batchSize;
private Integer timeout;
}

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

@ -20,8 +20,8 @@ import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
@Configuration
public class PipelineStarterConfig {
@Bean
public BiFunction<Long, String, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory(StaxXmlProcessor xml) {
return (packageId, packageName) -> new KickPipeLineImpl(xml, packageId, packageName);
public BiFunction<Long, String, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory(StaxXmlProcessor xml, CreditTrackerSettings creditTrackerSettings) {
return (packageId, packageName) -> new KickPipeLineImpl(xml, packageId, packageName, creditTrackerSettings);
}
@Bean

8
src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java
View file

@ -9,6 +9,7 @@ import java.util.function.Supplier;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import ru.nbch.credit_tracker.config.task.StageQueueFactory;
import ru.nbch.credit_tracker.model.impl.PartFilePayload;
import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload;
@ -34,6 +35,7 @@ public class PipelinePackageManager {
private final FidIdEnricher fidIdEnricherHandler;
private final Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory;
private final BiFunction<Long, String, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory;
private final CreditTrackerSettings creditTrackerSettings;
private final ExecutorService executorService = Executors.newFixedThreadPool(10, r -> {
Thread t = new Thread(r);
t.setName("pipe-starter-%d".formatted(t.threadId()));
@ -49,7 +51,8 @@ public class PipelinePackageManager {
SaveSubjectHandler saveSubjectHandler, FidIdEnricher fidIdEnricherHandler,
Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory,
BiFunction<Long, String, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory,
FlagsEnricherV2 flagsEnricherV2) {
FlagsEnricherV2 flagsEnricherV2,
CreditTrackerSettings creditTrackerSettings) {
this.signalRestService = signalRestService;
this.saveSubjectQueueFactory = saveSubjectQueueFactory;
this.idEnricherQueueFactory = idEnricherQueueFactory;
@ -59,6 +62,7 @@ public class PipelinePackageManager {
this.pipelineTerminatorFactory = pipelineTerminatorFactory;
this.pipelineStarterFactory = pipelineStarterFactory;
this.flagsEnricherV2 = flagsEnricherV2;
this.creditTrackerSettings = creditTrackerSettings;
this.semaphore = new Semaphore(MAX_PARALLEL_REQUESTS);
}
@ -86,7 +90,7 @@ public class PipelinePackageManager {
pipeline.getTerminator()
.doneFuture()
.orTimeout(5, TimeUnit.MINUTES)
.orTimeout(creditTrackerSettings.getTimeout(), TimeUnit.MINUTES)
.whenComplete((v, e) -> {
try {
if (e != null) {

7
src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java
View file

@ -10,6 +10,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest;
import ru.nbch.credit_tracker.model.SubjectProfile;
import ru.nbch.credit_tracker.model.impl.SubjectPayload;
@ -23,13 +24,15 @@ public class KickPipeLineImpl implements IPipeLineStarter<SubjectPayload> {
private final String pkgPath;
private final StaxXmlProcessor xmlProcessor;
private final static int BATCH_SIZE = 10_000;
private final int BATCH_SIZE;
public KickPipeLineImpl(StaxXmlProcessor xmlProcessor, Long packageId, String pkgPath) {
public KickPipeLineImpl(StaxXmlProcessor xmlProcessor, Long packageId, String pkgPath,
CreditTrackerSettings creditTrackerSettings) {
this.packageId = packageId;
this.pkgPath = pkgPath;
this.xmlProcessor = xmlProcessor;
this.BATCH_SIZE = creditTrackerSettings.getBatchSize();
}
@Override