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

./

Browse source
This commit is contained in:
etreschenkov 2025-11-25 19:43:11 +03:00
parent 0f98d8d28d
commit c0cd36cd0a
6 changed files with 69 additions and 59 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java
View file

@ -25,7 +25,7 @@ public class StageConfig {
@Bean("partFileQueueFactory")
public StageQueueFactory<PartFilePayload> partFileQueueFactory() {
return new StageQueueFactory<>(100);
return new StageQueueFactory<>(500);
}
}

9
src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java
View file

@ -37,8 +37,15 @@ public class PhoneRepository {
SELECT
pn.number,
pn.fid,
pn.file_since_dt
pn.file_since_dt,
nf.first,
nf.middle,
nf.name_1,
nf.birth_dt,
nf.file_since_dt as file_since_dt_name
FROM PHONE_NORM pn
JOIN NAME nf
ON nf.fid = pn.fid
WHERE number in (:phones) WITH UR
""";

70
src/main/java/ru/nbch/credit_tracker/service/task/Stage.java
View file

@ -67,7 +67,7 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
@Override
public void run() {
List<Future<Z>> inflight = new ArrayList<>();
List<Future<Void>> inflight = new ArrayList<>();
log.info("this is main thread for {}", dispatcherName);
try {
int counter = 0;
@ -79,7 +79,7 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
if (b.isPoison()) {
log.info("Got poison");
// дождаться всех in-flight задач
for (Future<Z> f : inflight) {
for (Future<Void> f : inflight) {
if (wasInterrupted()) {
log.info("stage {} was interrupted while waiting for tasks to finish", dispatcherName);
break;
@ -94,44 +94,44 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
counter++;
int finalCounter = counter;
AtomicInteger retryForSingleTask = new AtomicInteger(0);
Future<Z> f = ioPool.submit(() -> {
while (retryForSingleTask.incrementAndGet() < pipelineBackoff.getRetryCount()) {
try {
if (Thread.currentThread().isInterrupted()) {
log.info("stage {} was interrupted. skipping task {}...", dispatcherName, finalCounter);
return null;
}
log.info("stage {} task {} started.", dispatcher.getClass().getSimpleName(), finalCounter);
Z p;
try (var m = stopWatch.start(dispatcherName + "-handler")){
p = dispatcher.handle(b);
}
if (Thread.currentThread().isInterrupted()) {
log.info("after {} task {} finished, worker has been interrupted. not putting result in queue.",
dispatcherName, finalCounter);
return null;
}
putQuiet(out, p);
log.info("stage {} task {} ended.", dispatcher.getClass().getSimpleName(), finalCounter);
return p;
} catch (Exception e) {
boolean isRetryable = pipelineBackoff.isRetryable(e);
if (!isRetryable) {
log.error("{} ERROR: {}", dispatcherName, ExceptionUtils.getStackTrace(e));
pipeline.close();
return null;
}
log.warn("{} RETRY AFTER ERROR: {}", dispatcherName, ExceptionUtils.getStackTrace(e));
pipelineBackoff.backoff();
Future<Void> f = ioPool.submit(() -> {
while (retryForSingleTask.incrementAndGet() < pipelineBackoff.getRetryCount()) {
try {
if (Thread.currentThread().isInterrupted()) {
log.info("stage {} was interrupted. skipping task {}...", dispatcherName, finalCounter);
return null;
}
log.info("stage {} task {} started.", dispatcher.getClass().getSimpleName(), finalCounter);
Z p;
try (var m = stopWatch.start(dispatcherName + "-handler")) {
p = dispatcher.handle(b);
}
if (Thread.currentThread().isInterrupted()) {
log.info("after {} task {} finished, worker has been interrupted. not putting result in queue.",
dispatcherName, finalCounter);
return null;
}
putQuiet(out, p);
log.info("stage {} task {} ended.", dispatcher.getClass().getSimpleName(), finalCounter);
return null;
} catch (Exception e) {
boolean isRetryable = pipelineBackoff.isRetryable(e);
if (!isRetryable) {
log.error("{} ERROR: {}", dispatcherName, ExceptionUtils.getStackTrace(e));
pipeline.close();
return null;
}
log.warn("{} RETRY AFTER ERROR: {}", dispatcherName, ExceptionUtils.getStackTrace(e));
pipelineBackoff.backoff();
}
log.error("tried for {} times, failed batch...", retryForSingleTask.get());
pipeline.close();
return null;
}
);
log.error("tried for {} times, failed batch...", retryForSingleTask.get());
pipeline.close();
return null;
});
inflight.add(f);
if ((counter % 20) == 0) {
log.debug("Clean inflight future");
inflight.removeIf(Future::isDone);
}
}

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

@ -4,10 +4,10 @@ import java.io.BufferedInputStream;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
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;
@ -66,6 +66,7 @@ public class KickPipeLineImpl implements IPipeLineStarter<SubjectPayload> {
while (!page.isEmpty()) {
log.debug("Get page {}", page.size());
List<SubjectProfile> subjectProfiles = page.stream().map(phoneFidMap -> SubjectProfile.builder()
.mapId(phoneFidMap.getId())
.identifier(phoneFidMap.getSubjectId())
.phone(phoneFidMap.getPhone())
.build()).toList();
@ -104,17 +105,27 @@ public class KickPipeLineImpl implements IPipeLineStarter<SubjectPayload> {
xmlProcessor.read(in, MonitoringRequest.class,
request -> {
if (!request.getSubjects().getSubject().isEmpty()) {
List<SubjectProfile> subjectProfileBatch = request.getSubjects().getSubject().stream().map(subject ->
SubjectProfile.builder()
.identifier(subject.getId())
.phone(subject.getClientId())
.build()
).collect(Collectors.toList());
List<PhoneFidMap> subjectForSave = request.getSubjects().getSubject().stream().map(subject -> PhoneFidMap.builder()
.subjectId(subject.getId())
.packageId(packageId)
.phone(subject.getClientId())
.build()).toList();
List<Long> ids = phoneService2.registerSubjects(subjectForSave);
List<SubjectProfile> subjectProfiles = new ArrayList<>();
for (int i = 0; i < ids.size(); i++) {
PhoneFidMap phoneFidMap = subjectForSave.get(i);
subjectProfiles.add(SubjectProfile.builder()
.identifier(phoneFidMap.getSubjectId())
.phone(phoneFidMap.getPhone())
.mapId(ids.get(i))
.build());
}
int count = counter.incrementAndGet();
log.trace("Reading subject's {} batch. Total count: {}. PackageId: {}", count, request.getSubjects().getSubject().size(), packageId);
try {
firstQueue.put(new SubjectPayload(packageId,userPackage.getUserId(), userPackage.getPackageName(), subjectProfileBatch));
firstQueue.put(new SubjectPayload(packageId,userPackage.getUserId(), userPackage.getPackageName(), subjectProfiles));
} catch (InterruptedException e) {
isInterrupted = true;
log.error("Pipeline starter was interrupted!");

16
src/main/java/ru/nbch/credit_tracker/service/task/impl/Saver.java
View file

@ -13,7 +13,6 @@ import ru.nbch.credit_tracker.entities.app.PhoneFidMap;
import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest;
import ru.nbch.credit_tracker.enums.Status;
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
import ru.nbch.credit_tracker.model.SubjectProfile;
import ru.nbch.credit_tracker.service.indic.PhoneService2;
import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
@ -43,21 +42,14 @@ public class Saver {
xmlProcessor.read(in, MonitoringRequest.class,
request -> {
if (!request.getSubjects().getSubject().isEmpty()) {
List<SubjectProfile> subjects = request.getSubjects().getSubject().stream().map(subject ->
SubjectProfile.builder()
.identifier(subject.getId())
.phone(subject.getClientId())
.build()
).toList();
log.trace("Reading subject's {} batch. PackageId: {}", request.getSubjects().getSubject().size(), packageId);
List<PhoneFidMap> subjectForSave = subjects.stream().map(subject -> PhoneFidMap.builder()
.subjectId(subject.getIdentifier())
List<PhoneFidMap> subjectForSave = request.getSubjects().getSubject().stream().map(subject -> PhoneFidMap.builder()
.subjectId(subject.getId())
.packageId(packageId)
.phone(subject.getPhone())
.phone(subject.getClientId())
.build()).toList();
request.getSubjects().getSubject().clear();
phoneService2.registerSubjects(subjectForSave);
request.getSubjects().getSubject().clear();
}
},
request -> request.getSubjects().getSubject().isEmpty(),

4
src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilderV2.java
View file

@ -78,8 +78,8 @@ public class PipeLineBuilderV2<FIRST extends StagePayload, LAST extends StagePay
});
pipeLineStarter.startFillingQueue(firstStage.getIn());
});
pipeLineTerminator.setPollingQueue(lastQueue);
p.setTerminator(pipeLineTerminator);
pipeLineTerminator.setPollingQueue(lastQueue);
p.setTerminator(pipeLineTerminator);
p.setStages(stages);
return p;
}