MOD：Filter 重构

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/common/base/RankItem.java

@@ -1,14 +1,12 @@
 package com.tzld.piaoquan.recommend.server.common.base;
 
-import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
 import com.tzld.piaoquan.recommend.feature.domain.video.base.ItemFeature;
 import com.tzld.piaoquan.recommend.server.framework.candidiate.CandidateInfo;
 import com.tzld.piaoquan.recommend.server.model.Video;
 import lombok.Data;
 
-
-import com.google.common.collect.Maps;
-
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -35,11 +33,11 @@ public class RankItem implements Comparable<RankItem> {
     // 记录召回信息
     private String queue;
     private CandidateInfo candidateInfo;
-    private List<CandidateInfo> candidateInfoList; // 兼容多个召回key命中
+    private List<CandidateInfo> candidateInfoList = new ArrayList<>(); // 兼容多个召回key命中
 
     // merge信息
-    private List<String> mergeQueuePath;
-    private List<String> mergeDecisionLabels;
+    private List<String> mergeQueuePath = new ArrayList<>();
+    private List<String> mergeDecisionLabels = new ArrayList<>();
 
     // 多样性过滤因子
     // 排序因子
@@ -50,8 +48,8 @@ public class RankItem implements Comparable<RankItem> {
 
 
     // 排序侧信息
-    private Map<String, Double> rankerScore;
-    private Map<String, Integer> rankerIndex;
+    private Map<String, Double> rankerScore = Maps.newHashMap();
+    private Map<String, Integer> rankerIndex = Maps.newHashMap();
     public RankItem(){
 
     }
@@ -83,8 +81,6 @@ public class RankItem implements Comparable<RankItem> {
 
         // merge queue
         this.queue = other.queue;
-        this.mergeQueuePath = Lists.newArrayList();
-        this.mergeDecisionLabels = Lists.newArrayList();
         if (other.getMergeQueuePath() != null)
             this.mergeQueuePath.addAll(other.getMergeQueuePath());
         if (other.getMergeDecisionLabels() != null)
@@ -93,7 +89,6 @@ public class RankItem implements Comparable<RankItem> {
 
         // candidateinfo
         this.candidateInfo = other.getCandidateInfo() != null ? other.getCandidateInfo().deepcopy() : null;
-        this.candidateInfoList = Lists.newArrayList();
         if (other.getCandidateInfoList() != null) {
             for (CandidateInfo tmpCandidateInfo : other.getCandidateInfoList()) {
                 this.candidateInfoList.add(tmpCandidateInfo.deepcopy());

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/framework/candidiate/IndexCandidateQueue.java

@@ -74,4 +74,14 @@ public abstract class IndexCandidateQueue extends StrategyQueue {
             return num;
         }
     }
+
+    @Override
+    public String toString() {
+        return "IndexCandidateQueue{" +
+                "myCandidates=" + myCandidates +
+                ", rankerItemsList=" + rankerItemsList +
+                ", children=" + children +
+                ", items=" + items +
+                '}';
+    }
 }

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/framework/merger/MergeUtils.java

@@ -4,27 +4,18 @@ package com.tzld.piaoquan.recommend.server.framework.merger;
 import com.google.common.base.Function;
 import com.google.common.collect.ArrayListMultimap;
 import com.google.common.collect.Multimap;
-
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigFactory;
 import com.tzld.piaoquan.recommend.server.common.base.RankItem;
-
-
-import com.tzld.piaoquan.recommend.server.framework.common.User;
 import com.tzld.piaoquan.recommend.server.framework.candidiate.CandidateInfo;
+import com.tzld.piaoquan.recommend.server.framework.common.User;
 import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendRequest;
 import org.apache.commons.lang3.tuple.Pair;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.lang.reflect.InvocationTargetException;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.PriorityQueue;
+import java.util.*;
 
 
 public class MergeUtils {
@@ -93,14 +84,10 @@ public class MergeUtils {
                 mergeQueuesItems.put(mergeQueue, currentItem);
             }
         }
-        // debug log
-        if (LOGGER.isDebugEnabled()) {
-            for (Map.Entry<String, RankItem> entry : mergeQueuesItems.entries()) {
-                LOGGER.debug("after distribute item queue info: [{}] [{}] [{}]",
-                        new Object[]{entry.getKey(), entry.getValue().getId(), entry.getValue().getQueue()});
-            }
-        }
-
+//         debug log
+//        for (Map.Entry<String, RankItem> entry : mergeQueuesItems.entries()) {
+//            LOGGER.info("after distribute item queue info: entry: [{}]", entry);
+//        }
         for (StrategyQueue queue : strategyQueueList) {
             String mergeQueueName = queue.getStrategyQueueInfo().getQueueName();
             if (mergeQueuesItems.containsKey(mergeQueueName)) {

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/framework/merger/SimpleMergeQueue.java

@@ -24,4 +24,13 @@ public class SimpleMergeQueue extends StrategyQueue {
     public int candidate(Map<String, Candidate> candidateMap, final int recallNum, final User user, final RecommendRequest requestData, final int requestIndex, final int expId) {
         return simpleCandidateByPercentage(candidateMap, recallNum, user, requestData, requestIndex, expId);
     }
+
+    @Override
+    public String toString() {
+        return "SimpleMergeQueue{" +
+                "rankerItemsList=" + rankerItemsList +
+                ", children=" + children +
+                ", items=" + items +
+                '}';
+    }
 }

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/framework/merger/StrategyQueue.java

@@ -1,6 +1,7 @@
 package com.tzld.piaoquan.recommend.server.framework.merger;
 
 
+import com.google.common.collect.Maps;
 import com.tzld.piaoquan.recommend.server.common.base.RankItem;
 import com.tzld.piaoquan.recommend.server.framework.candidiate.Candidate;
 import com.tzld.piaoquan.recommend.server.framework.common.User;
@@ -16,20 +17,17 @@ import java.util.*;
 @Data
 public abstract class StrategyQueue {
     // 得等召回结果，定义为RankItem
-    protected final Map<String, Pair<MergeRule, List<RankItem>>> rankerItemsList = new HashMap<String, Pair<MergeRule, List<RankItem>>>();
+    protected final Map<String, Pair<MergeRule, List<RankItem>>> rankerItemsList = new HashMap<>();
     private final Logger LOGGER = LoggerFactory.getLogger(StrategyQueue.class);
-    protected Map<String, StrategyQueue> children;
+    protected Map<String, StrategyQueue> children = Maps.newHashMap();
 
-    protected List<RankItem> items;
+    protected List<RankItem> items = new ArrayList<>();
     private StrategyQueueInfo strategyQueueInfo;
     private List<StrategyQueue> allQueues = null;
 
     public StrategyQueue(StrategyQueueInfo strategyQueueInfo, StrategyQueueConfig strategyQueueConfig) {
         this.strategyQueueInfo = strategyQueueInfo;
 
-        items = new LinkedList<RankItem>();
-        children = new HashMap<String, StrategyQueue>();
-
         // 自顶向下构建child queue 初始化
         for (String childName : strategyQueueInfo.getChildren()) {
             if (!strategyQueueConfig.getStrategyQueueInfoMap().containsKey(childName)) {
@@ -64,6 +62,21 @@ public abstract class StrategyQueue {
 
     }
 
+    public List<StrategyQueue> getAllQueues() {
+        if (allQueues == null) {
+            allQueues = new LinkedList<>();
+
+            if (!strategyQueueInfo.isLeaf()) {
+                for (Map.Entry<String, StrategyQueue> child : children.entrySet()) {
+                    allQueues.addAll(child.getValue().getAllQueues());
+                }
+            }
+
+            allQueues.add(this);
+        }
+        return allQueues;
+    }
+
 
     private void putChild(String queueName, StrategyQueue child) {
         children.put(queueName, child);
@@ -176,4 +189,14 @@ public abstract class StrategyQueue {
         return n;
     }
 
+    @Override
+    public String toString() {
+        return "StrategyQueue{" +
+                "rankerItemsList=" + rankerItemsList +
+                ", children=" + children +
+                ", items=" + items +
+                ", strategyQueueInfo=" + strategyQueueInfo +
+                ", allQueues=" + allQueues +
+                '}';
+    }
 }

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/framework/recaller/AbstractFilter.java

@@ -6,6 +6,8 @@ import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendRequest;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.List;
+
 public abstract class AbstractFilter<T> {
     protected final static Logger LOGGER = LoggerFactory.getLogger(AbstractFilter.class);
     protected final FilterConfigInfo filterConfigInfo;
@@ -25,5 +27,5 @@ public abstract class AbstractFilter<T> {
         return filterConfigInfo;
     }
 
-    public abstract boolean predicate(Candidate candidate, T t);
+    public abstract void doFilter(Candidate candidate, List<T> t);
 }

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/framework/recaller/BaseRecaller.java

@@ -1,7 +1,6 @@
 package com.tzld.piaoquan.recommend.server.framework.recaller;
 
 
-import com.google.common.base.Predicate;
 import com.google.common.collect.FluentIterable;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
@@ -10,9 +9,9 @@ import com.tzld.piaoquan.recommend.server.framework.candidiate.*;
 import com.tzld.piaoquan.recommend.server.framework.common.User;
 import com.tzld.piaoquan.recommend.server.framework.recaller.provider.QueueProvider;
 import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendRequest;
+import com.tzld.piaoquan.recommend.server.util.CommonCollectionUtils;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang.exception.ExceptionUtils;
-import org.checkerframework.checker.nullness.qual.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -21,6 +20,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.*;
+import java.util.stream.Collectors;
 
 
 public class BaseRecaller<Video> {
@@ -88,6 +88,8 @@ public class BaseRecaller<Video> {
             candidateInfo.setCandidateQueueName(candidate.getCandidateKey());
             candidateInfo.setCandidate(candidate);
 
+            item.setCandidateInfo(candidateInfo);
+            item.addToCandidateInfoList(candidateInfo);
             result.add(item);
         }
         return result;
@@ -200,17 +202,24 @@ public class BaseRecaller<Video> {
             callables.add(() -> {
                 List<RankItem> candidateHits = new ArrayList<>();
                 final Candidate candidate = entry.getKey();
+                List<Entry<Video>> entriesValue = CommonCollectionUtils.iterableToList(entry.getValue());
+
                 try {
+                    final Map<Video, Entry<Video>> entryMap = new HashMap<>();
+                    List<Video> entriesList = entriesValue.stream().map(e -> {
+                        entryMap.put(e.item, e);
+                        return e.item;
+                    }).collect(Collectors.toList());
+                    List<Video> entriesAfterFilter = recallFilter.doFilter(candidate, entriesList);
                     // 1. filter  TODO 待后续增加自定义filter
-                    Iterable<Entry<Video>> entries = FluentIterable.from(entry.getValue()).filter(new Predicate<Entry<Video>>() {
-                        @Override
-                        public boolean apply(@Nullable Entry<Video> videoEntry) {
-                            return recallFilter.predicate(candidate, videoEntry.item);
-                        }
-                    }).limit(candidate.getCandidateNum());
-
-                    // 2. toHits
-                    candidateHits.addAll(toHits(entries, candidate));
+                    if (CollectionUtils.isNotEmpty(entriesAfterFilter)) {
+                        List<Entry<Video>> entries = entriesAfterFilter.stream()
+                                .limit(candidate.getCandidateNum())
+                                .map(e -> entryMap.get(e))
+                                .collect(Collectors.toList());
+                        // 2. toHits
+                        candidateHits.addAll(toHits(entries, candidate));
+                    }
 
                     // debug log for tracing
                     LOGGER.debug("recalled candidate [{}], queue length [{}], expected [{}], hit [{}]",
@@ -222,7 +231,7 @@ public class BaseRecaller<Video> {
             });
         }
 
-        Map<String, RankItem> hits = new HashMap<String, RankItem>();
+        Map<String, RankItem> hits = new HashMap<>();
         try {
             List<Future<List<RankItem>>> futures = filterExecutorService.invokeAll(callables, DEFAULT_PARALLEL_FILTER_TIMEOUT, TimeUnit.MILLISECONDS);
             for (Future<List<RankItem>> future : futures) {
@@ -252,7 +261,7 @@ public class BaseRecaller<Video> {
                     "", ExceptionUtils.getFullStackTrace(e));
         }
 
-        List<RankItem> result = new ArrayList<RankItem>(hits.values());
+        List<RankItem> result = new ArrayList<>(hits.values());
 
         return result;
     }

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/framework/recaller/FilterPipeline.java

@@ -53,12 +53,10 @@ public class FilterPipeline<T> {
         }
     }
 
-    public boolean predicate(Candidate candidate, T t) {
+    public List<T> doFilter(Candidate candidate, List<T> list) {
         for (AbstractFilter filter : filters) {
-            if (!filter.predicate(candidate, t)) {
-                return false;
-            }
+            filter.doFilter(candidate, list);
         }
-        return true;
+        return list;
     }
 }

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

@@ -5,7 +5,6 @@ import com.tzld.piaoquan.recommend.server.common.base.RankItem;
 import com.tzld.piaoquan.recommend.server.framework.candidiate.Candidate;
 import com.tzld.piaoquan.recommend.server.framework.common.User;
 import com.tzld.piaoquan.recommend.server.framework.merger.MergeUtils;
-import com.tzld.piaoquan.recommend.server.framework.merger.SimilarityUtils;
 import com.tzld.piaoquan.recommend.server.framework.merger.StrategyQueue;
 import com.tzld.piaoquan.recommend.server.framework.recaller.BaseRecaller;
 import com.tzld.piaoquan.recommend.server.framework.recaller.provider.RedisBackedQueue;

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/implement/candidate/Global24hHotCandidate.java

@@ -48,4 +48,14 @@ public class Global24hHotCandidate extends IndexCandidateQueue {
         return currRecall;
 
     }
+
+    @Override
+    public String toString() {
+        return "Global24hHotCandidate{" +
+                "myCandidates=" + myCandidates +
+                ", rankerItemsList=" + rankerItemsList +
+                ", children=" + children +
+                ", items=" + items +
+                '}';
+    }
 }

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/implement/candidate/HotCandidateQueue.java

@@ -23,7 +23,17 @@ public class HotCandidateQueue extends IndexCandidateQueue {
     @Override
     public int addCandidateKey(Map<String, Candidate> candidates, int recallNum, User user,
                                RecommendRequest requestData, int requestIndex, int expId) {
-        
+
         return 0;
     }
+
+    @Override
+    public String toString() {
+        return "HotCandidateQueue{" +
+                "myCandidates=" + myCandidates +
+                ", rankerItemsList=" + rankerItemsList +
+                ", children=" + children +
+                ", items=" + items +
+                '}';
+    }
 }

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/implement/candidate/Region1hHotCandidate.java

@@ -63,4 +63,14 @@ public class Region1hHotCandidate extends IndexCandidateQueue {
         candidates.putAll(simplifiedCandidates);
         return currRecall;
     }
+
+    @Override
+    public String toString() {
+        return "Region1hHotCandidate{" +
+                "myCandidates=" + myCandidates +
+                ", rankerItemsList=" + rankerItemsList +
+                ", children=" + children +
+                ", items=" + items +
+                '}';
+    }
 }

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/implement/candidate/Region24hHotCandidate.java

@@ -55,4 +55,14 @@ public class Region24hHotCandidate extends IndexCandidateQueue {
         return currRecall;
 
     }
+
+    @Override
+    public String toString() {
+        return "Region24hHotCandidate{" +
+                "myCandidates=" + myCandidates +
+                ", rankerItemsList=" + rankerItemsList +
+                ", children=" + children +
+                ", items=" + items +
+                '}';
+    }
 }

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

@@ -36,8 +36,21 @@ public class AllowListFilter extends AbstractFilter<Video> {
 
     }
 
-    public boolean predicate(Candidate candidate, Video video) {
-        return !isAllowList(user, video);
+
+
+    @Override
+    public void doFilter(Candidate candidate, List<Video> list) {
+        if (CollectionUtils.isEmpty(list)) {
+            return;
+        }
+        List<Video> result = Lists.newArrayList();
+        for (Video video : list) {
+            if (!isAllowList(user, video)) {
+                result.add(video);
+            }
+        }
+        list.clear();
+        list.addAll(result);
     }
 
 
@@ -92,7 +105,7 @@ public class AllowListFilter extends AbstractFilter<Video> {
 
 
 
-    
+
 }
 
 

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

@@ -7,18 +7,21 @@ import com.tzld.piaoquan.recommend.server.framework.recaller.FilterConfigInfo;
 import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendRequest;
 import com.tzld.piaoquan.recommend.server.model.Video;
 
+import java.util.List;
+
 /**
  * @author sunxy
  */
 public class HistoryLongPeriodFilter extends AbstractFilter<Video> {
 
+
     public HistoryLongPeriodFilter(FilterConfigInfo filterConfigInfo, RecommendRequest requestContext, User user) {
         super(filterConfigInfo, requestContext, user);
     }
 
     @Override
-    public boolean predicate(Candidate candidate, Video aLong) {
+    public void doFilter(Candidate candidate, List<Video> list) {
 
-        return false;
     }
+
 }

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

@@ -8,9 +8,11 @@ import com.tzld.piaoquan.recommend.server.framework.recaller.FilterConfigInfo;
 import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendRequest;
 import com.tzld.piaoquan.recommend.server.model.Video;
 import com.tzld.piaoquan.recommend.server.service.PreViewedService;
+import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 
+import java.util.List;
 import java.util.Set;
 
 public class PreViewedFilter extends AbstractFilter<Video> {
@@ -25,23 +27,21 @@ public class PreViewedFilter extends AbstractFilter<Video> {
 
     }
 
-    public boolean predicate(Candidate candidate, Video video) {
-        return !isPreViewed(user, video);
-    }
-
-
-    public boolean isPreViewed(User user, Video video) {
-
+    @Override
+    public void doFilter(Candidate candidate, List<Video> list) {
         if (StringUtils.isBlank(user.getMid())
-                || StringUtils.isBlank(String.valueOf(video.getVideoId()))) {
-            return false;
+                || CollectionUtils.isEmpty(list)) {
+            return;
         }
         Set<Long> preViewedVideoIds = preViewedService.getVideoIds(this.requestContext.getAppType(), user.getMid());
-
-        return preViewedVideoIds.contains(video.getVideoId());
+        if (CollectionUtils.isEmpty(preViewedVideoIds)) {
+            return;
+        }
+        list.removeIf(video -> preViewedVideoIds.contains(video.getVideoId()));
     }
 
 
+
 }
 
 

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

@@ -1,31 +1,20 @@
 package com.tzld.piaoquan.recommend.server.implement.recall;
 
 
-import com.tzld.piaoquan.recommend.server.common.ThreadPoolFactory;
 import com.tzld.piaoquan.recommend.server.framework.candidiate.Candidate;
 import com.tzld.piaoquan.recommend.server.framework.common.User;
 import com.tzld.piaoquan.recommend.server.framework.recaller.AbstractFilter;
 import com.tzld.piaoquan.recommend.server.framework.recaller.FilterConfigInfo;
 import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendRequest;
 import com.tzld.piaoquan.recommend.server.model.Video;
-import com.tzld.piaoquan.recommend.server.repository.WxVideoStatus;
 import com.tzld.piaoquan.recommend.server.repository.WxVideoStatusRepository;
 import com.tzld.piaoquan.recommend.server.service.PreViewedService;
 import org.apache.commons.collections4.CollectionUtils;
-import org.apache.commons.lang3.RandomUtils;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.commons.lang3.math.NumberUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.dao.DataAccessException;
-import org.springframework.data.redis.core.RedisOperations;
 import org.springframework.data.redis.core.RedisTemplate;
-import org.springframework.data.redis.core.SessionCallback;
-import org.springframework.data.redis.core.ValueOperations;
 
-import java.util.*;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
+import java.util.List;
 
 public class RecommendStatusFilter extends AbstractFilter<Video> {
 
@@ -51,16 +40,13 @@ public class RecommendStatusFilter extends AbstractFilter<Video> {
 
     }
 
-    public boolean predicate(Candidate candidate, Video video) {
-        return isStatusAllow(user, video);
+    @Override
+    public void doFilter(Candidate candidate, List<Video> t) {
+        if (CollectionUtils.isEmpty(t)) {
+            return;
+        }
     }
 
-    public boolean isStatusAllow(User user, Video video){
-        //
-        return true;
-
-
-    }
 
 
 

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

@@ -7,9 +7,8 @@ import com.tzld.piaoquan.recommend.server.framework.recaller.AbstractFilter;
 import com.tzld.piaoquan.recommend.server.framework.recaller.FilterConfigInfo;
 import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendRequest;
 import com.tzld.piaoquan.recommend.server.model.Video;
-import org.apache.commons.collections4.CollectionUtils;
 
-import java.util.*;
+import java.util.List;
 
 public class RiskVideoFilter extends AbstractFilter<Video> {
 
@@ -20,22 +19,10 @@ public class RiskVideoFilter extends AbstractFilter<Video> {
         super(filterConfigInfo, recommendRequest, user);
     }
 
-    public boolean predicate(Candidate candidate, Video video) {
-
-        return !isRiskVideo(video);
-
-
-    }
-
-
-    public boolean isRiskVideo(Video video) {
-
-        // TODO 风险过滤标签加载video中
-        //
-        return false;
+    @Override
+    public void doFilter(Candidate candidate, List<Video> t) {
 
     }
 
 
-
 }

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

@@ -5,7 +5,6 @@ import com.google.common.cache.CacheBuilder;
 import com.google.common.cache.CacheLoader;
 import com.google.common.cache.LoadingCache;
 import com.google.common.collect.Lists;
-import com.tzld.piaoquan.recommend.server.common.ThreadPoolFactory;
 import com.tzld.piaoquan.recommend.server.framework.candidiate.Candidate;
 import com.tzld.piaoquan.recommend.server.framework.common.User;
 import com.tzld.piaoquan.recommend.server.framework.recaller.AbstractFilter;
@@ -16,13 +15,14 @@ import com.tzld.piaoquan.recommend.server.repository.WxVideoTagRel;
 import com.tzld.piaoquan.recommend.server.repository.WxVideoTagRelRepository;
 import com.tzld.piaoquan.recommend.server.util.CommonCollectionUtils;
 import org.apache.commons.collections4.CollectionUtils;
-import org.apache.commons.lang.StringUtils;
 import org.apache.commons.lang.math.NumberUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 
-import javax.annotation.PostConstruct;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
 public class TagFilter extends AbstractFilter<Video> {
@@ -63,15 +63,20 @@ public class TagFilter extends AbstractFilter<Video> {
         }
     }
 
-
-    public boolean predicate(Candidate candidate, Video video) {
-        if (video == null ) {
-            return false;
+    @Override
+    public void doFilter(Candidate candidate, List<Video> t) {
+        if (CollectionUtils.isEmpty(t)) {
+            return;
+        }
+        List<Video> videos = Lists.newArrayList();
+        for (Video video : t) {
+            if (!hasVideoRelTagId(video.getVideoId(), videoFilterTagIds)) {
+                videos.add(video);
+            }
         }
-       return !hasVideoRelTagId(video.getVideoId(), this.videoFilterTagIds);
+        t.removeAll(videos);
     }
 
-
     private boolean hasVideoRelTagId(Long videoId, String videoFilterTagIds) {
         // TODO 主要是涉政标签
         List<Long> tagIds = new ArrayList<>();

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

@@ -46,12 +46,18 @@ public class ViewedHistoryFilter extends AbstractFilter<Video> {
                                User user) {
         super(filterConfigInfo, recommendRequest, user);
         historySet = this.loadUserHistory(user);
-        if (historySet == null)
+        if (historySet == null) {
             historySet = new HashSet<>();
+        }
     }
 
-    public boolean predicate(Candidate candidate, Video video) {
-        return !this.historySet.contains(String.valueOf(video.getVideoId()));
+    @Override
+    public void doFilter(Candidate candidate, List<Video> t) {
+        if (CollectionUtils.isEmpty(t)) {
+            return;
+        }
+
+        t.removeIf(video -> !this.historySet.contains(String.valueOf(video.getVideoId())));
     }
 
 

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

@@ -10,6 +10,15 @@ import java.util.stream.Collectors;
  * @author dyp
  */
 public class CommonCollectionUtils {
+
+    public static <T> List<T> iterableToList(Iterable<T> iterable) {
+        List<T> list = new ArrayList<>();
+        for (T element : iterable) {
+            list.add(element);
+        }
+        return list;
+    }
+
     public static <T, R> List<R> toList(List<T> list, Function<T, R> map) {
         if (CollectionUtils.isEmpty(list)) {
             return Collections.emptyList();