Merge remote-tracking branch 'origin/feature_20240507_supeng_supply_ab' into pre-master

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

@@ -15,4 +15,8 @@ public class Constant {
      * 流量池头部视频redis key
      */
     public static final String VIDEO_PERFORMANCE_DATA_REDIS_KEY = "video_performance_data_redis_key:";
+    /**
+     * 供给流量池实验 648 random
+     */
+    public static final String SUPPLY_AB_CODE = "60600";
 }

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

@@ -4,6 +4,7 @@ import com.ctrip.framework.apollo.spring.annotation.ApolloJsonValue;
 import com.google.common.base.Stopwatch;
 import com.google.common.base.Strings;
 import com.google.common.reflect.TypeToken;
+import com.tzld.piaoquan.recommend.server.common.base.Constant;
 import com.tzld.piaoquan.recommend.server.common.enums.AppTypeEnum;
 import com.tzld.piaoquan.recommend.server.gen.common.Result;
 import com.tzld.piaoquan.recommend.server.gen.recommend.MachineInfoProto;
@@ -100,8 +101,6 @@ public class RecommendService {
         ab_initial_config_map.put(AppTypeEnum.VLOG.getCode(), "095");
         ab_initial_config_map.put(AppTypeEnum.LOVE_LIVE.getCode(), "144");
         ab_initial_config_map.put(AppTypeEnum.LONG_VIDEO.getCode(), "121");
-        ab_initial_config_map.put(AppTypeEnum.LAO_HAO_KAN_VIDEO.getCode(), "074");
-        ab_initial_config_map.put(AppTypeEnum.ZUI_JING_QI.getCode(), "069");
     }
 
     public RecommendResponse homepageRecommend(RecommendRequest request) {
@@ -340,12 +339,21 @@ public class RecommendService {
 
         // 流量池分发实验组划分
         int flowPoolIdChoice = flowPoolIds.get(RandomUtils.nextInt(0, flowPoolIds.size()));
-        param.setFlowPoolId(flowPoolIdChoice);
-        param.setFlowPoolAbtestGroup("control_group");
         Map<String, List<Integer>> flowPoolConfig = flowPoolConfigService.getFlowPoolConfig();
-        for (Map.Entry<String, List<Integer>> entry : flowPoolConfig.entrySet()) {
-            if (entry.getValue().contains(flowPoolIdChoice)) {
-                param.setFlowPoolAbtestGroup(entry.getKey());
+        if (Objects.equals(Constant.SUPPLY_AB_CODE, param.getAbCode())) {
+            List<Integer> supplyFlowPoolIdList = flowPoolConfig.get(FlowPoolConstants.SUPPLY_FLOW_SET_LEVEL);
+            if (Objects.nonNull(supplyFlowPoolIdList) && !supplyFlowPoolIdList.isEmpty()) {
+                flowPoolIdChoice = supplyFlowPoolIdList.get(0);
+                param.setFlowPoolId(flowPoolIdChoice);
+                param.setFlowPoolAbtestGroup(FlowPoolConstants.SUPPLY_FLOW_SET_LEVEL);
+            }
+        } else {
+            param.setFlowPoolId(flowPoolIdChoice);
+            param.setFlowPoolAbtestGroup("control_group");
+            for (Map.Entry<String, List<Integer>> entry : flowPoolConfig.entrySet()) {
+                if (entry.getValue().contains(flowPoolIdChoice)) {
+                    param.setFlowPoolAbtestGroup(entry.getKey());
+                }
             }
         }
 
@@ -647,6 +655,9 @@ public class RecommendService {
         }
 
         switch (param.getFlowPoolAbtestGroup()) {
+            case FlowPoolConstants.SUPPLY_FLOW_SET_LEVEL:
+                flowPoolService.updateSupplyDistributeCountWithLevel(flowPoolVideos);
+                break;
             case FlowPoolConstants.EXPERIMENTAL_FLOW_SET_LEVEL:
                 flowPoolService.updateDistributeCountWithLevel(flowPoolVideos);
                 break;

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

@@ -1,5 +1,6 @@
 package com.tzld.piaoquan.recommend.server.service.filter;
 
+import com.tzld.piaoquan.recommend.server.common.base.Constant;
 import com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolService;
 import com.tzld.piaoquan.recommend.server.util.JSONUtils;
 import lombok.extern.slf4j.Slf4j;
@@ -9,9 +10,7 @@ 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.*;
 import java.util.stream.Collectors;
 
 /**
@@ -43,7 +42,15 @@ public class FlowPoolWithLevelFilterService extends AbstractFilterService {
                         v -> v,
                         v -> param.getFlowPoolMap().get(v)));
 
-        Map<Long, Integer> distributeCountMap = flowPoolService.getDistributeCountWithLevel(flowPoolMap);
+        Map<Long, Integer> distributeCountMap;
+        //供给流量池实验
+        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);
 
         List<Long> remainVideoIds = new ArrayList<>();
         for (Long videoId : videoIds) {

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

@@ -68,8 +68,6 @@ public class AllowListStrategy implements FilterStrategy {
         //不是新小程序 在白名单则不显示
         Set<Long> retainVideoIds = new LinkedHashSet<>();
         if (param.getAppType() != AppTypeEnum.WAN_NENG_VIDEO.getCode()
-                && param.getAppType() != AppTypeEnum.LAO_HAO_KAN_VIDEO.getCode()
-                && param.getAppType() != AppTypeEnum.ZUI_JING_QI.getCode()
                 && param.getAppType() != AppTypeEnum.H5.getCode()) {
 
             Iterator<Long> iterator = param.getVideoIds().iterator();
@@ -81,7 +79,6 @@ public class AllowListStrategy implements FilterStrategy {
                 }
             }
         } else if (param.getAppType() == AppTypeEnum.WAN_NENG_VIDEO.getCode()
-                || param.getAppType() == AppTypeEnum.ZUI_JING_QI.getCode()
                 || param.getAppType() == AppTypeEnum.H5.getCode()) {
             Iterator<Long> iterator = param.getVideoIds().iterator();
             while (iterator.hasNext()) {

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

@@ -6,11 +6,13 @@ 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";
 
     public static final String KEY_WITH_LEVEL_FORMAT = "flow:pool:level:item:%s:%s";
+    public static final String KEY_WITH_LEVEL_SUPPLY_FORMAT = "flow:pool:level:item:supply:%s:%s";
     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";

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

@@ -33,6 +33,11 @@ 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";
 
@@ -205,4 +210,71 @@ 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)) {
+            return;
+        }
+        Map<Long, String> removeMap = updateSupplyDistributeCount(videos);
+
+        asyncDelSupplyDistributeCountWithLevel(removeMap);
+
+    }
+
+    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);
+        });
+    }
 }

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

@@ -186,9 +186,23 @@ public class RankService {
                 Collections.sort(rovRecallRank, Comparator.comparingDouble(o -> -o.getSortScore()));
             }
         } else {
+            rovRecallRank.addAll(extractAndSort(param, RegionRealtimeRecallStrategyV1.PUSH_FORM));
             rovRecallRank.addAll(extractAndSort(param, SimHotVideoRecallStrategy.PUSH_FORM));
             rovRecallRank.addAll(extractAndSort(param, ReturnVideoRecallStrategy.PUSH_FORM));
             removeDuplicate(rovRecallRank);
+
+            // 融合排序
+            List<String> videoIdKeys = rovRecallRank.stream()
+                    .map(t -> param.getRankKeyPrefix() + t.getVideoId())
+                    .collect(Collectors.toList());
+            List<String> videoScores = redisTemplate.opsForValue().multiGet(videoIdKeys);
+            if (CollectionUtils.isNotEmpty(videoScores)
+                    && videoScores.size() == rovRecallRank.size()) {
+                for (int i = 0; i < videoScores.size(); i++) {
+                    rovRecallRank.get(i).setSortScore(NumberUtils.toDouble(videoScores.get(i), 0.0));
+                }
+                rovRecallRank.sort(Comparator.comparingDouble(o -> -o.getSortScore()));
+            }
         }
         return rovRecallRank;
     }
@@ -267,23 +281,11 @@ public class RankService {
     }
 
     public List<Video> mergeAndRankFlowPoolRecall(RankParam param) {
-        if (param.getAppType() == AppTypeEnum.LAO_HAO_KAN_VIDEO.getCode()
-                || param.getAppType() == AppTypeEnum.ZUI_JING_QI.getCode()) {
-            if (param.getAbCode().equals("60054")
-                    || param.getAbCode().equals("60068")
-                    || param.getAbCode().equals("60081")
-                    || param.getAbCode().equals("60084")) {
-                return extractAndSort(param, FlowPoolConstants.QUICK_PUSH_FORM);
-            } else {
-                return Collections.emptyList();
-            }
+        List<Video> quickFlowPoolVideos = extractAndSort(param, FlowPoolConstants.QUICK_PUSH_FORM);
+        if (CollectionUtils.isNotEmpty(quickFlowPoolVideos)) {
+            return quickFlowPoolVideos;
         } else {
-            List<Video> quickFlowPoolVideos = extractAndSort(param, FlowPoolConstants.QUICK_PUSH_FORM);
-            if (CollectionUtils.isNotEmpty(quickFlowPoolVideos)) {
-                return quickFlowPoolVideos;
-            } else {
-                return extractAndSort(param, FlowPoolConstants.PUSH_FORM);
-            }
+            return extractAndSort(param, FlowPoolConstants.PUSH_FORM);
         }
     }
 
@@ -307,17 +309,12 @@ public class RankService {
     }
 
     public double getFlowPoolP(RankParam param) {
-        if (param.getAppType() == AppTypeEnum.LAO_HAO_KAN_VIDEO.getCode()
-                || param.getAppType() == AppTypeEnum.ZUI_JING_QI.getCode()) {
-            return param.getFlowPoolP();
+        List<Video> quickFlowPoolVideos = extractAndSort(param, FlowPoolConstants.QUICK_PUSH_FORM);
+        if (CollectionUtils.isNotEmpty(quickFlowPoolVideos)) {
+            String quick_flow_pool_P = redisTemplate.opsForValue().get("flow:pool:quick:distribute:rate:3");
+            return NumberUtils.toDouble(quick_flow_pool_P, 0);
         } else {
-            List<Video> quickFlowPoolVideos = extractAndSort(param, FlowPoolConstants.QUICK_PUSH_FORM);
-            if (CollectionUtils.isNotEmpty(quickFlowPoolVideos)) {
-                String quick_flow_pool_P = redisTemplate.opsForValue().get("flow:pool:quick:distribute:rate:3");
-                return NumberUtils.toDouble(quick_flow_pool_P, 0);
-            } else {
-                return param.getFlowPoolP();
-            }
+            return param.getFlowPoolP();
         }
     }
 

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

@@ -1,6 +1,7 @@
 package com.tzld.piaoquan.recommend.server.service.recall;
 import com.ctrip.framework.apollo.spring.annotation.ApolloJsonValue;
 import com.tzld.piaoquan.recommend.server.common.ThreadPoolFactory;
+import com.tzld.piaoquan.recommend.server.common.base.Constant;
 import com.tzld.piaoquan.recommend.server.common.enums.AppTypeEnum;
 import com.tzld.piaoquan.recommend.server.model.Video;
 import com.tzld.piaoquan.recommend.server.service.filter.strategy.BlacklistContainer;
@@ -57,8 +58,10 @@ public class RecallService implements ApplicationContextAware {
         CountDownLatch cdl = new CountDownLatch(strategies.size());
         List<Future<RecallResult.RecallData>> recallResultFutures = new ArrayList<>();
         for (final RecallStrategy strategy : strategies) {
+            log.info("supply flowpool strategy = {}", strategy.getClass().toString());
             Future<RecallResult.RecallData> future = pool.submit(() -> {
                 List<Video> result = strategy.recall(param);
+                log.info("supply flowpool strategy = {}, result = {}", strategy.getClass().toString(), result);
                 cdl.countDown();
                 return new RecallResult.RecallData(strategy.pushFrom(), result);
             });
@@ -97,11 +100,8 @@ public class RecallService implements ApplicationContextAware {
         boolean isInBlacklist = CollectionUtils.isNotEmpty(blacklistAppTypeSet) && blacklistAppTypeSet.contains(param.getAppType());
 
         String abCode = param.getAbCode();
-        //1：通过“产品”控制“召回子策略”
-        if (param.getAppType() == AppTypeEnum.LAO_HAO_KAN_VIDEO.getCode()
-                || param.getAppType() == AppTypeEnum.ZUI_JING_QI.getCode()) {
-            strategies.addAll(getRegionRecallStrategy(param));
-        } else if (param.getAppType() == AppTypeEnum.PIAO_QUAN_MEIHAO_ZHUFU.getCode() && Objects.equals(param.getCategoryId(), 100)) {
+        //1：通过“产品”控制“召回子策略”. 票圈美好祝福与内部tab只走祝福召回。APP只走固定列表。
+        if (param.getAppType() == AppTypeEnum.PIAO_QUAN_MEIHAO_ZHUFU.getCode() && Objects.equals(param.getCategoryId(), 100)) {
             strategies.add(strategyMap.get(BlessRecallStrategy.class.getSimpleName()));
             return strategies;
         } else if (param.getAppType() == AppTypeEnum.APP_SPEED.getCode()) {
@@ -175,12 +175,17 @@ public class RecallService implements ApplicationContextAware {
                     strategies.add(strategyMap.get(TitleTagRecallStrategyV1.class.getSimpleName()));
                     strategies.addAll(getRegionRecallStrategy(param));
                 default:
+                    strategies.add(strategyMap.get(RegionRealtimeRecallStrategyV1.class.getSimpleName()));
                     strategies.addAll(getRegionRecallStrategy(param));
             }
             // 命中用户黑名单不走流量池
             if (!hitUserBlacklist || !isInBlacklist) {
+                if (Objects.equals(Constant.SUPPLY_AB_CODE, abCode)) {
+                    // 供给流量池策略 648 实验 random
+                    strategies.add(strategyMap.get(FlowPoolWithLevelSupplyRecallStrategy.class.getSimpleName()));
+                    log.info("supply flowpool {}", abCode);
                 //2:通过“流量池标记”控制“流量池召回子策略” 其中有9组会走EXPERIMENTAL_FLOW_SET_LEVEL 有1组会走EXPERIMENTAL_FLOW_SET_LEVEL_SCORE
-                if ("60116".equals(abCode)) {
+                } else if ("60116".equals(abCode)) {
                     int lastDigit = param.getLastDigit();
                     String lastDigitAB = lastDigitAbcode != null ? lastDigitAbcode.getOrDefault(lastDigit, "default") : "default";
                     switch (lastDigitAB) {
@@ -244,84 +249,81 @@ public class RecallService implements ApplicationContextAware {
         }
 
         //3:通过“abcode”控制“召回子策略”
-
-        if (abCode == null) {
-            // todo 做兜底吗？
-        } else {
-            switch (abCode) {
-                case "60096":
-                    strategies.add(strategyMap.get(RegionHWithoutDupRecallStrategy.class.getSimpleName()));
-                    strategies.add(strategyMap.get(SimHotVideoRecallStrategy.class.getSimpleName()));
-                    strategies.add(strategyMap.get(ReturnVideoRecallStrategy.class.getSimpleName()));
-                    break;
-                case "60111": // 561
-                case "60068":
-                case "60092":
-                case "60094":
-                case "60097": // 523
-                case "60098": // 533
-                case "60101": // 排序str实验
-                case "60102": // 通过更改param中的配置实现使用不同数据源 data66 rule68 + 无排序模块
-                case "60103": // 增加地域1小时扩量，通过配置实现
-                case "60120": // 576
-                    strategies.add(strategyMap.get(SimHotVideoRecallStrategy.class.getSimpleName()));
-                    strategies.add(strategyMap.get(ReturnVideoRecallStrategy.class.getSimpleName()));
-                    break;
-                case "60112": // 562
-                    strategies.add(strategyMap.get(SimHotVideoRecallStrategy.class.getSimpleName()));
-                    strategies.add(strategyMap.get(ReturnVideoRecallStrategy.class.getSimpleName()));
-                    strategies.add(strategyMap.get(RegionRealtimeRecallStrategyV5Hand.class.getSimpleName()));
-                    break;
-                case "60121": // 536
-                case "60122": // 537
-                case "60124": // 546
-                case "60125": // 547
-                case "60123": // 541
-                case "60126": // 548
-                case "60105": // 551
-                case "60106": // 552
-                case "60107": // 553
-                case "60116": // 566
-                case "60119": // 569
-                    strategies.add(strategyMap.get(SimHotVideoRecallStrategy.class.getSimpleName()));
-                    strategies.add(strategyMap.get(ReturnVideoRecallStrategy.class.getSimpleName()));
-                    strategies.add(strategyMap.get(FestivalRecallStrategyV1.class.getSimpleName()));
-                    break;
-                case "60113": // 563
-                case "60114": // 564
-                case "60115": // 565
-                case "60117": // 567
-                case "60118": // 568
-                case "60150": // 645
-                    if (!hitUserBlacklist || !isInBlacklist) {
-                        strategies.add(strategyMap.get(FlowPoolLastDayTopRecallStrategy.class.getSimpleName()));
-                    }
-                    strategies.add(strategyMap.get(SimHotVideoRecallStrategy.class.getSimpleName()));
-                    strategies.add(strategyMap.get(ReturnVideoRecallStrategy.class.getSimpleName()));
-                    strategies.add(strategyMap.get(FestivalRecallStrategyV1.class.getSimpleName()));
-                    break;
-                case "60104": // 去掉sim的对比实验
-                    strategies.add(strategyMap.get(ReturnVideoRecallStrategy.class.getSimpleName()));
-                    break;
-                case "60110": // 新内容的召回（流量池的Top内容）
-                    strategies.add(strategyMap.get(TopGoodPerformanceVideoRecallStrategy.class.getSimpleName()));
-                    if (!hitUserBlacklist || !isInBlacklist) {
-                        strategies.add(strategyMap.get(FlowPoolLastDayTopRecallStrategy.class.getSimpleName()));
-                    }
-                    strategies.add(strategyMap.get(SimHotVideoRecallStrategy.class.getSimpleName()));
-                    strategies.add(strategyMap.get(ReturnVideoRecallStrategy.class.getSimpleName()));
-                    break;
-                case "60130":
-                case "60131":
-                case "60132":
-                    strategies.add(strategyMap.get(SimHotVideoRecallStrategy.class.getSimpleName()));
-                    strategies.add(strategyMap.get(ReturnVideoRecallStrategy.class.getSimpleName()));
-                    strategies.add(strategyMap.get(FestivalRecallStrategyV1.class.getSimpleName()));
-                    break;
-                default:
-                    // todo 做兜底吗？
-                    break;
-            }
+        switch (abCode) {
+            case "60096":
+                strategies.add(strategyMap.get(RegionHWithoutDupRecallStrategy.class.getSimpleName()));
+                strategies.add(strategyMap.get(SimHotVideoRecallStrategy.class.getSimpleName()));
+                strategies.add(strategyMap.get(ReturnVideoRecallStrategy.class.getSimpleName()));
+                break;
+            case "60111": // 561
+            case "60068":
+            case "60092":
+            case "60094":
+            case "60097": // 523
+            case "60098": // 533
+            case "60101": // 排序str实验
+            case "60102": // 通过更改param中的配置实现使用不同数据源 data66 rule68 + 无排序模块
+            case "60103": // 增加地域1小时扩量，通过配置实现
+            case "60120": // 576
+                strategies.add(strategyMap.get(SimHotVideoRecallStrategy.class.getSimpleName()));
+                strategies.add(strategyMap.get(ReturnVideoRecallStrategy.class.getSimpleName()));
+                break;
+            case "60112": // 562
+                strategies.add(strategyMap.get(SimHotVideoRecallStrategy.class.getSimpleName()));
+                strategies.add(strategyMap.get(ReturnVideoRecallStrategy.class.getSimpleName()));
+                strategies.add(strategyMap.get(RegionRealtimeRecallStrategyV5Hand.class.getSimpleName()));
+                break;
+            case "60121": // 536
+            case "60122": // 537
+            case "60124": // 546
+            case "60125": // 547
+            case "60123": // 541
+            case "60126": // 548
+            case "60105": // 551
+            case "60106": // 552
+            case "60107": // 553
+            case "60116": // 566
+            case "60119": // 569
+                strategies.add(strategyMap.get(SimHotVideoRecallStrategy.class.getSimpleName()));
+                strategies.add(strategyMap.get(ReturnVideoRecallStrategy.class.getSimpleName()));
+                strategies.add(strategyMap.get(FestivalRecallStrategyV1.class.getSimpleName()));
+                break;
+            case "60113": // 563
+            case "60114": // 564
+            case "60115": // 565
+            case "60117": // 567
+            case "60118": // 568
+            case "60150": // 645
+                if (!hitUserBlacklist || !isInBlacklist) {
+                    strategies.add(strategyMap.get(FlowPoolLastDayTopRecallStrategy.class.getSimpleName()));
+                }
+                strategies.add(strategyMap.get(SimHotVideoRecallStrategy.class.getSimpleName()));
+                strategies.add(strategyMap.get(ReturnVideoRecallStrategy.class.getSimpleName()));
+                strategies.add(strategyMap.get(FestivalRecallStrategyV1.class.getSimpleName()));
+                break;
+            case "60104": // 去掉sim的对比实验
+                strategies.add(strategyMap.get(ReturnVideoRecallStrategy.class.getSimpleName()));
+                break;
+            case "60110": // 新内容的召回（流量池的Top内容）
+                strategies.add(strategyMap.get(TopGoodPerformanceVideoRecallStrategy.class.getSimpleName()));
+                if (!hitUserBlacklist || !isInBlacklist) {
+                    strategies.add(strategyMap.get(FlowPoolLastDayTopRecallStrategy.class.getSimpleName()));
+                }
+                strategies.add(strategyMap.get(SimHotVideoRecallStrategy.class.getSimpleName()));
+                strategies.add(strategyMap.get(ReturnVideoRecallStrategy.class.getSimpleName()));
+                break;
+            case "60130":
+            case "60131":
+            case "60132":
+                strategies.add(strategyMap.get(SimHotVideoRecallStrategy.class.getSimpleName()));
+                strategies.add(strategyMap.get(ReturnVideoRecallStrategy.class.getSimpleName()));
+                strategies.add(strategyMap.get(FestivalRecallStrategyV1.class.getSimpleName()));
+                break;
+            default:
+                strategies.add(strategyMap.get(SimHotVideoRecallStrategy.class.getSimpleName()));
+                strategies.add(strategyMap.get(ReturnVideoRecallStrategy.class.getSimpleName()));
+                strategies.add(strategyMap.get(FestivalRecallStrategyV1.class.getSimpleName()));
+                break;
         }
 
         return strategies;

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

@@ -6,6 +6,7 @@ import com.tzld.piaoquan.recommend.server.service.filter.FlowPoolWithLevelFilter
 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 lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang3.RandomUtils;
 import org.apache.commons.lang3.math.NumberUtils;
@@ -22,6 +23,7 @@ import java.util.Map;
 /**
  * @author dyp
  */
+@Slf4j
 public abstract class AbstractFlowPoolWithLevelRecallStrategy implements RecallStrategy {
     @Autowired
     @Qualifier("redisTemplate")
@@ -38,6 +40,7 @@ public abstract class AbstractFlowPoolWithLevelRecallStrategy implements RecallS
         int getSize = param.getSize() * 5;
         List<Video> results = new ArrayList<>();
         List<String> data = redisTemplate.opsForSet().randomMembers(flowPoolKey, getSize);
+        log.info("supply flowpool flowPoolKey = {} level = {} data = {}", flowPoolKey, level, data);
         if (CollectionUtils.isEmpty(data)) {
             return null;
         }
@@ -46,9 +49,9 @@ public abstract class AbstractFlowPoolWithLevelRecallStrategy implements RecallS
             String[] values = value.split("-");
             videoFlowPoolMap.put(NumberUtils.toLong(values[0], 0), values[1]);
         }
-
+        log.info("supply flowpool videoFlowPoolMap = {}", videoFlowPoolMap);
         FilterResult filterResult = filterService.filter(FilterParamFactory.create(param, videoFlowPoolMap));
-
+        log.info("supply flowpool filterResult = {}", filterResult);
         if (filterResult != null && CollectionUtils.isNotEmpty(filterResult.getVideoIds())) {
             filterResult.getVideoIds().stream().forEach(vid -> {
                 Video recallData = new Video();

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

@@ -0,0 +1,102 @@
+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
+public class FlowPoolWithLevelSupplyRecallStrategy extends AbstractFlowPoolWithLevelRecallStrategy {
+
+    @Autowired
+    private FlowPoolConfigService flowPoolConfigService;
+
+    @Override
+    Pair<String, String> flowPoolKeyAndLevel(RecallParam param) {
+        //# 1. 获取流量池各层级分发概率权重
+        Map<String, Double> levelWeightMap = flowPoolConfigService.getLevelWeight();
+        log.info("supply flowpool levelWeightMap = {}", levelWeightMap);
+        // 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());
+            log.info("supply flowpool levelKey = {}", levelKey);
+            if (redisTemplate.hasKey(levelKey)) {
+                LevelWeight lw = new LevelWeight();
+                lw.setLevel(entry.getKey());
+                lw.setLevelKey(levelKey);
+                lw.setWeight(entry.getValue());
+                availableLevels.add(lw);
+            }
+        }
+        log.info("supply flowpool availableLevels = {}", availableLevels);
+        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()) {
+                log.info("supply flowpool result {} {}", entry.getKey(), entry.getValue());
+                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;
+    }
+}