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

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

@@ -14,7 +14,6 @@ import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
 
 import java.util.Collections;
-import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
@@ -28,6 +27,8 @@ public class FlowPoolConfigService {
     @Autowired
     @Qualifier("redisTemplate")
     private RedisTemplate<String, String> redisTemplate;
+    @Value("${flow.pool.upgrade.switch:true}")
+    private boolean flowPoolUpgradeSwitch;
 
     private LoadingCache<String, Map<String, Double>> levelWeightConfigCache = CacheBuilder.newBuilder()
             .maximumSize(10)
@@ -49,7 +50,9 @@ public class FlowPoolConfigService {
 
     public Map<String, Double> getLevelWeight4FlowPoolWithLevel() {
         try {
-            return levelWeightConfigCache.get("flow:pool:level:weight:v2");
+            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();

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

@@ -7,7 +7,7 @@ public class FlowPoolConstants {
     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_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";
 
 }

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

@@ -15,6 +15,7 @@ import java.util.*;
 import java.util.concurrent.ExecutorService;
 import java.util.stream.Collectors;
 
+import static com.tzld.piaoquan.recommend.server.common.enums.AppTypeEnum.*;
 import static com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConstants.*;
 
 /**
@@ -35,6 +36,10 @@ public class FlowPoolService {
 
     private ExecutorService pool = ThreadPoolFactory.defaultPool();
 
+    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()};
+
     public void updateDistributeCountWithLevel(List<Video> videos) {
         if (CollectionUtils.isEmpty(videos)) {
             return;
@@ -49,6 +54,10 @@ public class FlowPoolService {
         asyncDelDistributeCountV2(videoFlowPoolMap, (appType, level, values) -> {
             redisTemplate.opsForSet().remove(String.format(KEY_WITH_LEVEL_FORMAT_V2, appType, level), values);
         });
+        asyncDelDistributeCount(videoFlowPoolMap, (appType, level, values) -> {
+            String key = String.format(KEY_WITH_LEVEL_FORMAT, appType, level);
+            Long count = redisTemplate.opsForSet().remove(key, values);
+        });
     }
 
     private Map<Long, String> updateDistributeCount(List<Video> videos) {
@@ -64,6 +73,34 @@ public class FlowPoolService {
         return removeMap;
     }
 
+    private void asyncDelDistributeCount(Map<Long, String> videoFlowPoolMap,
+                                         TripleConsumer<Integer, String, String[]> flowPoolRemoveConsumer) {
+        if (MapUtils.isEmpty(videoFlowPoolMap)) {
+            return;
+        }
+        pool.execute(() -> {
+            List<String> keys = videoFlowPoolMap.entrySet().stream()
+                    .map(v -> String.format(localDistributeCountFormat, v.getKey(), v.getValue()))
+                    .collect(Collectors.toList());
+            redisTemplate.delete(keys);
+
+            Map<String, Double> levelWeight = flowPoolConfigService.getLevelWeight4FlowPoolWithLevel();
+            String[] values = new String[videoFlowPoolMap.size()];
+            int i = 0;
+            for (Map.Entry v : videoFlowPoolMap.entrySet()) {
+                values[i++] = String.format(valueFormat, v.getKey(), v.getValue());
+            }
+            // remove 每小程序每层的数据
+            // TODO 现在视频只会出现在一个层级，所以可以做个优化
+            for (String level : levelWeight.keySet()) {
+                for (int appType : appTypes) {
+                    flowPoolRemoveConsumer.accept(appType, level, values);
+                }
+            }
+        });
+    }
+
+
     private void asyncDelDistributeCountV2(Map<Long, String> videoFlowPoolMap,
                                            TripleConsumer<Integer, String, String[]> flowPoolRemoveConsumer) {
         if (MapUtils.isEmpty(videoFlowPoolMap)) {
@@ -88,3 +125,4 @@ public class FlowPoolService {
         });
     }
 }
+

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

@@ -16,6 +16,7 @@ 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.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 
 import java.math.BigDecimal;
@@ -23,6 +24,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_FORMAT;
 import static com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConstants.KEY_WITH_LEVEL_FORMAT_V2;
 
 /**
@@ -33,6 +35,8 @@ import static com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConsta
 public class FlowPoolWithLevelRecallStrategyTomson extends AbstractFlowPoolWithLevelRecallStrategy {
     @Autowired
     private FlowPoolConfigService flowPoolConfigService;
+    @Value("${flow.pool.upgrade.switch:true}")
+    private boolean flowPoolUpgradeSwitch;
 
     @Override
     Pair<String, String> flowPoolKeyAndLevel(RecallParam param) {
@@ -42,7 +46,9 @@ public class FlowPoolWithLevelRecallStrategyTomson extends AbstractFlowPoolWithL
         // 2. 判断各层级是否有视频需分发
         List<LevelWeight> availableLevels = new ArrayList<>();
         for (Map.Entry<String, Double> entry : levelWeightMap.entrySet()) {
-            String levelKey = String.format(KEY_WITH_LEVEL_FORMAT_V2, 0, entry.getKey());
+            String levelKey = flowPoolUpgradeSwitch
+                    ? 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());

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

@@ -3,6 +3,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.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 
 /**
@@ -10,10 +11,14 @@ import org.springframework.stereotype.Service;
  */
 @Service
 public class QuickFlowPoolWithLevelRecallStrategy extends AbstractFlowPoolWithLevelRecallStrategy {
+    @Value("${flow.pool.upgrade.switch:true}")
+    private boolean flowPoolUpgradeSwitch;
 
     @Override
     Pair<String, String> flowPoolKeyAndLevel(RecallParam param) {
-        return Pair.of(String.format("flow:pool:quick:item:%s:3", param.getAppType()), "");
+        return flowPoolUpgradeSwitch
+                ? Pair.of(String.format("flow:pool:quick:item:%s:3", 0), "")
+                : Pair.of(String.format("flow:pool:quick:item:%s:3", param.getAppType()), "");
     }
 
     @Override