From 90b8d8ac4e4afedd50521efaac9d4ed46b1eba78 Mon Sep 17 00:00:00 2001 From: ialbert Date: Thu, 22 Jan 2026 13:18:21 +0300 Subject: [PATCH] flags new scheduller --- credit-tracker.properties | 1 + .../mapper/signal/PhonesMapMapper.java | 2 + .../service/UpdatingFlagsService.java | 72 +++++++++++++++---- .../mybatis/mapper/signal/PhonesMapMapper.xml | 12 +++- 4 files changed, 72 insertions(+), 15 deletions(-) diff --git a/credit-tracker.properties b/credit-tracker.properties index 805c183..fb3ee0c 100644 --- a/credit-tracker.properties +++ b/credit-tracker.properties @@ -67,6 +67,7 @@ ct.load-dir=D:\\folder_name\\signals ct.monitoring-interval=${MONITORING_INTERVAL:15} # days ct.update-flags-cron-expression=${UPDATE_FLAGS_CRON_EXPRESSION:0 25 18 * * *} +ct.new-flags-cron-expression=${UPDATE_FLAGS_CRON_EXPRESSION:0 25 18 * * *} ct.days-to-update-flags=${DAYS_TO_UPDATE_FLAGS:30} ct.update-fid-cron-expression=${UPDATE_FLAGS_CRON_EXPRESSION:0 10 12 * * *} ct.days-to-update-fids=${DAYS_TO_UPDATE_FIDS:30} diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java index 0829ec8..6c2a9c3 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java @@ -12,6 +12,8 @@ public interface PhonesMapMapper { PhonesMap findByPhone(@Param("phone") Long phone); List findEligibleForFlagsUpdate(@Param("lastId") Long lastId, @Param("batchSize") Integer batchSize); + List findEligibleForFlagsInsert(@Param("lastId") Long lastId, + @Param("batchSize") Integer batchSize); void updateIdCachedData(@Param("json") String json); void updateNameCachedData(@Param("json") String json); void updateFlagCachedData(@Param("json") String json); diff --git a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java index 0615c86..1b325fc 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java @@ -1,7 +1,10 @@ package ru.nbch.credit_tracker.service; import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; import lombok.extern.slf4j.Slf4j; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; @@ -17,9 +20,12 @@ import ru.nbch.credit_tracker.service.task.impl.FlagsEnricher; @Service @EnableScheduling public class UpdatingFlagsService { + private final Logger log = LoggerFactory.getLogger(this.getClass()); private final static int BATCH_SIZE = 1000; private final PhonesMapMapper phonesMapMapper; private final FlagsEnricher flagsEnricherV2; + private final AtomicBoolean insertIsRunning = new AtomicBoolean(false); + private final AtomicBoolean updateIsRunning = new AtomicBoolean(false); public UpdatingFlagsService(PhonesMapMapper phonesMapMapper, FlagsEnricher flagsEnricherV2) { this.phonesMapMapper = phonesMapMapper; @@ -27,21 +33,59 @@ public class UpdatingFlagsService { } @Scheduled(cron = "${ct.update-flags-cron-expression}") - private void update() { - log.info("Start filling flags"); - long lastId = 0L; - int batchNumber = 1; - List phonesMaps; - while (true) { - phonesMaps = phonesMapMapper.findEligibleForFlagsUpdate(lastId, BATCH_SIZE); - log.trace("batch {}, signals_phones_map records loaded: {}", batchNumber, phonesMaps.size()); - if (phonesMaps.isEmpty()) { - break; - } + private void updateFlags() { + if (!updateIsRunning.compareAndSet(false, true)) { + log.error("update flags processor already running... skipping scheduled update"); + return; + } + try { + log.info("start update flags processor"); + long lastId = 0L; + int batchNumber = 1; + List phonesMaps; + while (true) { + phonesMaps = phonesMapMapper.findEligibleForFlagsUpdate(lastId, BATCH_SIZE); + log.trace("update batch {}, signals_phones_map records loaded: {}", batchNumber, phonesMaps.size()); + if (phonesMaps.isEmpty()) { + break; + } - flagsEnricherV2.accept(phonesMaps); - lastId = phonesMaps.getLast().getId(); - log.trace("batch {}, lastId {}", batchNumber++, lastId); + flagsEnricherV2.accept(phonesMaps); + lastId = phonesMaps.getLast().getId(); + log.trace("update batch {}, lastId {}", batchNumber++, lastId); + } + log.info("end update flags processor"); + } finally { + updateIsRunning.set(false); } } + + @Scheduled(cron = "${ct.new-flags-cron-expression}") + private void newFlags() { + if (!insertIsRunning.compareAndSet(false, true)) { + log.warn("new flags processor already running... skipping scheduled insert"); + return; + } + try { + log.info("start processing new flags"); + long lastId = 0L; + int batchNumber = 1; + List phonesMaps; + while (true) { + phonesMaps = phonesMapMapper.findEligibleForFlagsInsert(lastId, BATCH_SIZE); + log.trace("new batch {}, signals_phones_map records loaded: {}", batchNumber, phonesMaps.size()); + if (phonesMaps.isEmpty()) { + break; + } + + flagsEnricherV2.accept(phonesMaps); + lastId = phonesMaps.getLast().getId(); + log.trace("new batch {}, lastId {}", batchNumber++, lastId); + } + log.info("end new flags processor"); + } finally { + insertIsRunning.set(false); + } + + } } diff --git a/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml b/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml index 44d1e87..e880b70 100644 --- a/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml @@ -120,7 +120,17 @@ select id, fid, phone from signals_phones_map where fid is not null - and (calculated_at is null or calculated_at < searched_at) + and (calculated_at is not null and calculated_at < searched_at) + and id > #{lastId} + order by id + limit #{batchSize} + + +