修改为并行打分

ad-engine-service/src/main/java/com/tzld/piaoquan/ad/engine/service/score/VlogMergeEcpmScorer.java

@@ -6,20 +6,25 @@ import com.tzld.piaoquan.ad.engine.commons.score.ScoreParam;
 import com.tzld.piaoquan.ad.engine.commons.score.ScorerConfigInfo;
 import com.tzld.piaoquan.ad.engine.commons.score.model.LRModel;
 import com.tzld.piaoquan.recommend.feature.domain.ad.base.AdRankItem;
+import com.tzld.piaoquan.recommend.feature.domain.ad.base.AdRequestContext;
+import com.tzld.piaoquan.recommend.feature.domain.ad.base.UserAdBytesFeature;
 import com.tzld.piaoquan.recommend.feature.domain.ad.base.UserAdFeature;
+import org.apache.commons.lang.exception.ExceptionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.*;
 
 
 //@Service
 public class VlogMergeEcpmScorer extends BaseLRModelScorer {
 
     private final static Logger LOGGER = LoggerFactory.getLogger(VlogMergeEcpmScorer.class);
-
-
+    private static final ExecutorService executorService = Executors.newFixedThreadPool(128);
+    private static final int LOCAL_TIME_OUT = 150;
     public VlogMergeEcpmScorer(ScorerConfigInfo configInfo) {
         super(configInfo);
     }
@@ -40,36 +45,48 @@ public class VlogMergeEcpmScorer extends BaseLRModelScorer {
     }
 
 
-    public List<AdRankItem> mergetEcpm(List<AdRankItem> items) {
-
 
+    public List<AdRankItem> mergetEcpm(List<AdRankItem> items) {
+        CountDownLatch countDownLatch = new CountDownLatch(items.size());
         for (AdRankItem item : items) {
-            try {
-                boolean isTfType=item.getItemFeature().getCreative_7day_cnt_features().getAdView()<10000;
-                double cpa = item.getCpa();
-                double bid1 = item.getBid1();
-                double bid2 = item.getBid2();
-                double pctr = isTfType?item.getTf_ctr():item.getCtr();
-                double pcvr = isTfType?item.getTf_cvr():item.getCvr();
-                item.setScore_type( isTfType?1:0);
-                //todo
-                double ecpm;
-                if(isTfType){
-                    ecpm = cpa * bid1 * bid2 * pcvr * pctr * 1000d;
-                }else {
-                    ecpm = (cpa+item.getPidLambda()) * bid1 * bid2 * pcvr * pctr * 1000d;
-                }
-//                double ecpm = cpa * bid1 * bid2 * pcvr * pctr;
-                item.setEcpm1(ecpm/1000d);
-                item.setScore(ecpm/1000d);
-            } catch (Exception e) {
-                LOGGER.error("merge Ecpm Score: {} error", "");
-                item.setCtr(0.0);
-            }
-            LOGGER.debug("merge Ecpm ranker , score: {}->{}", item.getScore());
+            executorService.execute(() -> {
+                callMergeScore(item,countDownLatch);
+            });
+        }
+        try {
+            countDownLatch.await(200, TimeUnit.MILLISECONDS);
+        } catch (InterruptedException e) {
+            LOGGER.error("mergetEcpm countDownLatch error ...", e);
         }
         Collections.sort(items);
         return items;
     }
 
+
+    public void callMergeScore(AdRankItem item,CountDownLatch countDownLatch){
+        try {
+            boolean isTfType=item.getItemFeature().getCreative_7day_cnt_features().getAdView()<10000;
+            double cpa = item.getCpa();
+            double bid1 = item.getBid1();
+            double bid2 = item.getBid2();
+            double pctr = isTfType?item.getTf_ctr():item.getCtr();
+            double pcvr = isTfType?item.getTf_cvr():item.getCvr();
+            item.setScore_type( isTfType?1:0);
+            //todo
+            double ecpm;
+            if(isTfType){
+                ecpm = cpa * bid1 * bid2 * pcvr * pctr * 1000d;
+            }else {
+                ecpm = (cpa+item.getPidLambda()) * bid1 * bid2 * pcvr * pctr * 1000d;
+            }
+//                double ecpm = cpa * bid1 * bid2 * pcvr * pctr;
+            item.setEcpm1(ecpm/1000d);
+            item.setScore(ecpm/1000d);
+        } catch (Exception e) {
+            LOGGER.error("merge Ecpm Score: {} error", "");
+            item.setCtr(0.0);
+        }finally {
+            countDownLatch.countDown();
+        }
+    }
 }