Add: 610 召回过滤前的视频数量

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/framework/recaller/BaseRecaller.java

@@ -21,6 +21,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
 
 
@@ -149,9 +150,11 @@ public class BaseRecaller<Video> {
      * @param requestData
      * @param user
      * @param recallCandidates
+     * @param timeLogMap
      * @return
      */
-    public List<RankItem> recalling(final RecommendRequest requestData, final User user, List<Candidate> recallCandidates) {
+    public List<RankItem> recalling(final RecommendRequest requestData, final User user, List<Candidate> recallCandidates,
+                                    Map<String, String> timeLogMap) {
         Stopwatch stopwatch = Stopwatch.createStarted();
         stopwatch.reset().start();
         // load from redis
@@ -192,7 +195,7 @@ public class BaseRecaller<Video> {
         }
 
         stopwatch.reset().start();
-        List<RankItem> result = convertToRankItem(candidateQueueMap, requestData, user);
+        List<RankItem> result = convertToRankItem(candidateQueueMap, requestData, user, timeLogMap);
 
         return result;
     }
@@ -202,12 +205,11 @@ public class BaseRecaller<Video> {
     // 同时给Filter预留处理
     private List<RankItem> convertToRankItem(Map<Candidate, Queue<Video>> candidateQueueMap,
                                              RecommendRequest requestData,
-                                             User user) {
-
-
+                                             User user, Map<String, String> timeLogMap) {
         final FilterPipeline<Video> recallFilter = new FilterPipeline<>(filterConfig, requestData, user);
 
         final List<Callable<List<RankItem>>> callables = new ArrayList<>();
+        AtomicLong videoCountBeforeFilter = new AtomicLong(0);
         for (final Map.Entry<Candidate, Queue<Video>> entry : candidateQueueMap.entrySet()) {
             callables.add(() -> {
                 List<RankItem> candidateHits = new ArrayList<>();
@@ -219,6 +221,7 @@ public class BaseRecaller<Video> {
                         entryMap.put(e.item, e);
                         return e.item;
                     }).collect(Collectors.toList());
+                    videoCountBeforeFilter.addAndGet(entriesList.size());
                     List<Video> entriesAfterFilter = recallFilter.doFilter(candidate, entriesList);
                     // 1. filter  TODO 待后续增加自定义filter
                     if (CollectionUtils.isNotEmpty(entriesAfterFilter)) {
@@ -265,6 +268,8 @@ public class BaseRecaller<Video> {
         } catch (Exception e) {
             LOGGER.error("parallel recall filter occur error, uid: [{}], Exception [{}]",
                     "", ExceptionUtils.getFullStackTrace(e));
+        } finally {
+            timeLogMap.put("recall-count-before-filter", String.valueOf(videoCountBeforeFilter.get()));
         }
 
         List<RankItem> result = new ArrayList<>(hits.values());

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/implement/TopRecommendPipeline.java

@@ -149,7 +149,7 @@ public class TopRecommendPipeline {
         // Step 4: Recalling & Basic Scoring
         stopwatch.reset().start();
         BaseRecaller recaller = new BaseRecaller(queueProvider);
-        List<RankItem> items = recaller.recalling(requestData, userInfo, new ArrayList<>(candidates.values()));
+        List<RankItem> items = recaller.recalling(requestData, userInfo, new ArrayList<>(candidates.values()), timeLogMap);
         if (logPrint) {
             log.info("traceId = {}, cost = {}, items = {}", requestData.getRequestId(),
                     stopwatch.elapsed().toMillis(), JSONUtils.toJson(items));