revert flowpool logic

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

@@ -9,12 +9,10 @@ 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;
 
 import java.util.Collections;
-import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
@@ -49,7 +47,8 @@ public class FlowPoolConfigService {
 
     public Map<String, Double> getLevelWeight4FlowPoolWithLevel() {
         try {
-            return levelWeightConfigCache.get("flow:pool:level:weight:v2");
+            // return levelWeightConfigCache.get("flow:pool:level:weight:v2");
+            return 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,6 +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

@@ -6,8 +6,10 @@ import com.tzld.piaoquan.recommend.server.common.ThreadPoolFactory;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections4.CollectionUtils;
 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;
 
@@ -15,6 +17,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 +38,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;
@@ -46,6 +53,15 @@ public class FlowPoolService {
     }
 
     private void asyncDelDistributeCountWithLevel(Map<Long, String> videoFlowPoolMap) {
+        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);
+        });
+
+
         asyncDelDistributeCountV2(videoFlowPoolMap, (appType, level, values) -> {
             redisTemplate.opsForSet().remove(String.format(KEY_WITH_LEVEL_FORMAT_V2, appType, level), values);
         });
@@ -64,6 +80,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 +132,4 @@ public class FlowPoolService {
         });
     }
 }
+

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

@@ -23,6 +23,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;
 
 /**
@@ -42,7 +43,8 @@ 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 = String.format(KEY_WITH_LEVEL_FORMAT_V2, 0, entry.getKey());
+            String levelKey = String.format(KEY_WITH_LEVEL_FORMAT, param.getAppType(), entry.getKey());
             if (redisTemplate.hasKey(levelKey)) {
                 LevelWeight lw = new LevelWeight();
                 lw.setLevel(entry.getKey());