homepage recommend

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

@@ -11,7 +11,7 @@ import java.util.Map;
  * @author dyp
  */
 @Service
-public class FlowPoolFilterService {
+public class FlowPoolWithLevelFilterService {
     @Autowired
     private RedisTemplate<String, String> redisTemplate;
     @Autowired

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

@@ -0,0 +1,80 @@
+package com.tzld.piaoquan.recommend.server.service.filter;
+
+import com.tzld.piaoquan.recommend.server.service.FlowPoolConfigService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Service;
+
+import java.util.Map;
+
+/**
+ * @author dyp
+ */
+@Service
+public class FlowPoolWithLevelScoreFilterService {
+    @Autowired
+    private RedisTemplate<String, String> redisTemplate;
+    @Autowired
+    private FlowPoolConfigService flowPoolConfigService;
+
+    @Autowired
+    private FilterService filterService;
+
+    public FilterResult filter(FilterParam param) {
+        FilterResult result = filterService.filter(param);
+        result.getVideoIds();
+        Map<String, Double> levelWeightMap = flowPoolConfigService.getLevelWeight();
+
+       //     def check_video_counts_new_with_level_score(self, video_ids, flow_pool_mapping):
+        //        """
+        //        检查视频剩余可分发数
+        //        :param video_ids: 视频id type-list
+        //        :param flow_pool_mapping: 视频id-流量池标记mapping, type-dict
+        //        :return: check_result, error_flag
+        //        """
+        //        # flow_pool_key = self.get_pool_redis_key('flow')
+        //        # videos = []
+        //        # level_weight = self.redis_helper.get_data_from_redis(key_name=config_.FLOWPOOL_LEVEL_WEIGHT_KEY_NAME)
+        //        # level_list = [level for level in json.loads(level_weight)]
+        //        if self.level_weight is None:
+        //            level_weight = {'1': 1, '2': 1, '3': 1, '4': 1, '5': 1, '6': 1}
+        //        else:
+        //            level_weight = self.level_weight
+        //        level_list = [level for level in level_weight]
+        //
+        //        check_result = []
+        //        for video_id in video_ids:
+        //            video_id = int(video_id)
+        //            for flow_pool in flow_pool_mapping.get(video_id, []):
+        //                # 判断是否有本地分发记录
+        //                cur_count = get_videos_local_distribute_count(video_id=video_id, flow_pool=flow_pool)
+        //                # 无记录
+        //                if cur_count is None:
+        //                    # videos.append({'videoId': video_id, 'flowPool': flow_pool})
+        //                    continue
+        //                # 本地分发数 cur_count > 0
+        //                elif cur_count > 0:
+        //                    check_result.append((video_id, flow_pool))
+        //                # 本地分发数 cur_count <= 0，从所有的流量召回池移除，删除本地分发记录key
+        //                else:
+        //                    add_remove_log = False
+        //                    remain_count_key = f"{config_.LOCAL_DISTRIBUTE_COUNT_PREFIX}{video_id}:{flow_pool}"
+        //                    self.redis_helper.del_keys(remain_count_key)
+        //                    value = '{}-{}'.format(video_id, flow_pool)
+        //                    for item in config_.APP_TYPE:
+        //                        for level in level_list:
+        //                            flow_pool_key = f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL_SCORE}{config_.APP_TYPE.get(item)}:{level}"
+        //                            remove_res = self.redis_helper.remove_value_from_zset(key_name=flow_pool_key, value=value)
+        //                            if remove_res > 0:
+        //                                add_remove_log = True
+        //                        quick_flow_pool_key = f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX}{config_.APP_TYPE.get(item)}" \
+        //                                              f":{config_.QUICK_FLOW_POOL_ID}"
+        //                        remove_res = self.redis_helper.remove_value_from_zset(key_name=quick_flow_pool_key, value=value)
+        //                        if remove_res > 0:
+        //                            add_remove_log = True
+        //                    if add_remove_log is True:
+        //                        log_.info({'tag': 'remove video_id from flow_pool', 'video_id': video_id, 'flow_pool': flow_pool})
+        //        return check_result
+        return null;
+    }
+}

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

@@ -0,0 +1,80 @@
+package com.tzld.piaoquan.recommend.server.service.filter;
+
+import com.tzld.piaoquan.recommend.server.service.FlowPoolConfigService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Service;
+
+import java.util.Map;
+
+/**
+ * @author dyp
+ */
+@Service
+public class FlowPoolWithScoreFilterService {
+    @Autowired
+    private RedisTemplate<String, String> redisTemplate;
+    @Autowired
+    private FlowPoolConfigService flowPoolConfigService;
+
+    @Autowired
+    private FilterService filterService;
+
+    public FilterResult filter(FilterParam param) {
+        FilterResult result = filterService.filter(param);
+        result.getVideoIds();
+        Map<String, Double> levelWeightMap = flowPoolConfigService.getLevelWeight();
+
+       //     def check_video_counts_new_with_level_score(self, video_ids, flow_pool_mapping):
+        //        """
+        //        检查视频剩余可分发数
+        //        :param video_ids: 视频id type-list
+        //        :param flow_pool_mapping: 视频id-流量池标记mapping, type-dict
+        //        :return: check_result, error_flag
+        //        """
+        //        # flow_pool_key = self.get_pool_redis_key('flow')
+        //        # videos = []
+        //        # level_weight = self.redis_helper.get_data_from_redis(key_name=config_.FLOWPOOL_LEVEL_WEIGHT_KEY_NAME)
+        //        # level_list = [level for level in json.loads(level_weight)]
+        //        if self.level_weight is None:
+        //            level_weight = {'1': 1, '2': 1, '3': 1, '4': 1, '5': 1, '6': 1}
+        //        else:
+        //            level_weight = self.level_weight
+        //        level_list = [level for level in level_weight]
+        //
+        //        check_result = []
+        //        for video_id in video_ids:
+        //            video_id = int(video_id)
+        //            for flow_pool in flow_pool_mapping.get(video_id, []):
+        //                # 判断是否有本地分发记录
+        //                cur_count = get_videos_local_distribute_count(video_id=video_id, flow_pool=flow_pool)
+        //                # 无记录
+        //                if cur_count is None:
+        //                    # videos.append({'videoId': video_id, 'flowPool': flow_pool})
+        //                    continue
+        //                # 本地分发数 cur_count > 0
+        //                elif cur_count > 0:
+        //                    check_result.append((video_id, flow_pool))
+        //                # 本地分发数 cur_count <= 0，从所有的流量召回池移除，删除本地分发记录key
+        //                else:
+        //                    add_remove_log = False
+        //                    remain_count_key = f"{config_.LOCAL_DISTRIBUTE_COUNT_PREFIX}{video_id}:{flow_pool}"
+        //                    self.redis_helper.del_keys(remain_count_key)
+        //                    value = '{}-{}'.format(video_id, flow_pool)
+        //                    for item in config_.APP_TYPE:
+        //                        for level in level_list:
+        //                            flow_pool_key = f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL_SCORE}{config_.APP_TYPE.get(item)}:{level}"
+        //                            remove_res = self.redis_helper.remove_value_from_zset(key_name=flow_pool_key, value=value)
+        //                            if remove_res > 0:
+        //                                add_remove_log = True
+        //                        quick_flow_pool_key = f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX}{config_.APP_TYPE.get(item)}" \
+        //                                              f":{config_.QUICK_FLOW_POOL_ID}"
+        //                        remove_res = self.redis_helper.remove_value_from_zset(key_name=quick_flow_pool_key, value=value)
+        //                        if remove_res > 0:
+        //                            add_remove_log = True
+        //                    if add_remove_log is True:
+        //                        log_.info({'tag': 'remove video_id from flow_pool', 'video_id': video_id, 'flow_pool': flow_pool})
+        //        return check_result
+        return null;
+    }
+}

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

@@ -0,0 +1,30 @@
+package com.tzld.piaoquan.recommend.server.service.filter;
+
+import com.tzld.piaoquan.recommend.server.service.FlowPoolConfigService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Service;
+
+import java.util.Map;
+
+/**
+ * @author dyp
+ */
+@Service
+public class VideoFilterService {
+    @Autowired
+    private RedisTemplate<String, String> redisTemplate;
+
+    @Autowired
+    private FilterService filterService;
+
+    public FilterResult filter(FilterParam param) {
+        FilterResult result = filterService.filter(param);
+        result.getVideoIds();
+
+        //     filter_ = FilterVideos(request_id=self.request_id,
+        //        //                               app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
+        //        //        filtered_viewed_videos = filter_.filter_videos_status(pool_type='normal')
+        return null;
+    }
+}

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

@@ -16,5 +16,6 @@ public class RecallParam {
     private String abCode;
     private int size;
     private String flowPoolAbtestGroup;
+    private String videoId;
 
 }

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

@@ -3,7 +3,7 @@ package com.tzld.piaoquan.recommend.server.service.recall.strategy;
 import com.google.common.collect.Lists;
 import com.tzld.piaoquan.recommend.server.service.filter.FilterParam;
 import com.tzld.piaoquan.recommend.server.service.filter.FilterResult;
-import com.tzld.piaoquan.recommend.server.service.filter.FlowPoolFilterService;
+import com.tzld.piaoquan.recommend.server.service.filter.FlowPoolWithLevelFilterService;
 import com.tzld.piaoquan.recommend.server.service.recall.RecallParam;
 import com.tzld.piaoquan.recommend.server.service.recall.RecallResult;
 import com.tzld.piaoquan.recommend.server.service.recall.RecallStrategy;
@@ -27,7 +27,7 @@ public abstract class AbstractFlowPoolWithLevelRecallStrategy implements RecallS
     protected RedisTemplate<String, String> redisTemplate;
 
     @Autowired
-    protected FlowPoolFilterService flowPoolFilterService;
+    protected FlowPoolWithLevelFilterService flowPoolFilterService;
 
     @Override
     public RecallResult recall(RecallParam param) {
@@ -60,7 +60,7 @@ public abstract class AbstractFlowPoolWithLevelRecallStrategy implements RecallS
                 recallData.setPushFrom(pushFrom());
                 recallData.setFlowPool(videoMap.get(vid));
                 recallData.setFlowPoolAbtestGroup(param.getFlowPoolAbtestGroup());
-                recallData.setLevel("level");
+                recallData.setLevel(level);
                 results.add(recallData);
             });
         }

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

@@ -0,0 +1,83 @@
+package com.tzld.piaoquan.recommend.server.service.recall.strategy;
+
+import com.google.common.collect.Lists;
+import com.tzld.piaoquan.recommend.server.service.filter.FilterParam;
+import com.tzld.piaoquan.recommend.server.service.filter.FilterResult;
+import com.tzld.piaoquan.recommend.server.service.filter.FlowPoolWithLevelFilterService;
+import com.tzld.piaoquan.recommend.server.service.filter.FlowPoolWithLevelScoreFilterService;
+import com.tzld.piaoquan.recommend.server.service.recall.RecallParam;
+import com.tzld.piaoquan.recommend.server.service.recall.RecallResult;
+import com.tzld.piaoquan.recommend.server.service.recall.RecallStrategy;
+import org.apache.commons.collections4.CollectionUtils;
+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.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.ZSetOperations;
+
+import java.util.*;
+
+/**
+ * @author dyp
+ */
+public abstract class AbstractFlowPoolWithLevelScoreRecallStrategy implements RecallStrategy {
+    @Autowired
+    protected RedisTemplate<String, String> redisTemplate;
+
+    @Autowired
+    protected FlowPoolWithLevelScoreFilterService flowPoolFilterService;
+
+    @Override
+    public RecallResult recall(RecallParam param) {
+        Pair<String, String> flowPoolKeyAndLevel = flowPoolKeyAndLevel(param);
+        String flowPoolKey = flowPoolKeyAndLevel.getLeft();
+        String level = flowPoolKeyAndLevel.getRight();
+
+        int idx = 0;
+        int getSize = param.getSize() * 5;
+        List<RecallResult.RecallData> results = new ArrayList<>();
+
+        Set<ZSetOperations.TypedTuple<String>> data = redisTemplate.opsForZSet().reverseRangeWithScores(flowPoolKey, idx,
+                idx + getSize - 1);
+        if (CollectionUtils.isEmpty(data)) {
+            return null;
+        }
+        idx += getSize;
+
+        Map<Long, Double> videoScoreMap = new HashMap<>();
+        Map<Long, String> videoFlowPoolMap = new HashMap<>();
+        for (ZSetOperations.TypedTuple<String> t : data) {
+            String[] values = t.getValue().split("-");
+            videoFlowPoolMap.put(NumberUtils.toLong(values[0], 0), values[1]);
+            videoScoreMap.put(NumberUtils.toLong(values[0], 0), t.getScore());
+        }
+
+        FilterParam filterParam = new FilterParam();
+        filterParam.setVideoIds(Lists.newArrayList(videoFlowPoolMap.keySet()));
+        FilterResult filterResult = flowPoolFilterService.filter(filterParam);
+
+        if (filterResult != null && CollectionUtils.isNotEmpty(filterResult.getVideoIds())) {
+            filterResult.getVideoIds().stream().forEach(vid -> {
+                RecallResult.RecallData recallData = new RecallResult.RecallData();
+                recallData.setVideoId(vid);
+                recallData.setAbCode(param.getAbCode());
+                recallData.setRovScore(videoScoreMap.get(vid));
+                recallData.setPushFrom(pushFrom());
+                recallData.setFlowPool(videoFlowPoolMap.get(vid));
+                recallData.setFlowPoolAbtestGroup(param.getFlowPoolAbtestGroup());
+                recallData.setLevel(level);
+                results.add(recallData);
+            });
+        }
+
+        RecallResult result = new RecallResult();
+        result.setRecallData(results.subList(0, results.size() < param.getSize() ? results.size() : param.getSize()));
+        return result;
+    }
+
+    abstract Pair<String, String> flowPoolKeyAndLevel(RecallParam param);
+
+    abstract String pushFrom();
+
+}

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

@@ -0,0 +1,77 @@
+package com.tzld.piaoquan.recommend.server.service.recall.strategy;
+
+import com.google.common.collect.Lists;
+import com.tzld.piaoquan.recommend.server.service.filter.FilterParam;
+import com.tzld.piaoquan.recommend.server.service.filter.FilterResult;
+import com.tzld.piaoquan.recommend.server.service.filter.FlowPoolWithScoreFilterService;
+import com.tzld.piaoquan.recommend.server.service.recall.RecallParam;
+import com.tzld.piaoquan.recommend.server.service.recall.RecallResult;
+import com.tzld.piaoquan.recommend.server.service.recall.RecallStrategy;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.math.NumberUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.ZSetOperations;
+
+import java.util.*;
+
+/**
+ * @author dyp
+ */
+public abstract class AbstractFlowPoolWithScoreRecallStrategy implements RecallStrategy {
+    @Autowired
+    protected RedisTemplate<String, String> redisTemplate;
+
+    @Autowired
+    protected FlowPoolWithScoreFilterService flowPoolFilterService;
+
+    @Override
+    public RecallResult recall(RecallParam param) {
+        String flowPoolKey = flowPoolKey(param);
+
+        int idx = 0;
+        int getSize = param.getSize() * 5;
+        List<RecallResult.RecallData> results = new ArrayList<>();
+
+        Set<ZSetOperations.TypedTuple<String>> data = redisTemplate.opsForZSet().reverseRangeWithScores(flowPoolKey, idx,
+                idx + getSize - 1);
+        if (CollectionUtils.isEmpty(data)) {
+            return null;
+        }
+        idx += getSize;
+
+        Map<Long, Double> videoScoreMap = new HashMap<>();
+        Map<Long, String> videoFlowPoolMap = new HashMap<>();
+        for (ZSetOperations.TypedTuple<String> t : data) {
+            String[] values = t.getValue().split("-");
+            videoFlowPoolMap.put(NumberUtils.toLong(values[0], 0), values[1]);
+            videoScoreMap.put(NumberUtils.toLong(values[0], 0), t.getScore());
+        }
+
+        FilterParam filterParam = new FilterParam();
+        filterParam.setVideoIds(Lists.newArrayList(videoFlowPoolMap.keySet()));
+        FilterResult filterResult = flowPoolFilterService.filter(filterParam);
+
+        if (filterResult != null && CollectionUtils.isNotEmpty(filterResult.getVideoIds())) {
+            filterResult.getVideoIds().stream().forEach(vid -> {
+                RecallResult.RecallData recallData = new RecallResult.RecallData();
+                recallData.setVideoId(vid);
+                recallData.setAbCode(param.getAbCode());
+                recallData.setRovScore(videoScoreMap.get(vid));
+                recallData.setPushFrom(pushFrom());
+                recallData.setFlowPool(videoFlowPoolMap.get(vid));
+                recallData.setFlowPoolAbtestGroup(param.getFlowPoolAbtestGroup());
+                results.add(recallData);
+            });
+        }
+
+        RecallResult result = new RecallResult();
+        result.setRecallData(results.subList(0, results.size() < param.getSize() ? results.size() : param.getSize()));
+        return result;
+    }
+
+    abstract String flowPoolKey(RecallParam param);
+
+    abstract String pushFrom();
+
+}

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

@@ -0,0 +1,83 @@
+package com.tzld.piaoquan.recommend.server.service.recall.strategy;
+
+import com.alibaba.fastjson.JSONObject;
+import com.google.common.collect.Lists;
+import com.tzld.piaoquan.recommend.server.service.filter.FilterParam;
+import com.tzld.piaoquan.recommend.server.service.filter.FilterResult;
+import com.tzld.piaoquan.recommend.server.service.filter.VideoFilterService;
+import com.tzld.piaoquan.recommend.server.service.recall.RecallParam;
+import com.tzld.piaoquan.recommend.server.service.recall.RecallResult;
+import com.tzld.piaoquan.recommend.server.service.recall.RecallStrategy;
+import lombok.Data;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.math.NumberUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.RedisTemplate;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @author dyp
+ */
+public abstract class AbstractVideoRecallStrategy implements RecallStrategy {
+    @Autowired
+    protected RedisTemplate<String, String> redisTemplate;
+
+    @Autowired
+    protected VideoFilterService simHotItemFilterService;
+
+    @Override
+    public RecallResult recall(RecallParam param) {
+
+
+        String recall_key = recallKey(param);
+        // [[16687915, 0.5], [16528093, 0.5]]
+        String value = redisTemplate.opsForValue().get(recall_key);
+        if (StringUtils.isBlank(value)) {
+            return null;
+        }
+        List<Item> items = JSONObject.parseArray(value, Item.class);
+        List<RecallResult.RecallData> results = new ArrayList<>();
+
+        Map<Long, Double> videoScoreMap = new HashMap<>();
+        for (int i = 0; i < items.size() && i < 20; i++) {
+            Item item = items.get(i);
+            videoScoreMap.put(NumberUtils.toLong(item.getVideoId(), 0),
+                    NumberUtils.toDouble(item.getScore(), 0));
+        }
+
+        FilterParam filterParam = new FilterParam();
+        filterParam.setVideoIds(Lists.newArrayList(videoScoreMap.keySet()));
+        FilterResult filterResult = simHotItemFilterService.filter(filterParam);
+
+        if (filterResult != null && CollectionUtils.isNotEmpty(filterResult.getVideoIds())) {
+            filterResult.getVideoIds().stream().forEach(vid -> {
+                RecallResult.RecallData recallData = new RecallResult.RecallData();
+                recallData.setVideoId(vid);
+                recallData.setAbCode(param.getAbCode());
+                recallData.setRovScore(videoScoreMap.get(vid));
+                recallData.setPushFrom(pushFrom());
+                recallData.setFlowPool("");
+                results.add(recallData);
+            });
+        }
+
+        RecallResult result = new RecallResult();
+        result.setRecallData(results);
+        return result;
+    }
+
+    @Data
+    class Item {
+        private String videoId;
+        private String score;
+    }
+
+    abstract String recallKey(RecallParam param);
+    abstract String pushFrom();
+}

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

@@ -0,0 +1,97 @@
+package com.tzld.piaoquan.recommend.server.service.recall.strategy;
+
+import com.tzld.piaoquan.recommend.server.service.FlowPoolConfigService;
+import com.tzld.piaoquan.recommend.server.service.recall.RecallParam;
+import lombok.Data;
+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.*;
+
+/**
+ * @author dyp
+ */
+@Service
+public class FlowPoolWithLevelScoreRecallStrategy extends AbstractFlowPoolWithLevelRecallStrategy {
+
+    @Autowired
+    private FlowPoolConfigService flowPoolConfigService;
+
+    @Override
+    Pair<String, String> flowPoolKeyAndLevel(RecallParam param) {
+        //# 1. 获取流量池各层级分发概率权重
+        Map<String, Double> levelWeightMap = flowPoolConfigService.getLevelWeight();
+
+        // 2. 判断各层级是否有视频需分发
+        List<LevelWeight> availableLevels = new ArrayList<>();
+        levelWeightMap.forEach((l, w) -> {
+            String levelKey = String.format("flow:pool:level:item:score:%s:%s", param.getAppType(), l);
+            if (redisTemplate.hasKey(levelKey)) {
+                LevelWeight lw = new LevelWeight();
+                lw.setLevel(l);
+                lw.setLevelKey(levelKey);
+                lw.setWeight(w);
+                availableLevels.add(lw);
+            }
+        });
+        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
+
+        int random_p = RandomUtils.nextInt(0, 1);
+        for (Map.Entry<String, LevelP> entry : level_p_mapping.entrySet()) {
+            if (random_p >= entry.getValue().getMin()
+                    && random_p >= entry.getValue().getMax()) {
+                return Pair.of(entry.getValue().getLevelKey(), entry.getKey());
+            }
+        }
+        return Pair.of("", "");
+    }
+
+    @Data
+    class LevelWeight {
+        private String level;
+        private String levelKey;
+        private Double weight;
+    }
+
+    @Data
+    class LevelP {
+        private String levelKey;
+        private double min;
+        private double max;
+    }
+
+    @Override
+    String pushFrom() {
+        return "flow_pool";
+    }
+}

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

@@ -0,0 +1,21 @@
+package com.tzld.piaoquan.recommend.server.service.recall.strategy;
+
+import com.tzld.piaoquan.recommend.server.service.recall.RecallParam;
+import org.springframework.stereotype.Service;
+
+/**
+ * @author dyp
+ */
+@Service
+public class FlowPoolWithScoreRecallStrategy extends AbstractFlowPoolWithScoreRecallStrategy {
+
+    @Override
+    String flowPoolKey(RecallParam param) {
+        return String.format("flow:pool:item:score:%s", param.getAppType());
+    }
+
+    @Override
+    String pushFrom() {
+        return "flow_pool";
+    }
+}

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

@@ -0,0 +1,22 @@
+package com.tzld.piaoquan.recommend.server.service.recall.strategy;
+
+import com.tzld.piaoquan.recommend.server.service.recall.RecallParam;
+import org.apache.commons.lang3.tuple.Pair;
+import org.springframework.stereotype.Service;
+
+/**
+ * @author dyp
+ */
+@Service
+public class QuickFlowPoolWithLevelScoreRecallStrategy extends AbstractFlowPoolWithLevelScoreRecallStrategy {
+
+    @Override
+    Pair<String, String> flowPoolKeyAndLevel(RecallParam param) {
+        return Pair.of(String.format("flow:pool:quick:item:score:%s:3", param.getAppType()), "");
+    }
+
+    @Override
+    String pushFrom() {
+        return "flow_pool";
+    }
+}

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

@@ -0,0 +1,21 @@
+package com.tzld.piaoquan.recommend.server.service.recall.strategy;
+
+import com.tzld.piaoquan.recommend.server.service.recall.RecallParam;
+import org.springframework.stereotype.Service;
+
+/**
+ * @author dyp
+ */
+@Service
+public class QuickFlowPoolWithScoreRecallStrategy extends AbstractFlowPoolWithScoreRecallStrategy {
+
+    @Override
+    String flowPoolKey(RecallParam param) {
+        return String.format("flow:pool:quick:item:score:%s:3", param.getAppType());
+    }
+
+    @Override
+    String pushFrom() {
+        return "flow_pool";
+    }
+}

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

@@ -0,0 +1,19 @@
+package com.tzld.piaoquan.recommend.server.service.recall.strategy;
+
+import com.tzld.piaoquan.recommend.server.service.recall.RecallParam;
+
+/**
+ * @author dyp
+ */
+public class ReturnVideoRecallStrategy extends AbstractVideoRecallStrategy {
+
+    @Override
+    String recallKey(RecallParam param) {
+        return "rv2:" + param.getVideoId();
+    }
+
+    @Override
+    protected String pushFrom() {
+        return "return_video_recall";
+    }
+}

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

@@ -0,0 +1,19 @@
+package com.tzld.piaoquan.recommend.server.service.recall.strategy;
+
+import com.tzld.piaoquan.recommend.server.service.recall.RecallParam;
+
+/**
+ * @author dyp
+ */
+public class SimHotVideoRecallStrategy extends AbstractVideoRecallStrategy {
+
+    @Override
+    String recallKey(RecallParam param) {
+        return "sim_hot_" + param.getVideoId();
+    }
+
+    @Override
+    protected String pushFrom() {
+        return "sim_hot_vid_recall";
+    }
+}