add supply flowpool local distribute count

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

@@ -647,6 +647,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

@@ -9,9 +9,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;
 
 /**
@@ -23,6 +21,11 @@ public class FlowPoolWithLevelFilterService extends AbstractFilterService {
     @Autowired
     private FlowPoolService flowPoolService;
 
+    /**
+     * 供给流量池实验 648 random
+     */
+    private static final String SUPPLY_AB_CODE = "60600";
+
     public FilterResult filter(FilterParam param) {
         List<Long> videoIds = viewFilter(param);
 
@@ -43,7 +46,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(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/flowpool/FlowPoolConstants.java

@@ -6,6 +6,7 @@ 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";

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);
+        });
+    }
 }