Improve batch loading; Added SourceService

src/main/java/ru/nbch/credit_tracker/entities/nbch/Source.java

@@ -0,0 +1,13 @@
+package ru.nbch.credit_tracker.entities.nbch;
+
+import lombok.*;
+
+@Getter
+@Setter
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class Source {
+    private String memberCode;
+    private String category;
+}

src/main/java/ru/nbch/credit_tracker/mapper/nbch/SourceMapper.java

@@ -2,11 +2,15 @@ package ru.nbch.credit_tracker.mapper.nbch;
 
 import org.apache.ibatis.annotations.Mapper;
 import org.apache.ibatis.annotations.Param;
+import ru.nbch.credit_tracker.entities.nbch.Source;
 
 import java.util.List;
 
 @Mapper
 public interface SourceMapper {
 
-    List<String> findBusCategories(@Param("memberCodes") List<String> memberCodes);
+    List<Source> findBusCategories(@Param("memberCodes") List<String> memberCodes);
+
+    String findBusCategory(@Param("memberCode") String memberCode);
+
 }

src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/MonitoringService.java

@@ -1,5 +1,7 @@
 package ru.nbch.credit_tracker.service.mode.monitoring;
 
+import lombok.Getter;
+import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.MDC;
@@ -21,6 +23,7 @@ import ru.nbch.credit_tracker.entities.user_response.monitoring_signal_response.
 import ru.nbch.credit_tracker.enums.Status;
 import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
 import ru.nbch.credit_tracker.model.UserPackagePhoneData;
+import ru.nbch.credit_tracker.service.nbch.SourceService;
 import ru.nbch.credit_tracker.service.signal.HitsService;
 import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
 import ru.nbch.credit_tracker.service.signal.SignalRestService;
@@ -39,7 +42,6 @@ import java.time.LocalDateTime;
 import java.util.*;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.function.Function;
 import java.util.stream.Collectors;
 
 /**
@@ -54,42 +56,64 @@ public class MonitoringService {
     private final StaxXmlProcessor staxXmlProcessor;
     private final HitsService hitsService;
     private final PackageRecordsService packageRecordsService;
+    private final SourceService sourceService;
     private final CreditTrackerSettings config;
 
     public MonitoringService(SignalRestService signalRestService,
                              UserPackageMapper userPackageMapper,
                              StaxXmlProcessor staxXmlProcessor,
                              HitsService hitsService,
-                             PackageRecordsService packageRecordsService,
+                             PackageRecordsService packageRecordsService, SourceService sourceService,
                              CreditTrackerSettings config) {
         this.signalRestService = signalRestService;
         this.userPackageMapper = userPackageMapper;
         this.staxXmlProcessor = staxXmlProcessor;
         this.hitsService = hitsService;
         this.packageRecordsService = packageRecordsService;
+        this.sourceService = sourceService;
         this.config = config;
     }
 
     public void run() {
-        List<Integer> fids = new ArrayList<>();
+        List<MessageData> messages = new ArrayList<>();
 
         for (int i = 1000; i < 2000; i++) {
-            fids.add(i);
+            MessageData messageData = new MessageData();
+            messageData.setFid(i);
+            messageData.setMemberCode("ZX01LL");
+            messages.add(messageData);
         }
 
+        MessageData messageData1 = new MessageData();
+        messageData1.setFid(114885613);
+        messageData1.setMemberCode("ZX01LL");
+        messages.add(messageData1);
+
+        MessageData messageData2 = new MessageData();
+        messageData2.setFid(16445000);
+        messageData2.setMemberCode("ZX01LL");
+        messages.add(messageData2);
+
+        MessageData messageData3 = new MessageData();
+        messageData3.setFid(109837221);
+        messageData3.setMemberCode("ZX01LL");
+        messages.add(messageData3);
+
+
         long start = System.currentTimeMillis();
         log.info("Start selecting by one");
-        monitorByOne(fids);
+        monitorByOne(messages);
         log.info("Finish selecting by one. Time: {} ms", System.currentTimeMillis() - start);
 
         start = System.currentTimeMillis();
         log.info("Start selecting by batch");
-        monitorbyBatch(fids);
+        monitorbyBatch(messages);
         log.info("Finish selecting by bacth. Time: {} ms", System.currentTimeMillis() - start);
     }
 
-    private void monitorByOne(List<Integer> fids) {
+    private void monitorByOne(List<MessageData> messages) {
-        for (Integer fid : fids) {
+        for (MessageData message : messages) {
+            Integer fid = message.getFid();
             List<PhonesMap> phonesMaps = packageRecordsService.findCachedByFid(fid);
             for (PhonesMap phonesMap : phonesMaps) {
                 if (phonesMap != null) {
@@ -104,6 +128,7 @@ public class MonitoringService {
                         String status = userPackage.getStatus();
                         String signalTypes = userPackage.getSignalTypes();
 
+                        String busCategory = sourceService.findBusCategory(message.getMemberCode());
                         // проверить статус ON_MPNITORING + разрешенные signalTypes
                     }
                 }
@@ -111,32 +136,71 @@ public class MonitoringService {
         }
     }
 
-    private void monitorbyBatch(List<Integer> fids) {
+    private void monitorbyBatch(List<MessageData> messages) {
+        List<Integer> fids = messages.stream().map(MessageData::getFid).collect(Collectors.toList());
+        List<String> memberCodes = messages.stream()
+                .map(message -> message.getMemberCode().substring(0,6))
+                .toList();
+
         List<PhonesMap> phonesMaps = packageRecordsService.findCachedByFids(fids);
-        // key: phone; value: флаги просрочек
+
-        Map<Long, PhonesMap> phonesDataMap = phonesMaps.stream()
+        // key: fid; value: список флагов просрочек (у одного фида может быть несколько номеров)
-                .collect(
+        Map<Integer, List<PhonesMap>> phonesDataMap = phonesMaps.stream()
-                        Collectors.toMap(PhonesMap::getPhone, Function.identity())
+                .collect(Collectors.groupingBy(PhonesMap::getFid));
-                );
+
+        Set<Long> uniquePhones = phonesMaps.stream().map(PhonesMap::getPhone).collect(Collectors.toSet());
+
+        // прелоад по memberCode категорий
+        Map<String, String> busCategories = sourceService.findBusCategories(memberCodes);
 
         // subjectId, packageId, signalTypes, phone только для статуса ON_MONITORING
-        List<UserPackagePhoneData> userPackagePhoneDataList = packageRecordsService
+        List<UserPackagePhoneData> packagePhoneData = packageRecordsService
-                .selectUserPackagePhoneData(phonesDataMap.keySet());
+                .selectUserPackagePhoneData(uniquePhones);
 
-        // остались только нужные номера, те что не вошли в userPackagePhoneDataList игнорируются
+        // 1 номер может быть у нескольких пакетов
-        for (UserPackagePhoneData userPackagePhoneData : userPackagePhoneDataList) {
+        Map<Long, List<UserPackagePhoneData>> userPackagePhoneDataMap = packagePhoneData.stream()
-            Long clientId = userPackagePhoneData.getPhone();
+                .collect(Collectors.groupingBy(UserPackagePhoneData::getPhone));
-            String subjectId = userPackagePhoneData.getSubjectId();
-            String signalTypes = userPackagePhoneData.getSignalTypes();
 
-            PhonesMap phonesMap = phonesDataMap.get(clientId);
+        for (MessageData message : messages) {
+            Integer fid = message.getFid();
+            String memberCode = message.getMemberCode();
+
+            String busCategory = busCategories.get(memberCode.substring(0, 6));
+            List<PhonesMap> phonesMapList = phonesDataMap.get(fid);
+
+            if (phonesMapList == null) {
+                continue;
+            }
+
+            for (PhonesMap phonesMap : phonesMapList) {
+                // флаги просрочек
                 Integer flagDebt5000 = phonesMap.getFlagDebt5000();
                 Integer flag90days = phonesMap.getFlag90days();
 
+                // в userPackagePhoneDataMap только нужные номера, те что не вошли в userPackagePhoneDataList игнорируются
+                List<UserPackagePhoneData> userPackagePhoneDataList = userPackagePhoneDataMap.get(phonesMap.getPhone());
+
+                if (userPackagePhoneDataList == null) {
+                    continue;
+                }
+
+                for (UserPackagePhoneData userPackagePhoneData : userPackagePhoneDataList) {
+                    String subjectId = userPackagePhoneData.getSubjectId();
+                    String signalTypes = userPackagePhoneData.getSignalTypes();
+
                     // проверить разрешенные signalTypes
                 }
             }
+        }
+    }
 
+    @Getter
+    @Setter
+    public static class MessageData {
+        private Integer fid;
+        private String memberCode;
+        // other data
+    }
 
     private void monitor() {
         try {

src/main/java/ru/nbch/credit_tracker/service/nbch/SourceService.java

@@ -0,0 +1,35 @@
+package ru.nbch.credit_tracker.service.nbch;
+
+import org.springframework.stereotype.Service;
+import ru.nbch.credit_tracker.entities.nbch.Source;
+import ru.nbch.credit_tracker.mapper.nbch.SourceMapper;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+@Service
+public class SourceService {
+
+    private final SourceMapper sourceMapper;
+
+    public SourceService(SourceMapper sourceMapper) {
+        this.sourceMapper = sourceMapper;
+    }
+
+    public Map<String, String> findBusCategories(List<String> memberCodes){
+        if (memberCodes.isEmpty()) {
+            return Collections.emptyMap();
+        }
+        return sourceMapper.findBusCategories(memberCodes).stream()
+                .collect(Collectors.toMap(
+                        Source::getMemberCode,
+                        Source::getCategory
+                ));
+    }
+
+    public String findBusCategory(String memberCode) {
+        return sourceMapper.findBusCategory(memberCode.substring(0, 6));
+    }
+}

src/main/resources/mybatis/mapper/nbch/SourceMapper.xml

@@ -5,12 +5,22 @@
 
 <mapper namespace="ru.nbch.credit_tracker.mapper.nbch.SourceMapper">
 
-    <select id="findBusCategories" resultType="java.lang.String">
+    <select id="findBusCategories" resultType="ru.nbch.credit_tracker.entities.nbch.Source">
-        SELECT CATEGORY
+        SELECT DISTINCT
+        MEMBER_CODE as memberCode,
+        CATEGORY as category
         FROM SOURCE
         WHERE MEMBER_CODE IN
         <foreach item='memberCode' collection='memberCodes' open='(' separator=',' close=')'>
             #{memberCode}
         </foreach>
     </select>
+
+    <select id="findBusCategory" resultType="java.lang.String">
+        SELECT CATEGORY
+        FROM SOURCE
+        WHERE MEMBER_CODE = #{memberCode}
+        LIMIT 1
+    </select>
+
 </mapper>