Merge branch 'feature_flowpool' of algorithm/recommend-server into master

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

@@ -18,7 +18,6 @@ import com.tzld.piaoquan.recommend.server.service.recall.RecallResult;
 import com.tzld.piaoquan.recommend.server.service.recall.RecallStrategy;
 import com.tzld.piaoquan.recommend.server.service.recall.strategy.FlowPoolWithLevelRecallStrategyTomson;
 import com.tzld.piaoquan.recommend.server.service.recall.strategy.FlowPoolWithLevelScoreRecallStrategy;
-import com.tzld.piaoquan.recommend.server.service.recall.strategy.FlowPoolWithScoreRecallStrategy;
 import com.tzld.piaoquan.recommend.server.service.score.ScoreParam;
 import com.tzld.piaoquan.recommend.server.service.score.ScorerUtils;
 import com.tzld.piaoquan.recommend.server.util.CommonCollectionUtils;
@@ -169,8 +168,6 @@ public class FlowPoolRecommendPipeline implements ApplicationContextAware {
             strategies.add(strategyMap.get(FlowPoolWithLevelRecallStrategyTomson.class.getSimpleName()));
         } else if (param.getFlowPoolAbtestGroup().equals(FlowPoolConstants.EXPERIMENTAL_FLOW_SET_LEVEL_SCORE)) {
             strategies.add(strategyMap.get(FlowPoolWithLevelScoreRecallStrategy.class.getSimpleName()));
-        } else {
-            strategies.add(strategyMap.get(FlowPoolWithScoreRecallStrategy.class.getSimpleName()));
         }
 
         CountDownLatch cdl = new CountDownLatch(strategies.size());

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

@@ -398,7 +398,6 @@ public class RecommendService {
         // 流量池分发实验组划分
         int flowPoolIdChoice = flowPoolIds.get(RandomUtils.nextInt(0, flowPoolIds.size()));
         Map<String, List<Integer>> flowPoolConfig = flowPoolConfigService.getFlowPoolConfig();
-        Set<String> abExpCodes = param.getAbExpCodes();
 
         param.setFlowPoolId(flowPoolIdChoice);
         param.setFlowPoolAbtestGroup("control_group");
@@ -482,14 +481,14 @@ public class RecommendService {
 
         long recallTime = stopwatch.elapsed(TimeUnit.MILLISECONDS);
         timerLogMapTL.get().put("recallTime", recallTime);
-        log.info("recallResult={}, videoRecommend recallResult cost={}", recallResult, recallTime);
+        //log.info("recallResult={}, videoRecommend recallResult cost={}", recallResult, recallTime);
         stopwatch.reset().start();
 
         RankResult rankResult = rankRouter.rank(convertToRankParam(param, recallResult));
 
         long rankTime = stopwatch.elapsed(TimeUnit.MILLISECONDS);
         timerLogMapTL.get().put("rankTime", rankTime);
-        log.info("rankResult={}, videoRecommend rank cost={}", rankResult, rankTime);
+        //log.info("rankResult={}, videoRecommend rank cost={}", rankResult, rankTime);
 
 
         if (rankResult == null || CollectionUtils.isEmpty(rankResult.getVideos())) {
@@ -648,9 +647,6 @@ public class RecommendService {
             case FlowPoolConstants.EXPERIMENTAL_FLOW_SET_LEVEL_SCORE:
                 flowPoolService.updateDistributeCountWithLevelScore(flowPoolVideos);
                 break;
-            default:
-                flowPoolService.updateDistributeCountWithScore(flowPoolVideos);
-                break;
         }
     }
 

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

@@ -43,15 +43,7 @@ public class FlowPoolWithLevelFilterService extends AbstractFilterService {
                         v -> param.getFlowPoolMap().get(v)));
 
         Map<Long, Integer> distributeCountMap;
-        //供给流量池实验
-        Set<String> abExpCodes = param.getAbExpCodes();
-        if (Objects.nonNull(abExpCodes) && abExpCodes.contains(Constant.SUPPLY_AB_CODE_ID)) {
-//        if (Objects.equals(Constant.SUPPLY_AB_CODE, param.getAbCode())) {
-            distributeCountMap = flowPoolService.getSupplyDistributeCountWithLevel(flowPoolMap);
-        } else {
-            distributeCountMap = flowPoolService.getDistributeCountWithLevel(flowPoolMap);
-        }
-//        Map<Long, Integer> distributeCountMap = flowPoolService.getDistributeCountWithLevel(flowPoolMap);
+        distributeCountMap = flowPoolService.getDistributeCountWithLevel(flowPoolMap);
 
         List<Long> remainVideoIds = new ArrayList<>();
         for (Long videoId : videoIds) {

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

@@ -1,54 +0,0 @@
-package com.tzld.piaoquan.recommend.server.service.filter;
-
-import com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolService;
-import com.tzld.piaoquan.recommend.server.util.JSONUtils;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.collections4.CollectionUtils;
-import org.apache.commons.collections4.MapUtils;
-import org.apache.commons.lang3.StringUtils;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.stream.Collectors;
-
-/**
- * @author dyp
- */
-@Service
-public class FlowPoolWithScoreFilterService extends AbstractFilterService {
-    @Autowired
-    private FlowPoolService flowPoolService;
-
-    public FilterResult filter(FilterParam param) {
-        List<Long> videoIds = viewFilter(param);
-
-        videoIds = filterDistributeCountAndDel(param, videoIds);
-        return new FilterResult(videoIds);
-    }
-
-    private List<Long> filterDistributeCountAndDel(FilterParam param, List<Long> videoIds) {
-        if (CollectionUtils.isEmpty(videoIds)
-                || MapUtils.isEmpty(param.getFlowPoolMap())) {
-            return videoIds;
-        }
-        Map<Long, String> flowPoolMap = videoIds.stream()
-                .filter(v -> param.getFlowPoolMap().containsKey(v)
-                        && StringUtils.isNotBlank(param.getFlowPoolMap().get(v)))
-                .collect(Collectors.toMap(
-                        v -> v,
-                        v -> param.getFlowPoolMap().get(v)));
-
-        Map<Long, Integer> distributeCountMap = flowPoolService.getDistributeCountWithScore(flowPoolMap);
-        List<Long> remainVideoIds = new ArrayList<>();
-        for (Long videoId : videoIds) {
-            if (distributeCountMap.get(videoId) != null
-                    && distributeCountMap.get(videoId) > 0) {
-                remainVideoIds.add(videoId);
-            }
-        }
-        return remainVideoIds;
-    }
-}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/flowpool/FlowPoolConfigService.java

@@ -9,6 +9,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
 
@@ -28,6 +29,9 @@ public class FlowPoolConfigService {
     @Qualifier("redisTemplate")
     private RedisTemplate<String, String> redisTemplate;
 
+    @Value("${flow.pool.upgrade.switch:true}")
+    private boolean flowPoolUpgradeSwitch;
+
     // TODO 这些配置可以用配置中心么？
     // {"control_group": [], "experimental_flow_set_level": [10, 11, 12, 13, 14, 15, 16, 17], "experimental_flow_set_level_score": [9, 18]}
     private LoadingCache<String, Map<String, List<Integer>>> flowPoolConfigCache = CacheBuilder.newBuilder()
@@ -59,7 +63,7 @@ public class FlowPoolConfigService {
                 public Map<String, Double> load(String key) throws Exception {
                     String value = redisTemplate.opsForValue().get(key);
                     if (StringUtils.isEmpty(value)) {
-                        value = "level_weight = {\"1\": 1, \"2\": 1, \"3\": 1, \"4\": 1, \"5\": 1, \"6\": 1}";
+                        value = "{\"1\": 1, \"2\": 1, \"3\": 1, \"4\": 1}";
                     }
                     return JSONUtils.fromJson(value, new TypeToken<Map<String, Double>>() {
                     }, Collections.emptyMap());
@@ -75,9 +79,22 @@ public class FlowPoolConfigService {
         }
     }
 
-    public Map<String, Double> getLevelWeight() {
+    public Map<String, Double> getLevelWeight4FlowPoolWithLevel() {
+        try {
+            return flowPoolUpgradeSwitch
+                    ? levelWeightConfigCache.get("flow:pool:level:weight:v2")
+                    : levelWeightConfigCache.get("flow:pool:level:recommend:weight");
+        } catch (ExecutionException e) {
+            log.error("getFlowPoolConfig error", e);
+            return Collections.emptyMap();
+        }
+    }
+
+    public Map<String, Double> getLevelWeight4FlowPoolWithLevelScore() {
         try {
-            return levelWeightConfigCache.get("flow:pool:level:recommend:weight");
+            return flowPoolUpgradeSwitch
+                    ? levelWeightConfigCache.get("flow:pool:level:score:weight:v2")
+                    : levelWeightConfigCache.get("flow:pool:level:recommend:weight");
         } catch (ExecutionException e) {
             log.error("getFlowPoolConfig error", e);
             return Collections.emptyMap();

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/flowpool/FlowPoolConstants.java

@@ -6,7 +6,6 @@ package com.tzld.piaoquan.recommend.server.service.flowpool;
 public class FlowPoolConstants {
     public static final String EXPERIMENTAL_FLOW_SET_LEVEL = "experimental_flow_set_level";
     public static final String EXPERIMENTAL_FLOW_SET_LEVEL_SCORE = "experimental_flow_set_level_score";
-    public static final String SUPPLY_FLOW_SET_LEVEL = "supply_flow_set_level";
 
     public static final String PUSH_FORM = "flow_pool";
     public static final String QUICK_PUSH_FORM = "quick_flow_pool";
@@ -16,16 +15,11 @@ public class FlowPoolConstants {
     public static final String KEY_QUICK_WITH_LEVEL_FORMAT = "flow:pool:quick:item:%s:3";
     public static final String KEY_WITH_LEVEL_SCORE_FORMAT = "flow:pool:level:item:score:%s:%s";
     public static final String KEY_QUICK_WITH_LEVEL_SCORE_FORMAT = "flow:pool:quick:item:score:%s:3";
-    public static final String KEY_WITH_SCORE_FORMAT = "flow:pool:item:score:%s";
-    public static final String KEY_QUICK_WITH_SCORE_FORMAT = "flow:pool:quick:item:score:%s:3";
 
 
-    public static final String KEY_WITH_LEVEL_FORMAT_V2 = "flow:pool:level:item:0";
-    public static final String KEY_WITH_LEVEL_SUPPLY_FORMAT_V2 = "flow:pool:level:item:supply:0";
+    public static final String KEY_WITH_LEVEL_FORMAT_V2 = "flow:pool:level:item:v2:%s:%s";
     public static final String KEY_QUICK_WITH_LEVEL_FORMAT_V2 = "flow:pool:quick:item:0:3";
-    public static final String KEY_WITH_LEVEL_SCORE_FORMAT_V2 = "flow:pool:level:item:score:0";
+    public static final String KEY_WITH_LEVEL_SCORE_FORMAT_V2 = "flow:pool:level:item:score:v2:%s:%s";
     public static final String KEY_QUICK_WITH_LEVEL_SCORE_FORMAT_V2 = "flow:pool:quick:item:score:0:3";
-    public static final String KEY_WITH_SCORE_FORMAT_V2 = "flow:pool:item:score:0";
-    public static final String KEY_QUICK_WITH_SCORE_FORMAT_V2 = "flow:pool:quick:item:score:0:3";
 
 }

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/flowpool/FlowPoolService.java

@@ -9,6 +9,7 @@ import org.apache.commons.collections4.MapUtils;
 import org.apache.commons.lang3.math.NumberUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
 
@@ -32,16 +33,14 @@ public class FlowPoolService {
     private FlowPoolConfigService flowPoolConfigService;
 
     private final String localDistributeCountFormat = "flow:pool:local:distribute:count:%s:%s";
-    /**
-     * 供给池 本地缓存
-     * flow:pool:supply:local:distribute:count:{videoId}:{flowPool标记}
-     */
-    private final String supplyLocalDistributeCountFormat = "flow:pool:supply:local:distribute:count:%s:%s";
 
     public final String valueFormat = "%s-%s";
 
     private ExecutorService pool = ThreadPoolFactory.defaultPool();
 
+    @Value("${flow.pool.upgrade.switch:true}")
+    private boolean flowPoolUpgradeSwitch;
+
     private int[] appTypes = {VLOG.getCode(), LOVE_MOVIE.getCode(), LOVE_LIVE.getCode(), LONG_VIDEO.getCode(),
             SHORT_VIDEO.getCode(), H5.getCode(), APP_SPEED.getCode(), WAN_NENG_VIDEO.getCode(),
             LAO_HAO_KAN_VIDEO.getCode(), ZUI_JING_QI.getCode(), PIAO_QUAN_VIDEO_PLUS.getCode(), JOURNEY.getCode()};
@@ -78,16 +77,20 @@ public class FlowPoolService {
     }
 
     private void asyncDelDistributeCountWithLevel(Map<Long, String> videoFlowPoolMap) {
-        asyncDelDistributeCount(videoFlowPoolMap, (appType, level, values) -> {
-            String key = String.format(KEY_WITH_LEVEL_FORMAT, appType, level);
-            Long count = redisTemplate.opsForSet().remove(key, values);
-            redisTemplate.opsForSet().remove(String.format(KEY_QUICK_WITH_LEVEL_FORMAT, appType), values);
-        });
+        if (flowPoolUpgradeSwitch) {
+            asyncDelDistributeCountV2(videoFlowPoolMap, (appType, level, values) -> {
+                redisTemplate.opsForSet().remove(String.format(KEY_WITH_LEVEL_FORMAT_V2, appType, level), values);
+                redisTemplate.opsForSet().remove(KEY_QUICK_WITH_LEVEL_FORMAT_V2, values);
+            });
+        } else {
+            asyncDelDistributeCount(videoFlowPoolMap, (appType, level, values) -> {
+                String key = String.format(KEY_WITH_LEVEL_FORMAT, appType, level);
+                Long count = redisTemplate.opsForSet().remove(key, values);
+                redisTemplate.opsForSet().remove(String.format(KEY_QUICK_WITH_LEVEL_FORMAT, appType), values);
+            });
+        }
+
 
-        asyncDelDistributeCount(videoFlowPoolMap, (appType, level, values) -> {
-            redisTemplate.opsForSet().remove(KEY_WITH_LEVEL_FORMAT_V2, values);
-            redisTemplate.opsForSet().remove(KEY_QUICK_WITH_LEVEL_FORMAT_V2, values);
-        });
     }
 
     public Map<Long, Integer> getDistributeCountWithLevelScore(Map<Long, String> videoFlowPoolMap) {
@@ -118,54 +121,18 @@ public class FlowPoolService {
     }
 
     private void asyncLocalDistributeCountWithLevelScore(Map<Long, String> videoFlowPoolMap) {
-        asyncDelDistributeCount(videoFlowPoolMap, (appType, level, values) -> {
-            String key = String.format(KEY_WITH_LEVEL_SCORE_FORMAT, appType, level);
-            redisTemplate.opsForZSet().remove(key, values);
-            redisTemplate.opsForZSet().remove(String.format(KEY_QUICK_WITH_LEVEL_SCORE_FORMAT, appType), values);
-        });
-
-        asyncDelDistributeCount(videoFlowPoolMap, (appType, level, values) -> {
-            redisTemplate.opsForZSet().remove(KEY_WITH_LEVEL_SCORE_FORMAT_V2, values);
-            redisTemplate.opsForZSet().remove(KEY_QUICK_WITH_LEVEL_SCORE_FORMAT_V2, values);
-        });
-    }
-
-    public Map<Long, Integer> getDistributeCountWithScore(Map<Long, String> videoFlowPoolMap) {
-        if (MapUtils.isEmpty(videoFlowPoolMap)) {
-            return Collections.emptyMap();
-        }
-        Map<Long, Integer> result = getDistributeCount(videoFlowPoolMap);
-        // 处理脏数据：分发数<0
-        Map<Long, String> dirties = videoFlowPoolMap.entrySet().stream()
-                .filter(e -> result.get(e.getKey()) <= 0)
-                .collect(Collectors.toMap(
-                        e -> e.getKey(),
-                        e -> e.getValue()
-                ));
-        asyncDelDistributeCountWithScore(dirties);
-        return result;
-    }
-
-    public void updateDistributeCountWithScore(List<Video> videos) {
-        if (CollectionUtils.isEmpty(videos)) {
-            return;
+        if (flowPoolUpgradeSwitch) {
+            asyncDelDistributeCountV2(videoFlowPoolMap, (appType, level, values) -> {
+                redisTemplate.opsForZSet().remove(String.format(KEY_WITH_LEVEL_SCORE_FORMAT_V2, appType, level), values);
+                redisTemplate.opsForZSet().remove(KEY_QUICK_WITH_LEVEL_SCORE_FORMAT_V2, values);
+            });
+        } else {
+            asyncDelDistributeCount(videoFlowPoolMap, (appType, level, values) -> {
+                String key = String.format(KEY_WITH_LEVEL_SCORE_FORMAT, appType, level);
+                redisTemplate.opsForZSet().remove(key, values);
+                redisTemplate.opsForZSet().remove(String.format(KEY_QUICK_WITH_LEVEL_SCORE_FORMAT, appType), values);
+            });
         }
-        Map<Long, String> removeMap = updateDistributeCount(videos);
-        asyncDelDistributeCountWithScore(removeMap);
-    }
-
-    private void asyncDelDistributeCountWithScore(Map<Long, String> videoFlowPoolMap) {
-        asyncDelDistributeCount(videoFlowPoolMap, (appType, level, values) -> {
-            String key = String.format(KEY_WITH_SCORE_FORMAT, appType, level);
-            Long count = redisTemplate.opsForZSet().remove(key, values);
-
-            redisTemplate.opsForZSet().remove(String.format(KEY_QUICK_WITH_SCORE_FORMAT, appType), values);
-        });
-
-        asyncDelDistributeCount(videoFlowPoolMap, (appType, level, values) -> {
-            redisTemplate.opsForZSet().remove(KEY_WITH_SCORE_FORMAT_V2, values);
-            redisTemplate.opsForZSet().remove(KEY_QUICK_WITH_SCORE_FORMAT_V2, values);
-        });
     }
 
     private Map<Long, Integer> getDistributeCount(Map<Long, String> videoFlowPoolMap) {
@@ -209,7 +176,7 @@ public class FlowPoolService {
                     .collect(Collectors.toList());
             redisTemplate.delete(keys);
 
-            Map<String, Double> levelWeight = flowPoolConfigService.getLevelWeight();
+            Map<String, Double> levelWeight = flowPoolConfigService.getLevelWeight4FlowPoolWithLevel();
             String[] values = new String[videoFlowPoolMap.size()];
             int i = 0;
             for (Map.Entry v : videoFlowPoolMap.entrySet()) {
@@ -225,70 +192,28 @@ public class FlowPoolService {
         });
     }
 
-    public Map<Long, Integer> getSupplyDistributeCountWithLevel(Map<Long, String> videoFlowPoolMap) {
-        if (MapUtils.isEmpty(videoFlowPoolMap)) {
-            return Collections.emptyMap();
-        }
-
-        Map<Long, Integer> result = getSupplyDistributeCount(videoFlowPoolMap);
-
-
-        // 处理脏数据：分发数<0
-        Map<Long, String> dirties = videoFlowPoolMap.entrySet().stream()
-                .filter(e -> result.get(e.getKey()) <= 0)
-                .collect(Collectors.toMap(
-                        e -> e.getKey(),
-                        e -> e.getValue()
-                ));
-        asyncDelSupplyDistributeCountWithLevel(dirties);
 
-        return result;
-    }
-
-    private Map<Long, Integer> getSupplyDistributeCount(Map<Long, String> videoFlowPoolMap) {
-        // 为了保证有序
-        List<Map.Entry<Long, String>> entries = videoFlowPoolMap.entrySet().stream()
-                .sorted(Comparator.comparingLong(e -> e.getKey()))
-                .collect(Collectors.toList());
-
-        List<String> keys = entries.stream()
-                .map(v -> String.format(supplyLocalDistributeCountFormat, v.getKey(), v.getValue()))
-                .collect(Collectors.toList());
-        List<String> counts = redisTemplate.opsForValue().multiGet(keys);
-        Map<Long, Integer> result = new HashMap<>();
-        for (int i = 0; i < entries.size(); i++) {
-            result.put(entries.get(i).getKey(), NumberUtils.toInt(counts.get(i), 0));
-        }
-        return result;
-    }
-
-    private Map<Long, String> updateSupplyDistributeCount(List<Video> videos) {
-        // TODO 异步更新
-        Map<Long, String> removeMap = new HashMap<>();
-        videos.stream().forEach(v -> {
-            String key = String.format(supplyLocalDistributeCountFormat, v.getVideoId(), v.getFlowPool());
-            Long count = redisTemplate.opsForValue().decrement(key);
-            if (count <= 0) {
-                removeMap.put(v.getVideoId(), v.getFlowPool());
-            }
-        });
-        return removeMap;
-    }
-
-    public void updateSupplyDistributeCountWithLevel(List<Video> videos) {
-        if (CollectionUtils.isEmpty(videos)) {
+    private void asyncDelDistributeCountV2(Map<Long, String> videoFlowPoolMap,
+                                           TripleConsumer<Integer, String, String[]> flowPoolRemoveConsumer) {
+        if (MapUtils.isEmpty(videoFlowPoolMap)) {
             return;
         }
-        Map<Long, String> removeMap = updateSupplyDistributeCount(videos);
-
-        asyncDelSupplyDistributeCountWithLevel(removeMap);
+        pool.execute(() -> {
+            List<String> keys = videoFlowPoolMap.entrySet().stream()
+                    .map(v -> String.format(localDistributeCountFormat, v.getKey(), v.getValue()))
+                    .collect(Collectors.toList());
+            redisTemplate.delete(keys);
 
-    }
+            String[] values = new String[videoFlowPoolMap.size()];
+            int i = 0;
+            for (Map.Entry v : videoFlowPoolMap.entrySet()) {
+                values[i++] = String.format(valueFormat, v.getKey(), v.getValue());
+            }
 
-    private void asyncDelSupplyDistributeCountWithLevel(Map<Long, String> videoFlowPoolMap) {
-        asyncDelDistributeCount(videoFlowPoolMap, (appType, level, values) -> {
-            String key = String.format(KEY_WITH_LEVEL_SUPPLY_FORMAT, appType, level);
-            Long count = redisTemplate.opsForSet().remove(key, values);
+            Map<String, Double> levelWeight = flowPoolConfigService.getLevelWeight4FlowPoolWithLevel();
+            for (String level : levelWeight.keySet()) {
+                flowPoolRemoveConsumer.accept(0, level, values);
+            }
         });
     }
 }

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/recall/RecallService.java

@@ -189,8 +189,6 @@ public class RecallService implements ApplicationContextAware {
                 strategies.add(strategyMap.get(FlowPoolWithLevelRecallStrategyTomson.class.getSimpleName()));
             } else if (param.getFlowPoolAbtestGroup().equals(FlowPoolConstants.EXPERIMENTAL_FLOW_SET_LEVEL_SCORE)) {
                 strategies.add(strategyMap.get(FlowPoolWithLevelScoreRecallStrategy.class.getSimpleName()));
-            } else {
-                strategies.add(strategyMap.get(FlowPoolWithScoreRecallStrategy.class.getSimpleName()));
             }
         }
 

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/recall/strategy/AbstractFlowPoolWithScoreRecallStrategy.java

@@ -1,68 +0,0 @@
-package com.tzld.piaoquan.recommend.server.service.recall.strategy;
-
-import com.tzld.piaoquan.recommend.server.model.Video;
-import com.tzld.piaoquan.recommend.server.service.filter.FilterResult;
-import com.tzld.piaoquan.recommend.server.service.filter.FlowPoolWithScoreFilterService;
-import com.tzld.piaoquan.recommend.server.service.recall.FilterParamFactory;
-import com.tzld.piaoquan.recommend.server.service.recall.RecallParam;
-import com.tzld.piaoquan.recommend.server.service.recall.RecallStrategy;
-import org.apache.commons.collections4.CollectionUtils;
-import org.apache.commons.lang3.math.NumberUtils;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.data.redis.core.RedisTemplate;
-import org.springframework.data.redis.core.ZSetOperations;
-
-import java.util.*;
-
-/**
- * @author dyp
- */
-public abstract class AbstractFlowPoolWithScoreRecallStrategy implements RecallStrategy {
-    @Autowired
-    @Qualifier("redisTemplate")
-    protected RedisTemplate<String, String> redisTemplate;
-
-    @Autowired
-    protected FlowPoolWithScoreFilterService filterService;
-
-    @Override
-    public List<Video> recall(RecallParam param) {
-        String flowPoolKey = flowPoolKey(param);
-
-        int getSize = param.getSize() * 5;
-        List<Video> results = new ArrayList<>();
-        Set<ZSetOperations.TypedTuple<String>> data = redisTemplate.opsForZSet().reverseRangeWithScores(flowPoolKey, 0,
-                getSize - 1);
-        if (CollectionUtils.isEmpty(data)) {
-            return null;
-        }
-
-        Map<Long, Double> videoScoreMap = new HashMap<>();
-        Map<Long, String> videoFlowPoolMap = new LinkedHashMap<>();
-        for (ZSetOperations.TypedTuple<String> t : data) {
-            String[] values = t.getValue().split("-");
-            videoFlowPoolMap.put(NumberUtils.toLong(values[0], 0), values[1]);
-            videoScoreMap.put(NumberUtils.toLong(values[0], 0), t.getScore());
-        }
-        FilterResult filterResult = filterService.filter(FilterParamFactory.create(param, videoFlowPoolMap));
-
-        if (filterResult != null && CollectionUtils.isNotEmpty(filterResult.getVideoIds())) {
-            filterResult.getVideoIds().stream().forEach(vid -> {
-                Video recallData = new Video();
-                recallData.setVideoId(vid);
-                recallData.setAbCode(param.getAbCode());
-                recallData.setRovScore(videoScoreMap.get(vid));
-                recallData.setPushFrom(pushFrom());
-                recallData.setFlowPool(videoFlowPoolMap.get(vid));
-                recallData.setFlowPoolAbtestGroup(param.getFlowPoolAbtestGroup());
-                results.add(recallData);
-            });
-        }
-
-        return results.subList(0, results.size() < param.getSize() ? results.size() : param.getSize());
-    }
-
-    abstract String flowPoolKey(RecallParam param);
-
-}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/recall/strategy/FlowPoolWithLevelRecallStrategy.java

@@ -1,105 +0,0 @@
-package com.tzld.piaoquan.recommend.server.service.recall.strategy;
-
-import com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConfigService;
-import com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConstants;
-import com.tzld.piaoquan.recommend.server.service.recall.RecallParam;
-import com.tzld.piaoquan.recommend.server.util.JSONUtils;
-import lombok.Data;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.collections4.CollectionUtils;
-import org.apache.commons.lang3.RandomUtils;
-import org.apache.commons.lang3.tuple.Pair;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
-
-import java.math.BigDecimal;
-import java.math.RoundingMode;
-import java.util.*;
-
-import static com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConstants.KEY_WITH_LEVEL_FORMAT;
-import static com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConstants.KEY_WITH_LEVEL_FORMAT_V2;
-
-/**
- * @author dyp
- */
-@Service
-@Slf4j
-@Deprecated
-public class FlowPoolWithLevelRecallStrategy extends AbstractFlowPoolWithLevelRecallStrategy {
-
-    @Autowired
-    private FlowPoolConfigService flowPoolConfigService;
-
-    @Override
-    Pair<String, String> flowPoolKeyAndLevel(RecallParam param) {
-        //# 1. 获取流量池各层级分发概率权重
-        Map<String, Double> levelWeightMap = flowPoolConfigService.getLevelWeight();
-
-        // 2. 判断各层级是否有视频需分发
-        List<LevelWeight> availableLevels = new ArrayList<>();
-        for (Map.Entry<String, Double> entry : levelWeightMap.entrySet()) {
-            String levelKey = String.format(KEY_WITH_LEVEL_FORMAT, param.getAppType(), entry.getKey());
-            if (redisTemplate.hasKey(levelKey)) {
-                LevelWeight lw = new LevelWeight();
-                lw.setLevel(entry.getKey());
-                lw.setLevelKey(levelKey);
-                lw.setWeight(entry.getValue());
-                availableLevels.add(lw);
-            }
-        }
-        if (CollectionUtils.isEmpty(availableLevels)) {
-            return Pair.of("", "");
-        }
-
-        // 3. 根据可分发层级权重设置分发概率
-        Collections.sort(availableLevels, Comparator.comparingDouble(LevelWeight::getWeight));
-
-        double weightSum = availableLevels.stream().mapToDouble(o -> o.getWeight()).sum();
-        BigDecimal weightSumBD = new BigDecimal(weightSum);
-        double level_p_low = 0;
-        double weight_temp = 0;
-        double level_p_up = 0;
-        Map<String, LevelP> level_p_mapping = new HashMap<>();
-        for (LevelWeight lw : availableLevels) {
-            BigDecimal bd = new BigDecimal(weight_temp + lw.getWeight());
-            level_p_up = bd.divide(weightSumBD, 2, RoundingMode.HALF_UP).doubleValue();
-            LevelP levelP = new LevelP();
-            levelP.setMin(level_p_low);
-            levelP.setMax(level_p_up);
-            levelP.setLevelKey(lw.getLevelKey());
-            level_p_mapping.put(lw.level, levelP);
-            level_p_low = level_p_up;
-
-            weight_temp += lw.getWeight();
-        }
-
-        // 4. 随机生成[0,1)之间数，返回相应概率区间的key
-        double random_p = RandomUtils.nextDouble(0, 1);
-        for (Map.Entry<String, LevelP> entry : level_p_mapping.entrySet()) {
-            if (random_p >= entry.getValue().getMin()
-                    && random_p <= entry.getValue().getMax()) {
-                return Pair.of(entry.getValue().getLevelKey(), entry.getKey());
-            }
-        }
-        return Pair.of("", "");
-    }
-
-    @Data
-    static class LevelWeight {
-        private String level;
-        private String levelKey;
-        private Double weight;
-    }
-
-    @Data
-    static class LevelP {
-        private String levelKey;
-        private double min;
-        private double max;
-    }
-
-    @Override
-    public String pushFrom() {
-        return FlowPoolConstants.PUSH_FORM;
-    }
-}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/recall/strategy/FlowPoolWithLevelRecallStrategyFilterDigit.java

@@ -1,146 +0,0 @@
-package com.tzld.piaoquan.recommend.server.service.recall.strategy;
-
-import com.tzld.piaoquan.recommend.server.model.Video;
-import com.tzld.piaoquan.recommend.server.service.filter.FilterResult;
-import com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConfigService;
-import com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConstants;
-import com.tzld.piaoquan.recommend.server.service.recall.FilterParamFactory;
-import com.tzld.piaoquan.recommend.server.service.recall.RecallParam;
-import lombok.Data;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.collections4.CollectionUtils;
-import org.apache.commons.lang3.RandomUtils;
-import org.apache.commons.lang3.math.NumberUtils;
-import org.apache.commons.lang3.tuple.Pair;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
-
-import java.math.BigDecimal;
-import java.math.RoundingMode;
-import java.util.*;
-import java.util.stream.Collectors;
-
-import static com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConstants.KEY_WITH_LEVEL_FORMAT;
-
-/**
- * @author dyp
- */
-@Service
-@Slf4j
-@Deprecated
-public class FlowPoolWithLevelRecallStrategyFilterDigit extends AbstractFlowPoolWithLevelRecallStrategy {
-
-    @Autowired
-    private FlowPoolConfigService flowPoolConfigService;
-
-    @Override
-    Pair<String, String> flowPoolKeyAndLevel(RecallParam param) {
-        //# 1. 获取流量池各层级分发概率权重
-        Map<String, Double> levelWeightMap = flowPoolConfigService.getLevelWeight();
-
-        // 2. 判断各层级是否有视频需分发
-        List<LevelWeight> availableLevels = new ArrayList<>();
-        for (Map.Entry<String, Double> entry : levelWeightMap.entrySet()) {
-            String levelKey = String.format(KEY_WITH_LEVEL_FORMAT, param.getAppType(), entry.getKey());
-            if (Boolean.TRUE.equals(redisTemplate.hasKey(levelKey))) {
-                LevelWeight lw = new LevelWeight();
-                lw.setLevel(entry.getKey());
-                lw.setLevelKey(levelKey);
-                lw.setWeight(entry.getValue());
-                availableLevels.add(lw);
-            }
-        }
-        if (CollectionUtils.isEmpty(availableLevels)) {
-            return Pair.of("", "");
-        }
-        // 3. 根据可分发层级权重设置分发概率
-        availableLevels.sort(Comparator.comparingDouble(LevelWeight::getWeight));
-
-        double weightSum = availableLevels.stream().mapToDouble(LevelWeight::getWeight).sum();
-        BigDecimal weightSumBD = new BigDecimal(weightSum);
-        double level_p_low = 0;
-        double weight_temp = 0;
-        double level_p_up = 0;
-        Map<String, LevelP> level_p_mapping = new HashMap<>();
-        for (LevelWeight lw : availableLevels) {
-            BigDecimal bd = new BigDecimal(weight_temp + lw.getWeight());
-            level_p_up = bd.divide(weightSumBD, 2, RoundingMode.HALF_UP).doubleValue();
-            LevelP levelP = new LevelP();
-            levelP.setMin(level_p_low);
-            levelP.setMax(level_p_up);
-            levelP.setLevelKey(lw.getLevelKey());
-            level_p_mapping.put(lw.level, levelP);
-            level_p_low = level_p_up;
-
-            weight_temp += lw.getWeight();
-        }
-
-        // 4. 随机生成[0,1)之间数，返回相应概率区间的key
-        double random_p = RandomUtils.nextDouble(0, 1);
-        for (Map.Entry<String, LevelP> entry : level_p_mapping.entrySet()) {
-            if (random_p >= entry.getValue().getMin()
-                    && random_p <= entry.getValue().getMax()) {
-                return Pair.of(entry.getValue().getLevelKey(), entry.getKey());
-            }
-        }
-        return Pair.of("", "");
-    }
-
-    @Data
-    static class LevelWeight {
-        private String level;
-        private String levelKey;
-        private Double weight;
-    }
-
-    @Data
-    static class LevelP {
-        private String levelKey;
-        private double min;
-        private double max;
-    }
-
-    @Override
-    public String pushFrom() {
-        return FlowPoolConstants.PUSH_FORM;
-    }
-
-    @Override
-    public List<Video> recall(RecallParam param) {
-        Pair<String, String> flowPoolKeyAndLevel = flowPoolKeyAndLevel(param);
-        String flowPoolKey = flowPoolKeyAndLevel.getLeft();
-        String level = flowPoolKeyAndLevel.getRight();
-        int getSize = param.getSize() * 5;
-        List<Video> results = new ArrayList<>();
-        List<String> list = Objects.requireNonNull(redisTemplate.opsForSet().members(flowPoolKey)).stream().filter(o ->
-                NumberUtils.toLong(o.split("-")[0], 0) % 10 == param.getLastDigit()).distinct().collect(Collectors.toList());
-        Collections.shuffle(list);
-        List<String> data = list.subList(0, Math.min(getSize, list.size()));
-        if (CollectionUtils.isEmpty(data)) {
-            return null;
-        }
-        Map<Long, String> videoFlowPoolMap = new LinkedHashMap<>();
-        for (String value : data) {
-            String[] values = value.split("-");
-            videoFlowPoolMap.put(NumberUtils.toLong(values[0], 0), values[1]);
-        }
-
-        FilterResult filterResult = filterService.filter(FilterParamFactory.create(param, videoFlowPoolMap));
-
-        if (filterResult != null && CollectionUtils.isNotEmpty(filterResult.getVideoIds())) {
-            filterResult.getVideoIds().forEach(vid -> {
-                Video recallData = new Video();
-                recallData.setVideoId(vid);
-                recallData.setAbCode(param.getAbCode());
-                recallData.setRovScore(RandomUtils.nextDouble(0, 100));
-                recallData.setPushFrom(pushFrom());
-                recallData.setFlowPool(videoFlowPoolMap.get(vid));
-                recallData.setFlowPoolAbtestGroup(param.getFlowPoolAbtestGroup());
-                recallData.setLevel(level);
-                results.add(recallData);
-            });
-        }
-
-        return results.subList(0, Math.min(results.size(), param.getSize()));
-    }
-}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/recall/strategy/FlowPoolWithLevelRecallStrategyTomson.java

@@ -42,16 +42,15 @@ public class FlowPoolWithLevelRecallStrategyTomson extends AbstractFlowPoolWithL
 
     @Override
     Pair<String, String> flowPoolKeyAndLevel(RecallParam param) {
-        if (param.isFlowPoolUpgrade()) {
-            return Pair.of(KEY_WITH_LEVEL_FORMAT_V2, "-1");
-        }
         //# 1. 获取流量池各层级分发概率权重
-        Map<String, Double> levelWeightMap = flowPoolConfigService.getLevelWeight();
+        Map<String, Double> levelWeightMap = flowPoolConfigService.getLevelWeight4FlowPoolWithLevel();
 
         // 2. 判断各层级是否有视频需分发
         List<LevelWeight> availableLevels = new ArrayList<>();
         for (Map.Entry<String, Double> entry : levelWeightMap.entrySet()) {
-            String levelKey = String.format(KEY_WITH_LEVEL_FORMAT, param.getAppType(), entry.getKey());
+            String levelKey = param.isFlowPoolUpgrade()
+                    ? String.format(KEY_WITH_LEVEL_FORMAT_V2, 0, entry.getKey())
+                    : String.format(KEY_WITH_LEVEL_FORMAT, param.getAppType(), entry.getKey());
             if (redisTemplate.hasKey(levelKey)) {
                 LevelWeight lw = new LevelWeight();
                 lw.setLevel(entry.getKey());
@@ -121,6 +120,7 @@ public class FlowPoolWithLevelRecallStrategyTomson extends AbstractFlowPoolWithL
     @Override
     public List<Video> recall(RecallParam param) {
         Pair<String, String> flowPoolKeyAndLevel = flowPoolKeyAndLevel(param);
+        //log.info("flowPoolKeyAndLevel {}", JSONUtils.toJson(flowPoolKeyAndLevel));
         String flowPoolKey = flowPoolKeyAndLevel.getLeft();
         String level = flowPoolKeyAndLevel.getRight();
         Set<String> data = redisTemplate.opsForSet().members(flowPoolKey);

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/recall/strategy/FlowPoolWithLevelRecallStrategyTomsonFilterDigit.java

@@ -1,162 +0,0 @@
-package com.tzld.piaoquan.recommend.server.service.recall.strategy;
-
-import com.tzld.piaoquan.recommend.server.model.Video;
-import com.tzld.piaoquan.recommend.server.service.filter.FilterParam;
-import com.tzld.piaoquan.recommend.server.service.filter.FilterResult;
-import com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConfigService;
-import com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConstants;
-import com.tzld.piaoquan.recommend.server.service.recall.FilterParamFactory;
-import com.tzld.piaoquan.recommend.server.service.recall.RecallParam;
-import com.tzld.piaoquan.recommend.server.service.score.ScorerUtils;
-import com.tzld.piaoquan.recommend.server.service.score4recall.ScorerPipeline4Recall;
-import lombok.Data;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.collections4.CollectionUtils;
-import org.apache.commons.lang3.RandomUtils;
-import org.apache.commons.lang3.math.NumberUtils;
-import org.apache.commons.lang3.tuple.Pair;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
-
-import java.math.BigDecimal;
-import java.math.RoundingMode;
-import java.util.*;
-import java.util.stream.Collectors;
-
-import static com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConstants.KEY_WITH_LEVEL_FORMAT;
-
-/**
- * @author zhangbo
- */
-@Service
-@Slf4j
-@Deprecated
-public class FlowPoolWithLevelRecallStrategyTomsonFilterDigit extends AbstractFlowPoolWithLevelRecallStrategy {
-
-    @Autowired
-    private FlowPoolConfigService flowPoolConfigService;
-
-    @Override
-    Pair<String, String> flowPoolKeyAndLevel(RecallParam param) {
-        //# 1. 获取流量池各层级分发概率权重
-        Map<String, Double> levelWeightMap = flowPoolConfigService.getLevelWeight();
-
-        // 2. 判断各层级是否有视频需分发
-        List<LevelWeight> availableLevels = new ArrayList<>();
-        for (Map.Entry<String, Double> entry : levelWeightMap.entrySet()) {
-            String levelKey = String.format(KEY_WITH_LEVEL_FORMAT, param.getAppType(), entry.getKey());
-            if (Boolean.TRUE.equals(redisTemplate.hasKey(levelKey))) {
-                LevelWeight lw = new LevelWeight();
-                lw.setLevel(entry.getKey());
-                lw.setLevelKey(levelKey);
-                lw.setWeight(entry.getValue());
-                availableLevels.add(lw);
-            }
-        }
-        if (CollectionUtils.isEmpty(availableLevels)) {
-            return Pair.of("", "");
-        }
-
-        // 3. 根据可分发层级权重设置分发概率
-        availableLevels.sort(Comparator.comparingDouble(LevelWeight::getWeight));
-
-        double weightSum = availableLevels.stream().mapToDouble(o -> o.getWeight()).sum();
-        BigDecimal weightSumBD = new BigDecimal(weightSum);
-        double level_p_low = 0;
-        double weight_temp = 0;
-        double level_p_up = 0;
-        Map<String, LevelP> level_p_mapping = new HashMap<>();
-        for (LevelWeight lw : availableLevels) {
-            BigDecimal bd = new BigDecimal(weight_temp + lw.getWeight());
-            level_p_up = bd.divide(weightSumBD, 2, RoundingMode.HALF_UP).doubleValue();
-            LevelP levelP = new LevelP();
-            levelP.setMin(level_p_low);
-            levelP.setMax(level_p_up);
-            levelP.setLevelKey(lw.getLevelKey());
-            level_p_mapping.put(lw.level, levelP);
-            level_p_low = level_p_up;
-
-            weight_temp += lw.getWeight();
-        }
-
-        // 4. 随机生成[0,1)之间数，返回相应概率区间的key
-        double random_p = RandomUtils.nextDouble(0, 1);
-        for (Map.Entry<String, LevelP> entry : level_p_mapping.entrySet()) {
-            if (random_p >= entry.getValue().getMin()
-                    && random_p <= entry.getValue().getMax()) {
-                return Pair.of(entry.getValue().getLevelKey(), entry.getKey());
-            }
-        }
-        return Pair.of("", "");
-    }
-
-    @Data
-    static class LevelWeight {
-        private String level;
-        private String levelKey;
-        private Double weight;
-    }
-
-    @Data
-    static class LevelP {
-        private String levelKey;
-        private double min;
-        private double max;
-    }
-
-    @Override
-    public String pushFrom() {
-        return FlowPoolConstants.PUSH_FORM;
-    }
-
-    @Override
-    public List<Video> recall(RecallParam param) {
-        Pair<String, String> flowPoolKeyAndLevel = flowPoolKeyAndLevel(param);
-        String flowPoolKey = flowPoolKeyAndLevel.getLeft();
-        String level = flowPoolKeyAndLevel.getRight();
-        List<String> data = Objects.requireNonNull(redisTemplate.opsForSet().members(flowPoolKey)).stream().filter(o ->
-                NumberUtils.toLong(o.split("-")[0], 0) % 10 == param.getLastDigit()).distinct().collect(Collectors.toList());
-
-        if (CollectionUtils.isEmpty(data)) {
-            return null;
-        }
-        Map<String, String> videoFlowPoolMap = new LinkedHashMap<>();
-        Map<Long, String> videoFlowPoolMap_ = new LinkedHashMap<>();
-        for (String value : data) {
-            String[] values = value.split("-");
-            videoFlowPoolMap.put(values[0], values[1]);
-            videoFlowPoolMap_.put(NumberUtils.toLong(values[0], 0), values[1]);
-        }
-        ScorerPipeline4Recall pipeline = ScorerUtils.getScorerPipeline4Recall("feeds_recall_config_tomson.conf");
-        List<List<Pair<Long, Double>>> results = pipeline.recall(videoFlowPoolMap);
-        List<Pair<Long, Double>> result = results.get(0);
-        Map<Long, Double> resultmap = result.stream()
-                .collect(Collectors.toMap(
-                        Pair::getLeft, // 键是Pair的left值
-                        Pair::getRight, // 值是Pair的right值
-                        (existingValue, newValue) -> existingValue, // 如果键冲突，选择保留现有的值（或者你可以根据需要定义其他合并策略）
-                        LinkedHashMap::new // 使用LinkedHashMap来保持插入顺序（如果需要的话）
-                ));
-        // 3 召回内部过滤
-        FilterParam filterParam = FilterParamFactory.create(param, result.stream()
-                .map(Pair::getLeft)
-                .collect(Collectors.toList()));
-        FilterResult filterResult = filterService.filter(filterParam);
-        List<Video> videosResult = new ArrayList<>();
-        if (filterResult != null && CollectionUtils.isNotEmpty(filterResult.getVideoIds())) {
-            filterResult.getVideoIds().forEach(vid -> {
-                Video recallData = new Video();
-                recallData.setVideoId(vid);
-                recallData.setAbCode(param.getAbCode());
-                recallData.setRovScore(resultmap.getOrDefault(vid, 0.0));
-                recallData.setPushFrom(pushFrom());
-                recallData.setFlowPool(videoFlowPoolMap_.get(vid));
-                recallData.setFlowPoolAbtestGroup(param.getFlowPoolAbtestGroup());
-                recallData.setLevel(level);
-                videosResult.add(recallData);
-            });
-        }
-        videosResult.sort(Comparator.comparingDouble(o -> -o.getRovScore()));
-        return videosResult;
-    }
-}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/recall/strategy/FlowPoolWithLevelRecallStrategyTomsonFilterDigitV2.java

@@ -1,162 +0,0 @@
-package com.tzld.piaoquan.recommend.server.service.recall.strategy;
-
-import com.tzld.piaoquan.recommend.server.model.Video;
-import com.tzld.piaoquan.recommend.server.service.filter.FilterParam;
-import com.tzld.piaoquan.recommend.server.service.filter.FilterResult;
-import com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConfigService;
-import com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConstants;
-import com.tzld.piaoquan.recommend.server.service.recall.FilterParamFactory;
-import com.tzld.piaoquan.recommend.server.service.recall.RecallParam;
-import com.tzld.piaoquan.recommend.server.service.score.ScorerUtils;
-import com.tzld.piaoquan.recommend.server.service.score4recall.ScorerPipeline4Recall;
-import lombok.Data;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.collections4.CollectionUtils;
-import org.apache.commons.lang3.RandomUtils;
-import org.apache.commons.lang3.math.NumberUtils;
-import org.apache.commons.lang3.tuple.Pair;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
-
-import java.math.BigDecimal;
-import java.math.RoundingMode;
-import java.util.*;
-import java.util.stream.Collectors;
-
-import static com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConstants.KEY_WITH_LEVEL_FORMAT;
-
-/**
- * @author zhangbo
- */
-@Service
-@Slf4j
-@Deprecated
-public class FlowPoolWithLevelRecallStrategyTomsonFilterDigitV2 extends AbstractFlowPoolWithLevelRecallStrategy {
-
-    @Autowired
-    private FlowPoolConfigService flowPoolConfigService;
-
-    @Override
-    Pair<String, String> flowPoolKeyAndLevel(RecallParam param) {
-        //# 1. 获取流量池各层级分发概率权重
-        Map<String, Double> levelWeightMap = flowPoolConfigService.getLevelWeight();
-
-        // 2. 判断各层级是否有视频需分发
-        List<LevelWeight> availableLevels = new ArrayList<>();
-        for (Map.Entry<String, Double> entry : levelWeightMap.entrySet()) {
-            String levelKey = String.format(KEY_WITH_LEVEL_FORMAT, param.getAppType(), entry.getKey());
-            if (Boolean.TRUE.equals(redisTemplate.hasKey(levelKey))) {
-                LevelWeight lw = new LevelWeight();
-                lw.setLevel(entry.getKey());
-                lw.setLevelKey(levelKey);
-                lw.setWeight(entry.getValue());
-                availableLevels.add(lw);
-            }
-        }
-        if (CollectionUtils.isEmpty(availableLevels)) {
-            return Pair.of("", "");
-        }
-
-        // 3. 根据可分发层级权重设置分发概率
-        availableLevels.sort(Comparator.comparingDouble(LevelWeight::getWeight));
-
-        double weightSum = availableLevels.stream().mapToDouble(o -> o.getWeight()).sum();
-        BigDecimal weightSumBD = new BigDecimal(weightSum);
-        double level_p_low = 0;
-        double weight_temp = 0;
-        double level_p_up = 0;
-        Map<String, LevelP> level_p_mapping = new HashMap<>();
-        for (LevelWeight lw : availableLevels) {
-            BigDecimal bd = new BigDecimal(weight_temp + lw.getWeight());
-            level_p_up = bd.divide(weightSumBD, 2, RoundingMode.HALF_UP).doubleValue();
-            LevelP levelP = new LevelP();
-            levelP.setMin(level_p_low);
-            levelP.setMax(level_p_up);
-            levelP.setLevelKey(lw.getLevelKey());
-            level_p_mapping.put(lw.level, levelP);
-            level_p_low = level_p_up;
-
-            weight_temp += lw.getWeight();
-        }
-
-        // 4. 随机生成[0,1)之间数，返回相应概率区间的key
-        double random_p = RandomUtils.nextDouble(0, 1);
-        for (Map.Entry<String, LevelP> entry : level_p_mapping.entrySet()) {
-            if (random_p >= entry.getValue().getMin()
-                    && random_p <= entry.getValue().getMax()) {
-                return Pair.of(entry.getValue().getLevelKey(), entry.getKey());
-            }
-        }
-        return Pair.of("", "");
-    }
-
-    @Data
-    static class LevelWeight {
-        private String level;
-        private String levelKey;
-        private Double weight;
-    }
-
-    @Data
-    static class LevelP {
-        private String levelKey;
-        private double min;
-        private double max;
-    }
-
-    @Override
-    public String pushFrom() {
-        return FlowPoolConstants.PUSH_FORM;
-    }
-
-    @Override
-    public List<Video> recall(RecallParam param) {
-        Pair<String, String> flowPoolKeyAndLevel = flowPoolKeyAndLevel(param);
-        String flowPoolKey = flowPoolKeyAndLevel.getLeft();
-        String level = flowPoolKeyAndLevel.getRight();
-        List<String> data = Objects.requireNonNull(redisTemplate.opsForSet().members(flowPoolKey)).stream().filter(o ->
-                NumberUtils.toLong(o.split("-")[0], 0) % 10 == param.getLastDigit()).distinct().collect(Collectors.toList());
-
-        if (CollectionUtils.isEmpty(data)) {
-            return null;
-        }
-        Map<String, String> videoFlowPoolMap = new LinkedHashMap<>();
-        Map<Long, String> videoFlowPoolMap_ = new LinkedHashMap<>();
-        for (String value : data) {
-            String[] values = value.split("-");
-            videoFlowPoolMap.put(values[0], values[1]);
-            videoFlowPoolMap_.put(NumberUtils.toLong(values[0], 0), values[1]);
-        }
-        ScorerPipeline4Recall pipeline = ScorerUtils.getScorerPipeline4Recall("feeds_recall_config_tomson_v2.conf");
-        List<List<Pair<Long, Double>>> results = pipeline.recall(videoFlowPoolMap);
-        List<Pair<Long, Double>> result = results.get(0);
-        Map<Long, Double> resultmap = result.stream()
-                .collect(Collectors.toMap(
-                        Pair::getLeft, // 键是Pair的left值
-                        Pair::getRight, // 值是Pair的right值
-                        (existingValue, newValue) -> existingValue, // 如果键冲突，选择保留现有的值（或者你可以根据需要定义其他合并策略）
-                        LinkedHashMap::new // 使用LinkedHashMap来保持插入顺序（如果需要的话）
-                ));
-        // 3 召回内部过滤
-        FilterParam filterParam = FilterParamFactory.create(param, result.stream()
-                .map(Pair::getLeft)
-                .collect(Collectors.toList()));
-        FilterResult filterResult = filterService.filter(filterParam);
-        List<Video> videosResult = new ArrayList<>();
-        if (filterResult != null && CollectionUtils.isNotEmpty(filterResult.getVideoIds())) {
-            filterResult.getVideoIds().forEach(vid -> {
-                Video recallData = new Video();
-                recallData.setVideoId(vid);
-                recallData.setAbCode(param.getAbCode());
-                recallData.setRovScore(resultmap.getOrDefault(vid, 0.0));
-                recallData.setPushFrom(pushFrom());
-                recallData.setFlowPool(videoFlowPoolMap_.get(vid));
-                recallData.setFlowPoolAbtestGroup(param.getFlowPoolAbtestGroup());
-                recallData.setLevel(level);
-                videosResult.add(recallData);
-            });
-        }
-        videosResult.sort(Comparator.comparingDouble(o -> -o.getRovScore()));
-        return videosResult;
-    }
-}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/recall/strategy/FlowPoolWithLevelScoreRecallStrategy.java

@@ -41,16 +41,15 @@ public class FlowPoolWithLevelScoreRecallStrategy extends AbstractFlowPoolWithLe
 
     @Override
     Pair<String, String> flowPoolKeyAndLevel(RecallParam param) {
-        if (param.isFlowPoolUpgrade()) {
-            return Pair.of(KEY_WITH_LEVEL_SCORE_FORMAT_V2, "-1");
-        }
         //# 1. 获取流量池各层级分发概率权重
-        Map<String, Double> levelWeightMap = flowPoolConfigService.getLevelWeight();
+        Map<String, Double> levelWeightMap = flowPoolConfigService.getLevelWeight4FlowPoolWithLevelScore();
 
         // 2. 判断各层级是否有视频需分发
         List<LevelWeight> availableLevels = new ArrayList<>();
         for (Map.Entry<String, Double> entry : levelWeightMap.entrySet()) {
-            String levelKey = String.format(KEY_WITH_LEVEL_SCORE_FORMAT, param.getAppType(), entry.getKey());
+            String levelKey = param.isFlowPoolUpgrade()
+                    ? String.format(KEY_WITH_LEVEL_SCORE_FORMAT_V2, 0, entry.getKey())
+                    : String.format(KEY_WITH_LEVEL_SCORE_FORMAT, param.getAppType(), entry.getKey());
             if (Boolean.TRUE.equals(redisTemplate.hasKey(levelKey))) {
                 LevelWeight lw = new LevelWeight();
                 lw.setLevel(entry.getKey());
@@ -131,7 +130,7 @@ public class FlowPoolWithLevelScoreRecallStrategy extends AbstractFlowPoolWithLe
         for (ZSetOperations.TypedTuple<String> value : data) {
             String[] values = Objects.requireNonNull(value.getValue()).split("-");
             videoFlowPoolMap.put(values[0], values[1]);
-            videoFlowPoolMap_.put(NumberUtils.toLong(values[0], 0),  values[1]);
+            videoFlowPoolMap_.put(NumberUtils.toLong(values[0], 0), values[1]);
         }
         Map<Long, Double> resultmap = null;
         if ("1".equals(level) && ifOneLevelRandom) {

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/recall/strategy/FlowPoolWithLevelSupplyRecallStrategy.java

@@ -1,99 +0,0 @@
-package com.tzld.piaoquan.recommend.server.service.recall.strategy;
-
-import com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConfigService;
-import com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConstants;
-import com.tzld.piaoquan.recommend.server.service.recall.RecallParam;
-import lombok.Data;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.collections4.CollectionUtils;
-import org.apache.commons.lang3.RandomUtils;
-import org.apache.commons.lang3.tuple.Pair;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
-
-import java.math.BigDecimal;
-import java.math.RoundingMode;
-import java.util.*;
-
-import static com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConstants.KEY_WITH_LEVEL_SUPPLY_FORMAT;
-
-@Service
-@Slf4j
-@Deprecated
-public class FlowPoolWithLevelSupplyRecallStrategy extends AbstractFlowPoolWithLevelRecallStrategy {
-
-    @Autowired
-    private FlowPoolConfigService flowPoolConfigService;
-
-    @Override
-    Pair<String, String> flowPoolKeyAndLevel(RecallParam param) {
-        //# 1. 获取流量池各层级分发概率权重
-        Map<String, Double> levelWeightMap = flowPoolConfigService.getLevelWeight();
-        // 2. 判断各层级是否有视频需分发
-        List<LevelWeight> availableLevels = new ArrayList<>();
-        for (Map.Entry<String, Double> entry : levelWeightMap.entrySet()) {
-            String levelKey = String.format(KEY_WITH_LEVEL_SUPPLY_FORMAT, param.getAppType(), entry.getKey());
-            if (redisTemplate.hasKey(levelKey)) {
-                LevelWeight lw = new LevelWeight();
-                lw.setLevel(entry.getKey());
-                lw.setLevelKey(levelKey);
-                lw.setWeight(entry.getValue());
-                availableLevels.add(lw);
-            }
-        }
-        if (CollectionUtils.isEmpty(availableLevels)) {
-            return Pair.of("", "");
-        }
-
-        // 3. 根据可分发层级权重设置分发概率
-        Collections.sort(availableLevels, Comparator.comparingDouble(LevelWeight::getWeight));
-
-        double weightSum = availableLevels.stream().mapToDouble(o -> o.getWeight()).sum();
-        BigDecimal weightSumBD = new BigDecimal(weightSum);
-        double level_p_low = 0;
-        double weight_temp = 0;
-        double level_p_up = 0;
-        Map<String, LevelP> level_p_mapping = new HashMap<>();
-        for (LevelWeight lw : availableLevels) {
-            BigDecimal bd = new BigDecimal(weight_temp + lw.getWeight());
-            level_p_up = bd.divide(weightSumBD, 2, RoundingMode.HALF_UP).doubleValue();
-            LevelP levelP = new LevelP();
-            levelP.setMin(level_p_low);
-            levelP.setMax(level_p_up);
-            levelP.setLevelKey(lw.getLevelKey());
-            level_p_mapping.put(lw.level, levelP);
-            level_p_low = level_p_up;
-
-            weight_temp += lw.getWeight();
-        }
-
-        // 4. 随机生成[0,1)之间数，返回相应概率区间的key
-        double random_p = RandomUtils.nextDouble(0, 1);
-        for (Map.Entry<String, LevelP> entry : level_p_mapping.entrySet()) {
-            if (random_p >= entry.getValue().getMin()
-                    && random_p <= entry.getValue().getMax()) {
-                return Pair.of(entry.getValue().getLevelKey(), entry.getKey());
-            }
-        }
-        return Pair.of("", "");
-    }
-
-    @Data
-    static class LevelWeight {
-        private String level;
-        private String levelKey;
-        private Double weight;
-    }
-
-    @Data
-    static class LevelP {
-        private String levelKey;
-        private double min;
-        private double max;
-    }
-
-    @Override
-    public String pushFrom() {
-        return FlowPoolConstants.PUSH_FORM;
-    }
-}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/recall/strategy/FlowPoolWithScoreRecallStrategy.java

@@ -1,30 +0,0 @@
-package com.tzld.piaoquan.recommend.server.service.recall.strategy;
-
-import com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConstants;
-import com.tzld.piaoquan.recommend.server.service.recall.RecallParam;
-import org.apache.commons.lang3.tuple.Pair;
-import org.springframework.stereotype.Service;
-
-import static com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConstants.KEY_WITH_LEVEL_SCORE_FORMAT_V2;
-import static com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConstants.KEY_WITH_SCORE_FORMAT;
-
-/**
- * @author dyp
- */
-@Service
-
-public class FlowPoolWithScoreRecallStrategy extends AbstractFlowPoolWithScoreRecallStrategy {
-
-    @Override
-    String flowPoolKey(RecallParam param) {
-        if (param.isFlowPoolUpgrade()) {
-            return FlowPoolConstants.KEY_WITH_SCORE_FORMAT_V2;
-        }
-        return String.format(KEY_WITH_SCORE_FORMAT, param.getAppType());
-    }
-
-    @Override
-    public String pushFrom() {
-        return FlowPoolConstants.PUSH_FORM;
-    }
-}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/recall/strategy/QuickFlowPoolWithScoreRecallStrategy.java

@@ -1,28 +0,0 @@
-package com.tzld.piaoquan.recommend.server.service.recall.strategy;
-
-import com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConstants;
-import com.tzld.piaoquan.recommend.server.service.recall.RecallParam;
-import org.apache.commons.lang3.tuple.Pair;
-import org.springframework.stereotype.Service;
-
-import static com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConstants.KEY_QUICK_WITH_SCORE_FORMAT;
-
-/**
- * @author dyp
- */
-@Service
-public class QuickFlowPoolWithScoreRecallStrategy extends AbstractFlowPoolWithScoreRecallStrategy {
-
-    @Override
-    String flowPoolKey(RecallParam param) {
-        if (param.isFlowPoolUpgrade()) {
-            return FlowPoolConstants.KEY_QUICK_WITH_SCORE_FORMAT_V2;
-        }
-        return String.format(KEY_QUICK_WITH_SCORE_FORMAT, param.getAppType());
-    }
-
-    @Override
-    public String pushFrom() {
-        return FlowPoolConstants.QUICK_PUSH_FORM;
-    }
-}