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

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

@@ -105,6 +105,9 @@ public class RecommendService {
     @ApolloJsonValue("${risk.scene:[]}")
     private Set<Long> riskScenes;
 
+    @Value("${flow.pool.upgrade.switch:true}")
+    private boolean flowPoolUpgradeSwitch;
+
     private LoadingCache<String, Set<String>> riskUserCache = CacheBuilder.newBuilder()
             .maximumSize(5)
             .refreshAfterWrite(300, TimeUnit.SECONDS)
@@ -396,21 +399,12 @@ public class RecommendService {
         int flowPoolIdChoice = flowPoolIds.get(RandomUtils.nextInt(0, flowPoolIds.size()));
         Map<String, List<Integer>> flowPoolConfig = flowPoolConfigService.getFlowPoolConfig();
         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())) {
-            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());
-                }
+
+        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());
             }
         }
         // @desc 新的流量池分发实验组划分，每个尾号不同策略分组。 @time 20240318 @author 张博
@@ -570,6 +564,8 @@ public class RecommendService {
         recallParam.setHotSceneType(param.getHotSceneType());
         recallParam.setClientIp(param.getClientIp());
 
+        recallParam.setFlowPoolUpgrade(flowPoolUpgradeSwitch);
+
         return recallParam;
     }
 
@@ -646,9 +642,6 @@ 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/flowpool/FlowPoolConstants.java

@@ -16,8 +16,16 @@ 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_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_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

@@ -3,7 +3,6 @@ package com.tzld.piaoquan.recommend.server.service.flowpool;
 import com.tzld.piaoquan.recommend.server.model.TripleConsumer;
 import com.tzld.piaoquan.recommend.server.model.Video;
 import com.tzld.piaoquan.recommend.server.common.ThreadPoolFactory;
-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;
@@ -84,6 +83,11 @@ public class FlowPoolService {
             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) {
@@ -119,6 +123,11 @@ public class FlowPoolService {
             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) {
@@ -152,6 +161,11 @@ public class FlowPoolService {
 
             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) {

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

@@ -41,4 +41,6 @@ public class RecallParam {
 
     private boolean riskUser;
 
+    private boolean flowPoolUpgrade;
+
 }

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

@@ -184,53 +184,13 @@ public class RecallService implements ApplicationContextAware {
                 strategies.addAll(getRegionRecallStrategy(param));
         }
         // 命中用户黑名单不走流量池
-        if (!param.isRiskUser() &&  (!hitUserBlacklist || !isInBlacklist)) {
-            Set<String> abExpCodes = param.getAbExpCodes();
-            if (Objects.nonNull(abExpCodes) && abExpCodes.contains(Constant.SUPPLY_AB_CODE_ID)) {
-//            if (Objects.equals(Constant.SUPPLY_AB_CODE, abCode)) {
-                // 供给流量池策略 648 实验 random
-                strategies.add(strategyMap.get(FlowPoolWithLevelSupplyRecallStrategy.class.getSimpleName()));
-                //2:通过“流量池标记”控制“流量池召回子策略” 其中有9组会走EXPERIMENTAL_FLOW_SET_LEVEL 有1组会走EXPERIMENTAL_FLOW_SET_LEVEL_SCORE
-            } else if ("60116".equals(abCode)) {
-                int lastDigit = param.getLastDigit();
-                String lastDigitAB = lastDigitAbcode != null ? lastDigitAbcode.getOrDefault(lastDigit, "default") : "default";
-                switch (lastDigitAB) {
-                    case "random":
-                        strategies.add(strategyMap.get(FlowPoolWithLevelRecallStrategyFilterDigit.class.getSimpleName()));
-                        break;
-                    case "tomson":
-                        strategies.add(strategyMap.get(FlowPoolWithLevelRecallStrategyTomsonFilterDigitV2.class.getSimpleName()));
-                        break;
-                    case "score":
-                        strategies.add(strategyMap.get(FlowPoolWithLevelScoreRecallStrategy.class.getSimpleName()));
-                        break;
-                    default:
-                        strategies.add(strategyMap.get(FlowPoolWithLevelRecallStrategyFilterDigit.class.getSimpleName()));
-                        break;
-                }
+        if (!param.isRiskUser() && (!hitUserBlacklist || !isInBlacklist)) {
+            if (param.getFlowPoolAbtestGroup().equals(FlowPoolConstants.EXPERIMENTAL_FLOW_SET_LEVEL)) {
+                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 {
-                if (param.getFlowPoolAbtestGroup().equals(FlowPoolConstants.EXPERIMENTAL_FLOW_SET_LEVEL)) {
-                    strategies.add(strategyMap.get(QuickFlowPoolWithLevelRecallStrategy.class.getSimpleName()));
-                    if ("60126".equals(abCode) || "60125".equals(abCode) || "60124".equals(abCode)
-                            || "60105".equals(abCode) || "60106".equals(abCode) || "60107".equals(abCode)
-                            || "60112".equals(abCode) || "60113".equals(abCode) || "60114".equals(abCode)
-                            || "60115".equals(abCode) || "60117".equals(abCode) || "60118".equals(abCode)
-                            || "60119".equals(abCode) || "60150".equals(abCode) || "60151".equals(abCode)
-                            || "60654".equals(abCode) || "60655".equals(abCode) || "60656".equals(abCode)
-                            || "60121".equals(abCode)
-                    ) {
-                        strategies.add(strategyMap.get(FlowPoolWithLevelRecallStrategyTomson.class.getSimpleName()));
-                    } else {
-                        strategies.add(strategyMap.get(FlowPoolWithLevelRecallStrategy.class.getSimpleName()));
-                    }
-                } else if (param.getFlowPoolAbtestGroup().equals(FlowPoolConstants.EXPERIMENTAL_FLOW_SET_LEVEL_SCORE)) {
-                    strategies.add(strategyMap.get(QuickFlowPoolWithLevelScoreRecallStrategy.class.getSimpleName()));
-                    // 在执行中
-                    strategies.add(strategyMap.get(FlowPoolWithLevelScoreRecallStrategy.class.getSimpleName()));
-                } else {
-                    strategies.add(strategyMap.get(QuickFlowPoolWithScoreRecallStrategy.class.getSimpleName()));
-                    strategies.add(strategyMap.get(FlowPoolWithScoreRecallStrategy.class.getSimpleName()));
-                }
+                strategies.add(strategyMap.get(FlowPoolWithScoreRecallStrategy.class.getSimpleName()));
             }
         }
 

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

@@ -8,6 +8,7 @@ import org.springframework.stereotype.Service;
  * @author sunxy
  */
 @Service
+@Deprecated
 public class FlowPoolLastDayTopRecallStrategy extends AbstractNewContentVideoRecallStrategy {
 
     public static final String PUSH_FORM = "flow_pool_top_video_recall";

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

@@ -17,12 +17,14 @@ 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

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

@@ -27,6 +27,7 @@ import static com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConsta
  */
 @Service
 @Slf4j
+@Deprecated
 public class FlowPoolWithLevelRecallStrategyFilterDigit extends AbstractFlowPoolWithLevelRecallStrategy {
 
     @Autowired

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

@@ -27,6 +27,7 @@ import java.util.*;
 import java.util.stream.Collectors;
 
 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 zhangbo
@@ -41,6 +42,9 @@ 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();
 

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

@@ -30,6 +30,7 @@ import static com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConsta
  */
 @Service
 @Slf4j
+@Deprecated
 public class FlowPoolWithLevelRecallStrategyTomsonFilterDigit extends AbstractFlowPoolWithLevelRecallStrategy {
 
     @Autowired

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

@@ -30,6 +30,7 @@ import static com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConsta
  */
 @Service
 @Slf4j
+@Deprecated
 public class FlowPoolWithLevelRecallStrategyTomsonFilterDigitV2 extends AbstractFlowPoolWithLevelRecallStrategy {
 
     @Autowired

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

@@ -27,7 +27,7 @@ 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_SCORE_FORMAT;
+import static com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConstants.*;
 
 /**
  * @author dyp
@@ -41,6 +41,9 @@ 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();
 

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

@@ -19,6 +19,7 @@ import static com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConsta
 
 @Service
 @Slf4j
+@Deprecated
 public class FlowPoolWithLevelSupplyRecallStrategy extends AbstractFlowPoolWithLevelRecallStrategy {
 
     @Autowired

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

@@ -2,8 +2,10 @@ 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;
 
 /**
@@ -15,6 +17,9 @@ public class FlowPoolWithScoreRecallStrategy extends AbstractFlowPoolWithScoreRe
 
     @Override
     String flowPoolKey(RecallParam param) {
+        if (param.isFlowPoolUpgrade()) {
+            return FlowPoolConstants.KEY_WITH_SCORE_FORMAT_V2;
+        }
         return String.format(KEY_WITH_SCORE_FORMAT, param.getAppType());
     }
 

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

@@ -13,6 +13,9 @@ public class QuickFlowPoolWithLevelRecallStrategy extends AbstractFlowPoolWithLe
 
     @Override
     Pair<String, String> flowPoolKeyAndLevel(RecallParam param) {
+        if (param.isFlowPoolUpgrade()) {
+            return Pair.of(FlowPoolConstants.KEY_QUICK_WITH_LEVEL_FORMAT_V2, "-1");
+        }
         return Pair.of(String.format("flow:pool:quick:item:%s:3", param.getAppType()), "");
     }
 

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

@@ -17,6 +17,9 @@ public class QuickFlowPoolWithLevelScoreRecallStrategy extends AbstractFlowPoolW
 
     @Override
     Pair<String, String> flowPoolKeyAndLevel(RecallParam param) {
+        if (param.isFlowPoolUpgrade()) {
+            return Pair.of(FlowPoolConstants.KEY_QUICK_WITH_LEVEL_SCORE_FORMAT_V2, "-1");
+        }
         return Pair.of(String.format(KEY_QUICK_WITH_LEVEL_SCORE_FORMAT, param.getAppType()), "");
     }
 

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

@@ -2,6 +2,7 @@ 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;
@@ -14,6 +15,9 @@ public class QuickFlowPoolWithScoreRecallStrategy extends AbstractFlowPoolWithSc
 
     @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());
     }