Merge branch 'feature_20240913_zb_vov' of algorithm/recommend-server into feature_20240910_zhaohaipeng_vov_filter

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/common/ThreadPoolFactory.java

@@ -12,8 +12,8 @@ import java.util.concurrent.TimeUnit;
  */
 public final class ThreadPoolFactory {
     private final static ExecutorService DEFAULT = new CommonThreadPoolExecutor(
-            32,
-            32,
+            64,
+            64,
             0L, TimeUnit.SECONDS,
             new LinkedBlockingQueue<>(1000),
             new ThreadFactoryBuilder().setNameFormat("DEFAULT-%d").build(),

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/filter/AbstractFilterService.java

@@ -128,7 +128,7 @@ public abstract class AbstractFilterService {
         }
 
         // VOV过滤实验
-        if (param.getAbExpCodes().contains("689")) {
+        if (param.getAbExpCodes().contains("564")) {
             strategies.add(ServiceBeanFactory.getBean(VovLowerStrategy.class));
         }
 

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/rank/RankRouter.java

@@ -82,7 +82,7 @@ public class RankRouter {
             case "60113": // 563
                 return rankStrategy4RegionMergeModelV563.rank(param);
             case "60114": // 564
-                return rankStrategy4RegionMergeModelV564.rank(param);
+                return rankStrategy4RegionMergeModelV552.rank(param);
             case "60115": // 565
                 return rankStrategy4RegionMergeModelV565.rank(param);
             case "60116": // 566

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/rank/strategy/RankStrategy4RegionMergeModelV552.java

@@ -1,6 +1,7 @@
 package com.tzld.piaoquan.recommend.server.service.rank.strategy;
 
 import com.ctrip.framework.apollo.spring.annotation.ApolloJsonValue;
+import com.tzld.piaoquan.recommend.server.common.ThreadPoolFactory;
 import com.tzld.piaoquan.recommend.server.common.base.RankItem;
 import com.tzld.piaoquan.recommend.server.model.Video;
 import com.tzld.piaoquan.recommend.server.service.FeatureService;
@@ -11,7 +12,9 @@ import com.tzld.piaoquan.recommend.server.service.score.ScorerUtils;
 import com.tzld.piaoquan.recommend.server.util.CommonCollectionUtils;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections4.MapUtils;
+import org.apache.commons.math3.util.Pair;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 
 import java.io.BufferedReader;
@@ -19,6 +22,9 @@ import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
 import java.util.*;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 @Service
@@ -33,6 +39,9 @@ public class RankStrategy4RegionMergeModelV552 extends RankStrategy4RegionMergeM
     Map<String, double[]> bucketsMap = new HashMap<>();
     Map<String, Double> bucketsLen = new HashMap<>();
 
+    @Value("${similarity.concurrent: false}")
+    private boolean similarityConcurrent;
+
     @Override
     public List<Video> mergeAndRankRovRecall(RankParam param) {
         Map<String, Double> mergeWeight = this.mergeWeight != null ? this.mergeWeight : new HashMap<>(0);
@@ -221,14 +230,41 @@ public class RankStrategy4RegionMergeModelV552 extends RankStrategy4RegionMergeM
 
             String title = videoInfo.getOrDefault("title", "");
             if (!title.isEmpty()) {
-                for (String name : Arrays.asList("c3_feature", "c4_feature", "c5_feature", "c6_feature", "c7_feature")) {
-                    for (String key_time : Arrays.asList("tags_1d", "tags_3d", "tags_7d")) {
-                        String tags = c34567Map.getOrDefault(name + "_" + key_time, "");
-                        if (!tags.isEmpty()) {
-                            Double[] doubles = ExtractorUtils.funcC34567ForTags(tags, title);
-                            featureMap.put(name + "_" + key_time + "_matchnum", doubles[0]);
-                            featureMap.put(name + "_" + key_time + "_maxscore", doubles[1]);
-                            featureMap.put(name + "_" + key_time + "_avgscore", doubles[2]);
+                if (similarityConcurrent) {
+                    List<Future<Pair<String, Double[]>>> futures = new ArrayList<>();
+                    for (String name : Arrays.asList("c3_feature", "c4_feature", "c5_feature", "c6_feature", "c7_feature")) {
+                        for (String key_time : Arrays.asList("tags_1d", "tags_3d", "tags_7d")) {
+                            String key = name + "_" + key_time;
+                            String tags = c34567Map.getOrDefault(key, "");
+                            if (!tags.isEmpty()) {
+                                Future<Pair<String, Double[]>> future = ThreadPoolFactory.defaultPool().submit(() -> {
+                                    Double[] doubles = ExtractorUtils.funcC34567ForTags(tags, title);
+                                    return Pair.create(key, doubles);
+                                });
+                                futures.add(future);
+                            }
+                        }
+                    }
+                    try {
+                        for (Future<Pair<String, Double[]>> future : futures) {
+                            Pair<String, Double[]> pair = future.get(1000, TimeUnit.MILLISECONDS);
+                            featureMap.put(pair.getFirst() + "_matchnum", pair.getSecond()[0]);
+                            featureMap.put(pair.getFirst() + "_maxscore", pair.getSecond()[1]);
+                            featureMap.put(pair.getFirst() + "_avgscore", pair.getSecond()[2]);
+                        }
+                    } catch (Exception e) {
+                        log.error("concurrent similarity error", e);
+                    }
+                } else {
+                    for (String name : Arrays.asList("c3_feature", "c4_feature", "c5_feature", "c6_feature", "c7_feature")) {
+                        for (String key_time : Arrays.asList("tags_1d", "tags_3d", "tags_7d")) {
+                            String tags = c34567Map.getOrDefault(name + "_" + key_time, "");
+                            if (!tags.isEmpty()) {
+                                Double[] doubles = ExtractorUtils.funcC34567ForTags(tags, title);
+                                featureMap.put(name + "_" + key_time + "_matchnum", doubles[0]);
+                                featureMap.put(name + "_" + key_time + "_maxscore", doubles[1]);
+                                featureMap.put(name + "_" + key_time + "_avgscore", doubles[2]);
+                            }
                         }
                     }
                 }

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/rank/strategy/RankStrategy4RegionMergeModelV562.java

@@ -1,6 +1,7 @@
 package com.tzld.piaoquan.recommend.server.service.rank.strategy;
 
 import com.ctrip.framework.apollo.spring.annotation.ApolloJsonValue;
+import com.tzld.piaoquan.recommend.server.common.ThreadPoolFactory;
 import com.tzld.piaoquan.recommend.server.common.base.RankItem;
 import com.tzld.piaoquan.recommend.server.model.Video;
 import com.tzld.piaoquan.recommend.server.service.FeatureService;
@@ -11,7 +12,9 @@ import com.tzld.piaoquan.recommend.server.service.score.ScorerUtils;
 import com.tzld.piaoquan.recommend.server.util.CommonCollectionUtils;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections4.MapUtils;
+import org.apache.commons.math3.util.Pair;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 
 import java.io.BufferedReader;
@@ -19,6 +22,8 @@ import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
 import java.util.*;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 @Service
@@ -33,6 +38,9 @@ public class RankStrategy4RegionMergeModelV562 extends RankStrategy4RegionMergeM
     Map<String, double[]> bucketsMap = new HashMap<>();
     Map<String, Double> bucketsLen = new HashMap<>();
 
+    @Value("${similarity.concurrent: false}")
+    private boolean similarityConcurrent;
+
     @Override
     public List<Video> mergeAndRankRovRecall(RankParam param) {
         Map<String, Double> mergeWeight = this.mergeWeight != null ? this.mergeWeight : new HashMap<>(0);
@@ -221,14 +229,41 @@ public class RankStrategy4RegionMergeModelV562 extends RankStrategy4RegionMergeM
 
             String title = videoInfo.getOrDefault("title", "");
             if (!title.isEmpty()) {
-                for (String name : Arrays.asList("c3_feature", "c4_feature", "c5_feature", "c6_feature", "c7_feature")) {
-                    for (String key_time : Arrays.asList("tags_1d", "tags_3d", "tags_7d")) {
-                        String tags = c34567Map.getOrDefault(name + "_" + key_time, "");
-                        if (!tags.isEmpty()) {
-                            Double[] doubles = ExtractorUtils.funcC34567ForTags(tags, title);
-                            featureMap.put(name + "_" + key_time + "_matchnum", doubles[0]);
-                            featureMap.put(name + "_" + key_time + "_maxscore", doubles[1]);
-                            featureMap.put(name + "_" + key_time + "_avgscore", doubles[2]);
+                if (similarityConcurrent) {
+                    List<Future<Pair<String, Double[]>>> futures = new ArrayList<>();
+                    for (String name : Arrays.asList("c3_feature", "c4_feature", "c5_feature", "c6_feature", "c7_feature")) {
+                        for (String key_time : Arrays.asList("tags_1d", "tags_3d", "tags_7d")) {
+                            String key = name + "_" + key_time;
+                            String tags = c34567Map.getOrDefault(key, "");
+                            if (!tags.isEmpty()) {
+                                Future<Pair<String, Double[]>> future = ThreadPoolFactory.defaultPool().submit(() -> {
+                                    Double[] doubles = ExtractorUtils.funcC34567ForTags(tags, title);
+                                    return Pair.create(key, doubles);
+                                });
+                                futures.add(future);
+                            }
+                        }
+                    }
+                    try {
+                        for (Future<Pair<String, Double[]>> future : futures) {
+                            Pair<String, Double[]> pair = future.get(1000, TimeUnit.MILLISECONDS);
+                            featureMap.put(pair.getFirst() + "_matchnum", pair.getSecond()[0]);
+                            featureMap.put(pair.getFirst() + "_maxscore", pair.getSecond()[1]);
+                            featureMap.put(pair.getFirst() + "_avgscore", pair.getSecond()[2]);
+                        }
+                    } catch (Exception e) {
+                        log.error("concurrent similarity error", e);
+                    }
+                } else {
+                    for (String name : Arrays.asList("c3_feature", "c4_feature", "c5_feature", "c6_feature", "c7_feature")) {
+                        for (String key_time : Arrays.asList("tags_1d", "tags_3d", "tags_7d")) {
+                            String tags = c34567Map.getOrDefault(name + "_" + key_time, "");
+                            if (!tags.isEmpty()) {
+                                Double[] doubles = ExtractorUtils.funcC34567ForTags(tags, title);
+                                featureMap.put(name + "_" + key_time + "_matchnum", doubles[0]);
+                                featureMap.put(name + "_" + key_time + "_maxscore", doubles[1]);
+                                featureMap.put(name + "_" + key_time + "_avgscore", doubles[2]);
+                            }
                         }
                     }
                 }