some refactoring :]
This commit is contained in:
parent
f2df1c0183
commit
9cb5a16490
2 changed files with 5 additions and 2 deletions
|
|
@ -28,21 +28,23 @@ public class PipelinePackageManager {
|
|||
private final StaxXmlProcessor xmlProcessor;
|
||||
private final StageQueueFactory<SubjectPayload> saveSubjectQueueFactory;
|
||||
private final StageQueueFactory<SubjectPayload> idEnricherQueueFactory;
|
||||
private final SaveSubjectHandler saveSubjectHandler;
|
||||
private final ExecutorService executorService = Executors.newFixedThreadPool(10);
|
||||
private final static int BATCH_SIZE = 10_000;
|
||||
|
||||
public PipelinePackageManager(ApplicationContext applicationContext,
|
||||
StaxXmlProcessor xmlProcessor,
|
||||
@Qualifier("saveSubjectQueueFactory") StageQueueFactory<SubjectPayload> saveSubjectQueueFactory,
|
||||
@Qualifier("idEnricherQueueFactory") StageQueueFactory<SubjectPayload> idEnricherQueueFactory) {
|
||||
@Qualifier("idEnricherQueueFactory") StageQueueFactory<SubjectPayload> idEnricherQueueFactory,
|
||||
SaveSubjectHandler saveSubjectHandler) {
|
||||
this.applicationContext = applicationContext;
|
||||
this.xmlProcessor = xmlProcessor;
|
||||
this.saveSubjectQueueFactory = saveSubjectQueueFactory;
|
||||
this.idEnricherQueueFactory = idEnricherQueueFactory;
|
||||
this.saveSubjectHandler = saveSubjectHandler;
|
||||
}
|
||||
|
||||
public void buildAndRun(Long packageId, String pkgPath) throws Exception {
|
||||
SaveSubjectHandler saveSubjectHandler = applicationContext.getBean(SaveSubjectHandler.class);
|
||||
BlockingQueue<SubjectPayload> saveSubjectQueue = saveSubjectQueueFactory.getObject();
|
||||
BlockingQueue<SubjectPayload> idEnricherQueue = idEnricherQueueFactory.getObject();
|
||||
|
||||
|
|
|
|||
|
|
@ -42,6 +42,7 @@ public class SaveSubjectHandler implements Handler<SubjectPayload, SubjectPayloa
|
|||
phoneFidMapService.register(subjects, nameByPhone, packageId);
|
||||
|
||||
//fill name info
|
||||
//todo сделать один проход для установки FID для вставки в PHONE_FID_MAP и одновременного заполнения инфо из Name для дальнейшей обработки
|
||||
subjects.stream()
|
||||
.filter(subject -> nameByPhone.containsKey(CTUtil.normalizePhoneNumber(subject.getPhone())))
|
||||
.forEach(subject -> {
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue