v0

recommend-server-service/pom.xml

@@ -28,6 +28,33 @@
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-pool2</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+            <version>3.4</version>
+        </dependency>
+        <dependency>
+            <groupId>com.typesafe</groupId>
+            <artifactId>config</artifactId>
+            <version>1.2.1</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-hdfs</artifactId>
+            <version>3.3.1</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-common</artifactId>
+            <version>3.3.1</version>
+        </dependency>
+        <dependency>
+            <groupId>it.unimi.dsi</groupId>
+            <artifactId>fastutil</artifactId>
+            <version>7.0.12</version>
+        </dependency>
+
+
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-test</artifactId>
@@ -105,7 +132,7 @@
         <dependency>
             <groupId>org.projectlombok</groupId>
             <artifactId>lombok</artifactId>
-            <version>1.16.8</version>
+            <version>1.18.24</version>
         </dependency>
 
         <dependency>

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

@@ -0,0 +1,8 @@
+package com.tzld.piaoquan.recommend.server.common.base;
+
+public class UserFeature {
+    public final byte[] sex;
+
+
+
+}

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

@@ -0,0 +1,10 @@
+package com.tzld.piaoquan.recommend.server.common.base;
+
+public class VideoRankFeature {
+    public final byte[] videoid;
+
+    public VideoRankFeature(String item) {
+        this.videoid = item.getBytes();
+    }
+
+}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/common/enums/VlogFeatureGroup.java

@@ -0,0 +1,39 @@
+package com.tzld.piaoquan.recommend.server.common.enums;
+
+public enum VlogFeatureGroup {
+    // video
+    VIDEOID,
+
+    // context
+    DAY_OF_WEEK,
+    SEX,
+    ;
+
+    // user
+
+
+    private final byte[] idBytes;
+    private final byte[] nameBytes;
+
+    VlogFeatureGroup() {
+        this.nameBytes = name().toLowerCase().getBytes();
+        this.idBytes = String.valueOf(ordinal()).getBytes();
+    }
+
+    public final int getId() {
+        return ordinal();
+    }
+
+    public final String getGroupName() {
+        return name().toLowerCase();
+    }
+
+    public final byte[] getGroupNameBytes() {
+        return getGroupName().getBytes();
+    }
+
+    public final byte[] getIdBytes() {
+        return idBytes;
+    }
+
+}

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

@@ -5,6 +5,8 @@ import com.tzld.piaoquan.recommend.server.model.Video;
 import com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConstants;
 import com.tzld.piaoquan.recommend.server.service.recall.RecallResult;
 import com.tzld.piaoquan.recommend.server.service.recall.strategy.*;
+import com.tzld.piaoquan.recommend.server.service.score.ScorerPipeline;
+import com.tzld.piaoquan.recommend.server.service.score.ScorerUtils;
 import com.tzld.piaoquan.recommend.server.util.JSONUtils;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections4.CollectionUtils;
@@ -27,6 +29,10 @@ public class RankService {
     private RedisTemplate<String, String> redisTemplate;
 
     public RankResult rank(RankParam param) {
+
+
+        ScorerPipeline scorerPipeline = ScorerUtils.getScorerPipeline();
+
         if (param == null
                 || param.getRecallResult() == null
                 || CollectionUtils.isEmpty(param.getRecallResult().getData())) {

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

@@ -0,0 +1,65 @@
+package com.tzld.piaoquan.recommend.server.service.score;
+
+
+import com.tzld.piaoquan.recommend.server.model.Video;
+import com.tzld.piaoquan.recommend.server.service.rank.RankParam;
+import com.tzld.piaoquan.recommend.server.service.rank.RankResult;
+import com.tzld.piaoquan.recommend.server.service.score.model.Model;
+import com.tzld.piaoquan.recommend.server.service.score.model.ModelManager;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.List;
+
+
+public abstract class AbstractScorer {
+    public static Logger LOGGER = LoggerFactory.getLogger(AbstractScorer.class);
+    protected ScorerConfigInfo scorerConfigInfo;
+    protected ModelManager modelManager = ModelManager.getInstance();
+
+    public AbstractScorer(ScorerConfigInfo scorerConfigInfo) {
+        this.scorerConfigInfo = scorerConfigInfo;
+    }
+
+    public void loadModel() {
+    }
+
+    public boolean isEnable() {
+        return !getScorerConfigInfo().getDisableSwitch();
+    }
+
+
+    public void doLoadModel(Class<? extends Model> modelClass) {
+
+        String modelPath = scorerConfigInfo.getModelPath();
+        if (StringUtils.isNotBlank(modelPath)) {
+            try {
+                // 使用 modelPath 作为 modelName 注册
+                modelManager.registerModel(modelPath, modelPath, modelClass);
+                LOGGER.info("register model success, model path [{}], model class [{}]", modelPath, modelClass);
+            } catch (ModelManager.ModelRegisterException e) {
+                LOGGER.error("register model fail [{}]:[{}]", modelPath, e);
+            } catch (IOException e) {
+                LOGGER.error("register model fail [{}]:[{}]", modelPath, e);
+            }
+        } else {
+            LOGGER.error("modelpath is null, for model class [{}]", modelClass);
+        }
+    }
+
+    public Model getModel() {
+        if (StringUtils.isBlank(scorerConfigInfo.getModelPath())) {
+            return null;
+        }
+        return modelManager.getModel(scorerConfigInfo.getModelPath());
+    }
+
+    public ScorerConfigInfo getScorerConfigInfo() {
+        return scorerConfigInfo;
+    }
+
+    public abstract List<Video> scoring(final RankParam param);
+
+}

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

@@ -0,0 +1,19 @@
+package com.tzld.piaoquan.recommend.server.service.score;
+
+import com.tzld.piaoquan.recommend.server.service.score.model.GBDTModel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class BaseGBDTModelScorer extends AbstractScorer {
+    private static Logger LOGGER = LoggerFactory.getLogger(BaseGBDTModelScorer.class);
+
+    public BaseGBDTModelScorer(ScorerConfigInfo scorerConfigInfo) {
+        super(scorerConfigInfo);
+    }
+
+    @Override
+    public synchronized void loadModel() {
+        doLoadModel(GBDTModel.class);
+    }
+
+}

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

@@ -0,0 +1,21 @@
+package com.tzld.piaoquan.recommend.server.service.score;
+
+import com.tzld.piaoquan.recommend.server.service.score.model.LRModel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import com.tzld.piaoquan.recommend.server.service.score.model.Model;
+
+
+public abstract class BaseLRModelScorer extends AbstractScorer {
+
+    private static Logger LOGGER = LoggerFactory.getLogger(BaseLRModelScorer.class);
+
+    public BaseLRModelScorer(ScorerConfigInfo scorerConfigInfo) {
+        super(scorerConfigInfo);
+    }
+
+    @Override
+    public void loadModel() {
+        doLoadModel(LRModel.class);
+    }
+}

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

@@ -0,0 +1,175 @@
+package com.tzld.piaoquan.recommend.server.service.score;
+
+
+
+import com.tzld.piaoquan.recommend.server.common.base.UserFeature;
+import com.tzld.piaoquan.recommend.server.common.base.VideoRankFeature;
+import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendRequest;
+import com.tzld.piaoquan.recommend.server.model.Video;
+import com.tzld.piaoquan.recommend.server.service.rank.RankParam;
+import com.tzld.piaoquan.recommend.server.service.recall.RecallResult;
+import com.tzld.piaoquan.recommend.server.service.score.feature.VlogShareFeatureExtractor;
+import com.tzld.piaoquan.recommend.server.service.score.model.GBDTModel;
+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.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+
+public class FeedsShareLTRScorer extends BaseGBDTModelScorer{
+
+    private final static int CORE_POOL_SIZE = 64;
+    private final static int TIME_OUT = 150;
+    private final static Logger LOGGER = LoggerFactory.getLogger(FeedsShareLTRScorer.class);
+    private final static ExecutorService executorService = Executors.newFixedThreadPool(CORE_POOL_SIZE);
+
+    public FeedsShareLTRScorer(ScorerConfigInfo configInfo) {
+
+    }
+
+    private RequestContext getRequestContext(RecommendRequest request) {
+
+    }
+
+    @Override
+    public List<Video> scoring(final RecommendRequest recommendRequest,
+                               final RankParam param,
+                               final RecallResult recallResult,
+                               final UserFeature userFeature,
+                               final VideoRankFeature videoFeature) {
+
+        feedsShareScore();
+        Collections.sort();
+        return items;
+    }
+
+
+
+    public void feedsShareScore(final List<Video> items,
+                                       final GBDTModel model,
+                                       final UserFeature user,
+                                       final RecommendRequest requestData) {
+        final int size = items.size();
+        if (size == 0) {
+            return;
+        }
+        final Map<String, Double> userFeatures;
+        final Map<String, Double> contextFeatures;
+
+
+        // context feature
+        contextFeatures = VlogShareFeatureExtractor.extractContextFeatures(requestContext);
+        // user feature
+        userFeatures = VlogShareFeatureExtractor.extractUserFeatures();
+
+        // score item
+        List<Callable<Object>> callables = new ArrayList<Callable<Object>>();
+        for (int index = 0; index < items.size(); index++) {
+            final int fIndex = index;
+            callables.add(new Callable<Object>() {
+                              @Override
+                              public Object call() throws Exception {
+                                  try {
+                                      dwelltimeScore(items.get(fIndex), model, requestContext, userFeatures, contextFeatures, requestData, user);
+                                  } catch (Exception e) {
+                                      LOGGER.error("dwelltime exception: [{}] [{}]", items.get(fIndex).getId(), ExceptionUtils.getFullStackTrace(e));
+                                  }
+                                  return new Object();
+                              }
+                          }
+            );
+        }
+
+        List<Future<Object>> futures = null;
+        try {
+            futures = executorService.invokeAll(callables, TIME_OUT, TimeUnit.MILLISECONDS);
+        } catch (InterruptedException e) {
+            LOGGER.error("execute invoke fail: {}", ExceptionUtils.getFullStackTrace(e));
+        }
+
+        //等待所有请求的结果返回, 超时也返回
+        if (futures != null) {
+            for (Future<Object> future : futures) {
+                try {
+                    if (future != null && future.isDone() && !future.isCancelled() && future.get() != null) {
+                    } else {
+                        LOGGER.debug("Canceled Dwelltime Score {}", requestContext.getId());
+                    }
+                } catch (Exception e) {
+                    LOGGER.error("InterruptedException {},{}", ExceptionUtils.getFullStackTrace(e));
+                }
+            }
+        }
+    }
+
+
+
+
+    // GBDT计算score
+    private void shareScore(final RankerItem item,
+                                final GBDTModel model,
+                                final RequestContext requestContext,
+                                final Map<String, Double> userFeatures,
+                                final Map<String, Double> contextFeatures,
+                                final int debugLevel,
+                                final RecommendRequest requestData, final User user) {
+        try {
+            //judge null
+            if (item == null || item.getItemInfo() == null) {
+                return;
+            }
+
+            ArticleInfo articleInfo = (ArticleInfo) item.getItemInfo();
+
+            Map<String, Double> articleFeatures;
+            double freshness;
+            Map<String, Double> features;
+            articleFeatures = DurationFeatureExtractor.extractArticleFeatures(articleInfo);
+
+
+            if (model != null) {
+                Map<String, Double> featureScoreMap = new HashMap<String, Double>();
+                double pro = model.score(features, featureScoreMap, debugLevel);
+                featureScoreMap.put("TOTAL_SCORE", pro);
+
+                pro = pro > 1 ? pro : 1;
+                LOGGER.debug("xgb score = {}, lgb score = {}", item.getDwelltimeScore(), pro);
+
+                double lgbWeight = Configuration.getDouble("mivideo-recommend-service.dwelltime_ranker.lgb.weight", 1.0);
+                pro = (item.getDwelltimeScore() + lgbWeight * pro) / (1.0 + lgbWeight);
+                item.setDwelltimeScore(pro);
+                double duration = 0;
+                if(item.getItemInfo() != null) {
+                    duration = ((ArticleInfo) item.getItemInfo()).getDuration();
+                }
+                duration = Math.min(Math.max(duration, 1), 1800);
+                item.setScore(predictDwelltimeScore(item.getRecScore(), pro, duration, item.getId(), requestData, user));
+            }
+
+        } catch (Exception e) {
+            LOGGER.error("Exception {},{}", requestContext.getId(), ExceptionUtils.getFullStackTrace(e));
+        }
+    }
+
+
+
+
+
+
+
+
+
+
+
+
+}

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

@@ -0,0 +1,141 @@
+package com.tzld.piaoquan.recommend.server.service.score;
+
+
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigObject;
+import com.typesafe.config.ConfigValue;
+
+import org.apache.commons.lang.exception.ExceptionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+
+public class ScorerConfig {
+
+    private static Logger LOGGER = LoggerFactory.getLogger(ScorerConfig.class);
+    private List<ScorerConfigInfo> configInfoList = new ArrayList<ScorerConfigInfo>();
+
+    public ScorerConfig(Config config) {
+        this.load(config);
+    }
+
+    public ScorerConfig(String configFile) {
+        this.load(configFile);
+    }
+
+    public ScorerConfig() {
+    }
+
+    public static Config mergeConfig(Config baselineConfig, List<Config> expConfigs) {
+        if (expConfigs == null || expConfigs.size() == 0) {
+            return baselineConfig;
+        }
+        Config resultConfig = expConfigs.get(0);
+        for (Config config : expConfigs) {
+            resultConfig = resultConfig.withFallback(config);
+        }
+        resultConfig = resultConfig.withFallback(baselineConfig);
+        return resultConfig;
+    }
+
+    public boolean load(String configFile) {
+        Config config = ConfigFactory.parseResources(configFile);
+        return load(config);
+    }
+
+    public boolean load(Config baselineConfig, List<Config> expConfigs) {
+        Config config = mergeConfig(baselineConfig, expConfigs);
+        return load(config);
+    }
+
+    public boolean load(Config config) {
+        Config scorerConfig = config.getConfig("scorer-config");
+
+        try {
+            loadScorers(scorerConfig);
+            int pos = 0;
+            for (ScorerConfigInfo scorerConfigInfo : configInfoList) {
+                LOGGER.debug("scorer at position [{}], priority [{}], scorer name [{}]",
+                        new Object[]{pos++, scorerConfigInfo.getScorerPriority(), scorerConfigInfo.getScorerName()});
+            }
+            LOGGER.debug("Load scorer config success");
+        } catch (Exception e) {
+            LOGGER.error("Load scorer config failed, [{}]", ExceptionUtils.getFullStackTrace(e));
+            return false;
+        }
+
+        return true;
+    }
+
+    public List<ScorerConfigInfo> getConfigInfoList() {
+        return configInfoList;
+    }
+
+
+    public String loadOptionStringConfig(Config config, String path) {
+        return config.hasPath(path) ? config.getString(path) : null;
+    }
+
+    private Config loadOptionConfig(Config config, String path) {
+        return config.hasPath(path) ? config.getConfig(path) : ConfigFactory.empty();
+    }
+
+    private void loadScorers(Config config) throws Exception {
+
+        ConfigObject confObj = config.root();
+        for (ConfigObject.Entry<String, ConfigValue> it : confObj.entrySet()) {
+            Config conf = ((ConfigObject) it.getValue()).toConfig();
+            // parse config
+            String configName = it.getKey();
+            String scorerName = conf.getString("scorer-name");
+            int scorerPriority = 0;
+            if (conf.hasPath("scorer-priority"))
+                scorerPriority = conf.getInt("scorer-priority");
+            Boolean disableSwitch = false;
+            if (conf.hasPath("disable-switch")) {
+                disableSwitch = conf.getBoolean("disable-switch");
+            }
+            Config paramConfig = loadOptionConfig(conf, "param-config");
+            // model path
+            String modelPath = loadOptionStringConfig(conf, "model-path");
+            if (modelPath == null) {
+                modelPath = loadOptionStringConfig(conf, "default-model-path");
+                LOGGER.debug("model-path is not exists in config file, use default-model-path instead, modelPath={}", modelPath);
+            }
+            // enable queues
+            Set<String> enableQueues = new HashSet<String>();
+            if (conf.hasPath("enable-queues")) {
+                enableQueues.addAll(conf.getStringList("enable-queues"));
+            }
+            ScorerConfigInfo configInfo = new ScorerConfigInfo(configName,
+                    scorerName,
+                    scorerPriority,
+                    disableSwitch,
+                    enableQueues,
+                    modelPath,
+                    paramConfig
+            );
+            LOGGER.debug("parse scorer config info [{}]", configInfo);
+            // add to ConfigInfoList
+            addConfigByPriority(configInfoList, configInfo);
+        }
+    }
+
+    private void addConfigByPriority(List<ScorerConfigInfo> configInfoList, ScorerConfigInfo addConfigInfo) {
+
+        int pos = 0;
+        for (; pos < configInfoList.size(); pos++) {
+            if (configInfoList.get(pos).getScorerPriority() <= addConfigInfo.getScorerPriority()) {
+                break;
+            }
+        }
+
+        configInfoList.add(pos, addConfigInfo);
+    }
+}

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

@@ -0,0 +1,77 @@
+package com.tzld.piaoquan.recommend.server.service.score;
+
+import com.google.gson.Gson;
+import com.typesafe.config.Config;
+import java.util.Set;
+
+
+public class ScorerConfigInfo {
+
+    private String configName;
+    private Integer scorerPriority;
+    private Boolean disableSwitch;
+    private String scorerName;
+    private Set<String> enableQueues;
+    private String modelPath;
+    private Config paramConfig; // param config
+
+    public ScorerConfigInfo(String configName,
+                            String scorerName,
+                            Integer scorerPriority,
+                            Boolean disableSwitch,
+                            Set<String> enableQueues,
+                            String modelPath,
+                            Config paramConfig) {
+
+        this.configName = configName;
+        this.scorerName = scorerName;
+        this.scorerPriority = scorerPriority;
+        this.disableSwitch = disableSwitch;
+        this.enableQueues = enableQueues;
+        this.modelPath = modelPath;
+        this.paramConfig = paramConfig;
+    }
+
+    public Config getParamConfig() {
+        return paramConfig;
+    }
+
+    public Set<String> getEnableQueues() {
+        return enableQueues;
+    }
+
+    public void setEnableQueues(Set<String> enableQueues) {
+        this.enableQueues = enableQueues;
+    }
+
+    public Integer getScorerPriority() {
+        return scorerPriority;
+    }
+
+    public String getScorerName() {
+        return scorerName;
+    }
+
+    public String getConfigName() {
+        return configName;
+    }
+
+    public boolean isQueueEnable(String queueName) {
+        return this.enableQueues == null ||
+                this.enableQueues.isEmpty() ||
+                this.enableQueues.contains(queueName);
+    }
+
+    public String getModelPath() {
+        return modelPath;
+    }
+
+    public Boolean getDisableSwitch() {
+        return disableSwitch;
+    }
+
+    @Override
+    public String toString() {
+        return new Gson().toJson(this);
+    }
+}

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

@@ -0,0 +1,83 @@
+package com.tzld.piaoquan.recommend.server.service.score;
+
+import com.tzld.piaoquan.recommend.server.common.base.UserFeature;
+import com.tzld.piaoquan.recommend.server.common.base.VideoRankFeature;
+import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendRequest;
+import com.tzld.piaoquan.recommend.server.model.Video;
+import com.tzld.piaoquan.recommend.server.service.rank.RankParam;
+import com.tzld.piaoquan.recommend.server.service.rank.RankResult;
+import com.tzld.piaoquan.recommend.server.service.recall.RecallResult;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang.exception.ExceptionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+
+public class ScorerPipeline {
+    public static final int corePoolSize = 128;
+    public static final int SCORE_TIME_OUT = 400;
+    public static final Logger LOGGER = LoggerFactory.getLogger(ScorerPipeline.class);
+    public static final ExecutorService executorService = Executors.newFixedThreadPool(corePoolSize);
+
+    public List<AbstractScorer> scorers;
+
+    public ScorerPipeline(List<AbstractScorer> scorers) {
+        this.scorers = scorers;
+    }
+
+    /**
+     * scoring
+     * @return
+     */
+    public List<Video> scoring(final RecommendRequest recommendRequest,
+                               final RankParam param,
+                               final RecallResult recallResult,
+                               final UserFeature userFeature,
+                               final VideoRankFeature videoFeature
+                               ) {
+        // check recall is empty?
+        if (CollectionUtils.isEmpty(recallResult.getData())) {
+            // log.error
+        }
+
+        if (CollectionUtils.isEmpty(scorers)) {
+            // log.error()
+        }
+        long scoreStart = System.currentTimeMillis();
+        List<Video> items = new ArrayList<Video>();
+        for (final AbstractScorer scorer : scorers) {
+            if (!scorer.isEnable()) {
+                continue;
+            }
+
+            final int beforeSize = items.size();
+            final long startTime = System.currentTimeMillis();
+
+            String fullScorerName = scorer.getScorerConfigInfo().getScorerName();
+            String[] scorerNames = fullScorerName.split("\\.");
+            final String scorerName = scorerNames.length > 0 ? scorerNames[scorerNames.length - 1] : fullScorerName;
+
+            final List<Video> scoreRankerItems = items;
+            Callable<List<Video>> callable = new Callable<List<Video>>() {
+                @Override
+                public List<Video> call() throws Exception {
+                    return scorer.scoring(param);
+                }
+            };
+           //
+            long spentTime = System.currentTimeMillis() - startTime;
+            LOGGER.debug("after scorer [{}], spentTime [{}], before size [{}], remaining size [{}]",
+                    new Object[]{scorerName, spentTime, beforeSize, items.size()});
+        }
+        return items;
+    }
+}

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

@@ -0,0 +1,99 @@
+package com.tzld.piaoquan.recommend.server.service.score;
+
+
+import com.typesafe.config.Config;
+import org.apache.commons.lang.exception.ExceptionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+
+public class ScorerUtils {
+    private static Logger LOGGER = LoggerFactory.getLogger(ScorerUtils.class);
+
+    /**
+     * init load model
+     * @param scorers
+     */
+    public static void initLoadModel(List<AbstractScorer> scorers) {
+        for (AbstractScorer scorer : scorers) {
+            if (scorer.isEnable()) {
+                scorer.loadModel();
+            }
+        }
+    }
+
+    public static void initLoadModel(Config config) {
+        ScorerConfig scorerConfig = new ScorerConfig();
+        scorerConfig.load(config);
+        List<AbstractScorer> scorers = ScorerUtils.constructScorers(scorerConfig);
+        initLoadModel(scorers);
+    }
+
+    public static ScorerPipeline getScorerPipeline(Config mergeConfig) {
+        ScorerConfig scorerConfig = new ScorerConfig();
+        scorerConfig.load(mergeConfig);
+        List<AbstractScorer> scorers = ScorerUtils.constructScorers(scorerConfig);
+        return new ScorerPipeline(scorers);
+    }
+
+    public static ScorerPipeline getScorerPipeline(Config baselineConfig, List<Config> configList) {
+        ScorerConfig scorerConfig = new ScorerConfig();
+        scorerConfig.load(baselineConfig, configList);
+        List<AbstractScorer> scorers = ScorerUtils.constructScorers(scorerConfig);
+        return new ScorerPipeline(scorers);
+    }
+
+    /**
+     * construct scorers
+     *
+     * @param scorerConfig
+     * @return
+     */
+    public static List<AbstractScorer> constructScorers(ScorerConfig scorerConfig) {
+
+        List<AbstractScorer> scorers = new ArrayList<AbstractScorer>();
+        for (ScorerConfigInfo configInfo : scorerConfig.getConfigInfoList()) {
+            if (!configInfo.getDisableSwitch()) {
+                try {
+                    AbstractScorer scorer = (AbstractScorer) Class.forName(configInfo.getScorerName())
+                            .getConstructor(ScorerConfigInfo.class)
+                            .newInstance(configInfo);
+                    scorers.add(scorer);
+                    LOGGER.debug("construct score [{}]", configInfo.getScorerName());
+                } catch (Exception e) {
+                    LOGGER.error("instance scorer {} failed {}", configInfo.getScorerName(), ExceptionUtils.getFullStackTrace(e));
+                }
+
+            }
+        }
+        return scorers;
+    }
+
+    /**
+     * construct scorers
+     *
+     * @param scorerConfig
+     * @return
+     */
+    public static List<AbstractScorer> constructQueueScorers(ScorerConfig scorerConfig, String queueName) {
+
+        List<AbstractScorer> scorers = new ArrayList<AbstractScorer>();
+        for (ScorerConfigInfo configInfo : scorerConfig.getConfigInfoList()) {
+            if (!configInfo.getDisableSwitch() && configInfo.isQueueEnable(queueName)) {
+                try {
+                    AbstractScorer scorer = (AbstractScorer) Class.forName(configInfo.getScorerName())
+                            .getConstructor(ScorerConfigInfo.class)
+                            .newInstance(configInfo);
+                    scorers.add(scorer);
+                    LOGGER.debug("construct queue scorer [{}] [{}]", queueName, configInfo.getScorerName());
+                } catch (Exception e) {
+                    LOGGER.error("instance scorer {} failed {}", configInfo.getScorerName(), ExceptionUtils.getFullStackTrace(e));
+                }
+            }
+        }
+        return scorers;
+    }
+}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/score/feature/BytesGroup.java

@@ -0,0 +1,35 @@
+package com.tzld.piaoquan.recommend.server.service.score.feature;
+
+
+public class BytesGroup {
+    private int id;
+    private String name;
+    private byte[] nameBytes;
+    private byte[] buffer;
+
+    public BytesGroup(int id, String name, byte[] nameBytes) {
+        this.id = id;
+        this.name = name;
+        this.nameBytes = nameBytes;
+    }
+
+    public int getId() {
+        return id;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public byte[] getNameBytes() {
+        return nameBytes;
+    }
+
+    public byte[] getBuffer() {
+        return buffer;
+    }
+
+    public void setBuffer(byte[] buffer) {
+        this.buffer = buffer;
+    }
+}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/score/feature/BytesUtils.java

@@ -0,0 +1,202 @@
+package com.tzld.piaoquan.recommend.server.service.score.feature;
+
+
+import com.tzld.piaoquan.recommend.server.gen.recommend.BaseFeature;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Extract features from user, item & context info. Returns 64-bit murmurhash of feature string as results.
+ */
+public class BytesUtils {
+    private static final byte[] SEPARATOR = "_".getBytes();
+    private static final byte[] FEATURE_SEPARATOR = "#".getBytes();
+
+    private static final int MAX_FEATURE_BYTES_LENGTH = 512;
+    private static final long SEED = 11L;
+
+    private BytesGroup[] groups;
+    private final int debugLevel;
+
+    /**
+     * 一个种特殊的List,在尝试写入null的时候回默默地扔掉.
+     * @param <E> List的元素类型.
+     */
+    public static class NullRejectingArrayList<E> extends ArrayList<E> {
+        public NullRejectingArrayList(int capacity) {
+            super(capacity);
+        }
+
+        public NullRejectingArrayList() {
+            super();
+        }
+
+        @Override
+        public boolean add(E e) {
+            return e != null && super.add(e);
+        }
+    }
+
+    public BytesUtils(BytesGroup[] groups, FeatureUsage usage, int debugLevel) {
+        this.groups = groups;
+        this.debugLevel = debugLevel;
+        for (BytesGroup g : groups) {
+            byte[] buffer = prepareBuffer(g.getName(), g.getNameBytes(), usage);
+            groups[g.getId()].setBuffer(buffer);
+        }
+    }
+
+    public byte[] prepareBuffer(String name, byte[] nameBytes, FeatureUsage usage) {
+        if (usage.disable(name)) {
+            return null;
+        }
+        byte[] buffer = new byte[MAX_FEATURE_BYTES_LENGTH];
+        System.arraycopy(nameBytes, 0, buffer, 0, nameBytes.length);
+        System.arraycopy(FEATURE_SEPARATOR, 0, buffer, nameBytes.length, 1);
+        return buffer;
+    }
+
+    public BaseFeature baseFea(byte[] buffer, int length) {
+        long hash = FeatureHash.MurmurHash64(buffer, 0, length, SEED);
+        String fea = new String(buffer, 0, length);
+        // 初始化protobuf并赋值
+        BaseFeature.Builder tmp = BaseFeature.newBuilder();
+        tmp.setIdentifier(hash);
+        if (debugLevel > 0) {
+            tmp.setFea(fea);
+        }
+        return tmp.build();
+    }
+
+    public BaseFeature makeFea(int id, byte[] value) {
+        byte[] buffer = groups[id].getBuffer();
+        if (buffer == null || value == null) {
+            return null;
+        }
+
+        final int nameLength = groups[id].getNameBytes().length + 1;
+        final int length = nameLength + value.length;
+        System.arraycopy(value, 0, buffer, nameLength, value.length);
+
+        return baseFea(buffer, length);
+    }
+
+    public BaseFeature makeFea(int id, final byte[] p1, final byte[] p2) {
+        byte[] buffer = groups[id].getBuffer();
+        if (buffer == null || p1 == null || p2 == null) {
+            return null;
+        }
+
+        final int nameLength = groups[id].getNameBytes().length + 1;
+        final int length = nameLength + p1.length + 1 + p2.length;
+
+        System.arraycopy(p1, 0, buffer, nameLength, p1.length);
+        System.arraycopy(SEPARATOR, 0, buffer, nameLength + p1.length, 1);
+        System.arraycopy(p2, 0, buffer, nameLength + p1.length + 1, p2.length);
+
+        return baseFea(buffer, length);
+    }
+
+    public BaseFeature makeFea(int id, final byte[] p1, final byte[] p2, final byte[] p3) {
+        byte[] buffer = groups[id].getBuffer();
+        if (buffer == null || p1 == null || p2 == null || p3 == null) {
+            return null;
+        }
+
+        final int nameLength = groups[id].getNameBytes().length + 1;
+        final int length = nameLength + p1.length + 1 + p2.length + 1 + p3.length;
+        System.arraycopy(p1, 0, buffer, nameLength, p1.length);
+        System.arraycopy(SEPARATOR, 0, buffer, nameLength + p1.length, 1);
+        System.arraycopy(p2, 0, buffer, nameLength + p1.length + 1, p2.length);
+        System.arraycopy(SEPARATOR, 0, buffer, nameLength + p1.length + 1 + p2.length, 1);
+        System.arraycopy(p3, 0, buffer, nameLength + p1.length + 1 + p2.length + 1, p3.length);
+
+        return baseFea(buffer, length);
+    }
+
+    public BaseFeature makeFea(int id, final byte[] p1, final byte[] p2, final byte[] p3, final byte[] p4) {
+        byte[] buffer = groups[id].getBuffer();
+        if (buffer == null || p1 == null || p2 == null || p3 == null || p4 == null) {
+            return null;
+        }
+
+        final int nameLength = groups[id].getNameBytes().length + 1;
+        final int length = nameLength + p1.length + 1 + p2.length + 1 + p3.length + 1 + p4.length;
+        System.arraycopy(p1, 0, buffer, nameLength, p1.length);
+        System.arraycopy(SEPARATOR, 0, buffer, nameLength + p1.length, 1);
+        System.arraycopy(p2, 0, buffer, nameLength + p1.length + 1, p2.length);
+        System.arraycopy(SEPARATOR, 0, buffer, nameLength + p1.length + 1 + p2.length, 1);
+        System.arraycopy(p3, 0, buffer, nameLength + p1.length + 1 + p2.length + 1, p3.length);
+        System.arraycopy(SEPARATOR, 0, buffer, nameLength + p1.length + 1 + p2.length + 1 + p3.length, 1);
+        System.arraycopy(p4, 0, buffer, nameLength + p1.length + 1 + p2.length + 1 + p3.length + 1, p4.length);
+
+        return baseFea(buffer, length);
+    }
+
+    public List<BaseFeature> makeFea(int id, byte[][] list) {
+        List<BaseFeature> result = new NullRejectingArrayList<BaseFeature>(list.length);
+        for (byte[] t: list) {
+            result.add(makeFea(id, t));
+        }
+        return result;
+    }
+
+    public List<BaseFeature> makeFea(int id, byte[][] left, byte[] right) {
+        List<BaseFeature> result = new NullRejectingArrayList<BaseFeature>(left.length);
+        for (byte[] l: left) {
+            result.add(makeFea(id, l, right));
+        }
+        return result;
+    }
+
+    public List<BaseFeature> makeFea(int id, byte[][] left, byte[] right1, byte[] right2) {
+        List<BaseFeature> result = new NullRejectingArrayList<BaseFeature>(left.length);
+        for (byte[] l: left) {
+            result.add(makeFea(id, l, right1, right2));
+        }
+        return result;
+    }
+
+    public List<BaseFeature> makeFea(int id, byte[][] left, byte[] right1, byte[] right2, byte[] right3) {
+        List<BaseFeature> result = new NullRejectingArrayList<BaseFeature>(left.length);
+        for (byte[] l: left) {
+            result.add(makeFea(id, l, right1, right2, right3));
+        }
+        return result;
+    }
+
+    public List<BaseFeature> makeFea(int id, byte[] left, byte[][] right) {
+        List<BaseFeature> result = new NullRejectingArrayList<BaseFeature>(right.length);
+        for (byte[] r : right) {
+            result.add(makeFea(id, left, r));
+        }
+        return result;
+    }
+
+    public List<BaseFeature> makeFea(int id, byte[] left1, byte[] left2, byte[][] right) {
+        List<BaseFeature> result = new NullRejectingArrayList<BaseFeature>(right.length);
+        for (byte[] r : right) {
+            result.add(makeFea(id, left1, left2, r));
+        }
+        return result;
+    }
+
+    public List<BaseFeature> makeFea(int id, byte[] left1, byte[] left2, byte[] left3, byte[][] right) {
+        List<BaseFeature> result = new NullRejectingArrayList<BaseFeature>(right.length);
+        for (byte[] r : right) {
+            result.add(makeFea(id, left1, left2, left3, r));
+        }
+        return result;
+    }
+
+    public List<BaseFeature> makeFea(int id, byte[][] left, byte[][] right) {
+        List<BaseFeature> result = new NullRejectingArrayList<BaseFeature>(left.length * right.length);
+        for (byte[] l: left) {
+            for (byte[] r: right) {
+                result.add(makeFea(id, l, r));
+            }
+        }
+        return result;
+    }
+}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/score/feature/FeatureHash.java

@@ -0,0 +1,230 @@
+package com.tzld.piaoquan.recommend.server.service.score.feature;
+
+
+import java.math.BigInteger;
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+
+public class FeatureHash {
+    public static Charset CharSetUTF8 = Charset.forName("UTF-8");
+
+    public static long getUInt32(byte a, byte b, byte c, byte d) {
+        return (d << 24 | (c & 0xFF) << 16 | (b & 0xFF) << 8 | (a & 0xFF));
+    }
+
+    public static long hash64(byte[] data) {
+        return MurmurHash64A(ByteBuffer.wrap(data), 0, data.length, 11L);
+    }
+
+    public static long MurmurHash64A(ByteBuffer buffer, int from, int len, long seed) {
+        final long m = 0xc6a4a7935bd1e995L;
+        final int r = 47;
+
+        long h = (seed) ^ (len * m);
+        int longLength = len / 8;
+
+        for (int i = 0; i < longLength; ++i) {
+            final int bytePos = from + i * 8;
+            long k = buffer.getLong(bytePos);
+
+            k *= m;
+            k ^= k >> r;
+            k *= m;
+            h ^= k;
+            h *= m;
+        }
+
+        final int remainingPos = len & ~7;
+        switch (len % 8) {
+            case 7: h ^= (long)(buffer.get(remainingPos + 6) & 0xFF) << 48;
+            case 6: h ^= (long)(buffer.get(remainingPos + 5) & 0xFF) << 40;
+            case 5: h ^= (long)(buffer.get(remainingPos + 4) & 0xFF) << 32;
+            case 4: h ^= (long)(buffer.get(remainingPos + 3) & 0xFF) << 24;
+            case 3: h ^= (long)(buffer.get(remainingPos + 2) & 0xFF) << 16;
+            case 2: h ^= (long)(buffer.get(remainingPos + 1) & 0xFF) << 8;
+            case 1:
+                h ^= (long)(buffer.get(remainingPos) & 0xFF);
+                h *= m;
+        }
+
+        h ^= h >>> r;
+        h *= m;
+        h ^= h >>> r;
+        return h;
+    }
+
+    public static long MurmurHash32(byte data[], int len, long seed) {
+        long m = 0x5bd1e995L;
+        int r = 24;
+
+        long h = seed ^ len;
+
+        int offset = 0;
+        while (len >= 4) {
+            long k = getUInt32(data[offset], data[offset + 1], data[offset + 2], data[offset + 3]);
+
+            k *= m;
+            k &= 0xFFFFFFFFL;
+            k ^= k >> r;
+            k *= m;
+            k &= 0xFFFFFFFFL;
+
+            h *= m;
+            h &= 0xFFFFFFFFL;
+            h ^= k;
+
+            offset += 4;
+            len -= 4;
+        }
+
+        // Handle the last few bytes of the input array
+        switch (len) {
+            case 3: h ^= data[offset + 2] << 16;
+            case 2: h ^= data[offset + 1] << 8;
+            case 1: h ^= data[offset];
+                h *= m;
+                h &= 0xFFFFFFFFL;
+        } ;
+
+        // Do a few final mixes of the hash to ensure the last few
+        // bytes are well-incorporated.
+
+        h ^= h >> 13;
+        h *= m;
+        h &= 0xFFFFFFFFL;
+        h ^= h >> 15;
+
+        return h;
+    }
+
+    // 64-bit hash for 32-bit platforms
+    public static long MurmurHash64(byte[] buffer, int start, int len, long seed) {
+        final long m = 0x5bd1e995L;
+        final int r = 24;
+        final int original = len;
+
+        long h1 = (seed) ^ len;
+        long h2 = (seed >> 32);
+
+        int offset = start;
+        while (len >= 8) {
+            long k1 = getUInt32(buffer[offset], buffer[offset + 1], buffer[offset + 2], buffer[offset + 3]);
+            // long k1 = buffer.getInt(offset);
+
+            k1 *= m; k1 &= 0xFFFFFFFFL; k1 ^= k1 >> r; k1 *= m; k1 &= 0xFFFFFFFFL;
+            h1 *= m; h1 &= 0xFFFFFFFFL; h1 ^= k1;
+            offset += 4;
+
+            long k2 = getUInt32(buffer[offset], buffer[offset + 1], buffer[offset + 2], buffer[offset + 3]);
+            // long k2 = buffer.getInt(offset);
+            k2 *= m; k2 &= 0xFFFFFFFFL; k2 ^= k2 >> r; k2 *= m; k2 &= 0xFFFFFFFFL;
+            h2 *= m; h2 &= 0xFFFFFFFFL; h2 ^= k2;
+
+            offset += 4;
+            len -= 8;
+        }
+
+        if (len >= 4) {
+            long k1 = getUInt32(buffer[offset], buffer[offset + 1], buffer[offset + 2], buffer[offset + 3]);
+            // long k1 = buffer.getInt(offset);
+            k1 *= m; k1 &= 0xFFFFFFFFL; k1 ^= k1 >> r; k1 *= m; k1 &= 0xFFFFFFFFL;
+            h1 *= m; h1 &= 0xFFFFFFFFL; h1 ^= k1;
+            offset += 4;
+            len -= 4;
+        }
+
+        switch (len) {
+            case 3: h2 ^= (buffer[offset + 2] & 0xFF) << 16;
+            case 2: h2 ^= (buffer[offset + 1] & 0xFF) << 8;
+            case 1: h2 ^= (buffer[offset] & 0xFF);
+                h2 *= m;
+                h2 &= 0xFFFFFFFFL;
+        } ;
+
+        h1 ^= h2 >> 18;
+        h1 *= m; h1 &= 0xFFFFFFFFL;
+        h2 ^= h1 >> 22;
+        h2 *= m; h2 &= 0xFFFFFFFFL;
+        h1 ^= h2 >> 17;
+        h1 *= m; h1 &= 0xFFFFFFFFL;
+        h2 ^= h1 >> 19;
+        h2 *= m; h2 &= 0xFFFFFFFFL;
+
+        /*BigInteger ans = BigInteger.valueOf(h1).shiftLeft(32).or(BigInteger.valueOf(h2));
+        return ans.longValue();*/
+        //System.err.println("feature: " + new String(buffer, 0, original) + " length: " + original + " hash: " + (h1 << 32 | h2) + " daze");
+        return h1 << 32 | h2;
+    }
+
+    // 64-bit hash for 32-bit platforms
+    public static BigInteger MurmurHash64(byte data[], int len, long seed) {
+        long m = 0x5bd1e995L;
+        int r = 24;
+
+        long h1 = (seed) ^ len;
+        long h2 = (seed >> 32);
+
+        int offset = 0;
+        while (len >= 8) {
+            long k1 = getUInt32(data[offset], data[offset + 1], data[offset + 2], data[offset + 3]);
+            k1 *= m; k1 &= 0xFFFFFFFFL; k1 ^= k1 >> r; k1 *= m; k1 &= 0xFFFFFFFFL;
+            h1 *= m; h1 &= 0xFFFFFFFFL; h1 ^= k1;
+
+            long k2 = getUInt32(data[offset + 4], data[offset + 5], data[offset + 6], data[offset + 7]);
+            k2 *= m; k2 &= 0xFFFFFFFFL; k2 ^= k2 >> r; k2 *= m; k2 &= 0xFFFFFFFFL;
+            h2 *= m; h2 &= 0xFFFFFFFFL; h2 ^= k2;
+
+            offset += 8;
+            len -= 8;
+        }
+
+        if (len >= 4) {
+            long k1 = getUInt32(data[offset], data[offset + 1], data[offset + 2], data[offset + 3]);
+            k1 *= m; k1 &= 0xFFFFFFFFL; k1 ^= k1 >> r; k1 *= m; k1 &= 0xFFFFFFFFL;
+            h1 *= m; h1 &= 0xFFFFFFFFL; h1 ^= k1;
+            offset += 4;
+            len -= 4;
+        }
+
+        switch (len) {
+            case 3: h2 ^= (data[offset + 2] & 0xFF) << 16;
+            case 2: h2 ^= (data[offset + 1] & 0xFF) << 8;
+            case 1: h2 ^= (data[offset] & 0xFF);
+                h2 *= m;
+                h2 &= 0xFFFFFFFFL;
+        } ;
+
+        h1 ^= h2 >> 18;
+        h1 *= m; h1 &= 0xFFFFFFFFL;
+        h2 ^= h1 >> 22;
+        h2 *= m; h2 &= 0xFFFFFFFFL;
+        h1 ^= h2 >> 17;
+        h1 *= m; h1 &= 0xFFFFFFFFL;
+        h2 ^= h1 >> 19;
+        h2 *= m; h2 &= 0xFFFFFFFFL;
+
+        BigInteger ans = BigInteger.valueOf(h1).shiftLeft(32).or(BigInteger.valueOf(h2));
+        return ans;
+    }
+
+    public static String hash(String input) {
+        byte[] tt = input.getBytes(CharSetUTF8);
+        return MurmurHash64(tt, tt.length, 11L).toString();
+    }
+
+    public static Long hashToLong(String input) {
+        byte[] tt = input.getBytes(CharSetUTF8);
+        return MurmurHash64(tt, tt.length, 11L).longValue();
+    }
+
+    /** the constant 2^64 */
+    private static final BigInteger TWO_64 = BigInteger.ONE.shiftLeft(64);
+
+    public static String asUnsignedLongString(long l) {
+        BigInteger b = BigInteger.valueOf(l);
+        if (b.signum() < 0) {
+            b = b.add(TWO_64);
+        }
+        return b.toString();
+    }
+}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/score/feature/FeatureUsage.java

@@ -0,0 +1,43 @@
+package com.tzld.piaoquan.recommend.server.service.score.feature;
+
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigObject;
+import com.typesafe.config.ConfigValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.HashSet;
+import java.util.Set;
+
+public class FeatureUsage implements Serializable {
+    private final Logger LOGGER = LoggerFactory.getLogger(FeatureUsage.class);
+    private Set<String> hash;
+
+    public FeatureUsage() {
+        hash = new HashSet<String>();
+    }
+
+    public FeatureUsage(String confName) {
+        // note that these fields are NOT lazy, because if we're going to get any exceptions, we want to get them on startup.
+        Config config = ConfigFactory.load(confName);
+        config.checkValid(config, "features");
+        ConfigObject conf = config.getConfig("features").root();
+        LOGGER.info("create_feature_usage: " + confName);
+        hash = new HashSet<String>();
+        for (ConfigObject.Entry<String, ConfigValue> it : conf.entrySet()) {
+            String key = it.getKey();
+            Config cf = ((ConfigObject) it.getValue()).toConfig();
+            LOGGER.info("key: " + key + ", value: " + cf.toString());
+            boolean disable = cf.getBoolean("disable");
+            if (disable) {
+                hash.add(key);
+            }
+        }
+    }
+
+    public boolean disable(String fea) {
+        return hash.contains(fea);
+    }
+}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/score/feature/LRFeatureExtractorBase.java

@@ -0,0 +1,77 @@
+package com.tzld.piaoquan.recommend.server.service.score.feature;
+
+import com.google.common.collect.ArrayListMultimap;
+import com.google.common.collect.ListMultimap;
+
+import com.tzld.piaoquan.recommend.server.common.base.UserFeature;
+import com.tzld.piaoquan.recommend.server.common.base.VideoRankFeature;
+import com.tzld.piaoquan.recommend.server.common.enums.VlogFeatureGroup;
+import com.tzld.piaoquan.recommend.server.gen.recommend.GroupedFeature;
+import com.tzld.piaoquan.recommend.server.gen.recommend.LRSamples;
+import com.tzld.piaoquan.recommend.server.gen.recommend.BaseFeature;
+import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendRequest;
+import com.tzld.piaoquan.recommend.server.gen.recommend.FeatureGroup;
+import com.tzld.piaoquan.recommend.server.service.recall.RecallResult;
+import lombok.Synchronized;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+
+
+public abstract class LRFeatureExtractorBase {
+    private static final double DEFAULT_USER_CTR_GROUP = 10.0;
+    private static final double DEFAULT_ARTICLE_CTR_GROUP = 100.0;
+
+    double userCtrGroup = 10.0;
+    double videoCtrGroup = 100.0;
+
+    private BytesUtils utils;
+    ListMultimap<FeatureGroup, BaseFeature> features = ArrayListMultimap.create();
+    int groupCount;
+    LRFeatureExtractorBase() {
+    };
+
+    LRFeatureExtractorBase(FeatureUsage usage) {
+        this(usage, DEFAULT_USER_CTR_GROUP, DEFAULT_ARTICLE_CTR_GROUP);
+    }
+
+    LRFeatureExtractorBase(FeatureUsage usage, double userCtrGroup, double videoCtrGroup) {
+        this.userCtrGroup = userCtrGroup;
+        this.videoCtrGroup = videoCtrGroup;
+        groupCount = VlogFeatureGroup.values().length;
+        BytesGroup[] groups = new BytesGroup[groupCount];
+        for (VlogFeatureGroup g: VlogFeatureGroup.values()) {
+            groups[g.ordinal()] = new BytesGroup(g.ordinal(),
+                    g.getGroupName(), g.getGroupNameBytes());
+        }
+        utils = new BytesUtils(groups, usage, 0);
+    }
+
+    private FeatureGroup makeGroup(VlogFeatureGroup group){
+        FeatureGroup.Builder g = FeatureGroup.newBuilder();
+        g.setType("1");
+        g.setName(group.getGroupName());
+        g.setId(group.ordinal());
+        return g.build();
+    };
+
+
+    void makeFea(VlogFeatureGroup group, byte[] value) {
+        FeatureGroup g = makeGroup(group);
+        BaseFeature feature = utils.makeFea(group.ordinal(), value);
+        features.put(g, feature);
+    }
+
+    void makeFea(VlogFeatureGroup group, byte[][] list) {
+        FeatureGroup g = makeGroup(group);
+        List<BaseFeature> featureList = utils.makeFea(group.ordinal(), list);
+        features.putAll(g, featureList);
+    }
+
+    public abstract LRSamples single(RecommendRequest request, UserFeature user, VideoRankFeature itemFeature);
+
+
+}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/score/feature/VlogShareFeatureExtractor.java

@@ -0,0 +1,77 @@
+package com.tzld.piaoquan.recommend.server.service.score.feature;
+
+import com.tzld.piaoquan.recommend.server.common.base.UserFeature;
+import com.tzld.piaoquan.recommend.server.common.base.VideoRankFeature;
+import com.tzld.piaoquan.recommend.server.common.enums.VlogFeatureGroup;
+import com.tzld.piaoquan.recommend.server.gen.recommend.BaseFeature;
+import com.tzld.piaoquan.recommend.server.gen.recommend.GroupedFeature;
+import com.tzld.piaoquan.recommend.server.gen.recommend.LRSamples;
+import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendRequest;
+import com.tzld.piaoquan.recommend.server.model.Video;
+import com.tzld.piaoquan.recommend.server.service.recall.RecallResult;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class VlogShareFeatureExtractor extends LRFeatureExtractorBase {
+
+    public VlogShareFeatureExtractor() {
+        super();
+    }
+    public VlogShareFeatureExtractor(FeatureUsage usage) {
+        super(usage);
+    }
+    public VlogShareFeatureExtractor(FeatureUsage usage, double userCtrGroup, double videoCtrGroup, int debugLevel) {
+        super(usage, userCtrGroup, videoCtrGroup);
+    }
+
+    private void getUserFeatures(UserFeature user) {
+        makeFea(VlogFeatureGroup.SEX, user.sex);
+    }
+
+    private void getContextFeature(RecommendRequest request) {
+        makeFea();
+    }
+
+
+    private void getItemFeatures(VideoRankFeature item) {
+        makeFea(VlogFeatureGroup.VIDEOID, item.videoid);
+    }
+
+
+    public LRSamples single(RecommendRequest request, UserFeature user, VideoRankFeature itemfeature) {
+        features.clear();
+        //
+        getUserFeatures(user);
+        getContextFeature(request);
+        getItemFeatures(itemfeature);
+
+        LRSamples.Builder lr =  com.tzld.piaoquan.recommend.server.gen.recommend.LRSamples.newBuilder();
+        lr.setGroupNum(groupCount);
+        List<com.tzld.piaoquan.recommend.server.gen.recommend.FeatureGroup> keys = new ArrayList<>(features.keySet());
+        int count = 0;
+        for(com.tzld.piaoquan.recommend.server.gen.recommend.FeatureGroup group : keys) {
+            List<BaseFeature> fea = features.get(group);
+            GroupedFeature.Builder gf = GroupedFeature.newBuilder();
+            gf.setGroup(group);
+            gf.setCount(fea.size());
+            gf.addAllFeatures(fea);
+            count += fea.size();
+            lr.addFeatures(gf);
+        }
+        lr.setCount(count);
+        return lr.build();
+    }
+
+
+
+
+
+
+
+
+
+
+
+
+}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/score/model/GBDTModel.java

@@ -0,0 +1,223 @@
+package com.tzld.piaoquan.recommend.server.service.score.model;
+
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Created by liulixiang on 4/27/16.
+ */
+public class GBDTModel extends Model {
+    private static final Logger LOGGER = LoggerFactory.getLogger(GBDTModel.class);
+    private static final float LINEAR_TRANSFORM_LOWER_BOUND = 0f;
+    private static final float LINEAR_TRANSFORM_SLOPE = 1f / 1f;
+    private int featureCount = 0;
+    private Map<String, Integer> featureIdMap = null;
+    private List<HashMap<Integer, Node>> boosterModel = null;
+
+    private float transform(final float score) {
+        return (score - LINEAR_TRANSFORM_LOWER_BOUND) * LINEAR_TRANSFORM_SLOPE;
+    }
+
+    private int getFeatureId(final String feature) {
+        return featureIdMap.containsKey(feature) ? featureIdMap.get(feature) : -1;
+    }
+
+    @Override
+    public int getModelSize() {
+        return boosterModel.size();
+    }
+
+    @Override
+    public boolean loadFromStream(InputStreamReader in) throws Exception {
+        List<HashMap<Integer, Node>> model = new ArrayList<HashMap<Integer, Node>>();
+        HashMap<Integer, Node> tree = null;
+
+        featureIdMap = new HashMap<String, Integer>();
+        featureCount = 0;
+
+        BufferedReader input = new BufferedReader(in);
+        String line;
+        while ((line = input.readLine()) != null) {
+            String[] tokens = line.trim().split(":");
+            if (tokens.length == 1) {
+                if (tokens[0].startsWith("booster")) {
+                    if (tree != null && tree.size() > 0) {
+                        model.add(tree);
+                    }
+
+                    tree = new HashMap<Integer, Node>();
+                }
+            } else if (tokens.length == 2) {
+                Node node = new Node();
+
+                Integer id = Integer.parseInt(tokens[0]);
+
+                String[] items = tokens[1].split(" ");
+                if (items.length == 1) {
+                    node.isLeaf = true;
+
+                    String[] dt = items[0].split("=");
+                    node.value = Float.parseFloat(dt[1]);
+                } else if (items.length == 2) {
+                    node.isLeaf = false;
+                    String[] fieldDescriptions = null;
+                    node.compareType = NodeCompareType.LT;
+                    if (items[0].substring(1, items[0].length() - 1).contains("<=")) {
+                        fieldDescriptions = items[0].substring(1, items[0].length() - 1).split("<=");
+                        node.compareType = NodeCompareType.LE;
+                    } else {
+                        fieldDescriptions = items[0].substring(1, items[0].length() - 1).split("<");
+                    }
+                    // feature to id
+                    int featureId = getFeatureId(fieldDescriptions[0]);
+                    if (featureId < 0) {
+                        featureIdMap.put(fieldDescriptions[0], featureCount);
+                        featureId = featureCount;
+                        featureCount += 1;
+                    }
+                    node.splitFeatureId = featureId;
+
+                    if (fieldDescriptions.length == 1) {
+                        node.fieldType = NodeFieldType.BINARY;
+                    } else {
+                        node.fieldType = NodeFieldType.QUANTITATIVE;
+                        node.splitCondition = Float.parseFloat(fieldDescriptions[1]);
+                    }
+
+                    String[] childrenDescriptions = items[1].split(",");
+                    for (String childDescription : childrenDescriptions) {
+                        String[] descs = childDescription.split("=");
+                        if (descs[0].equals("yes")) {
+                            node.yes = Integer.parseInt(descs[1]);
+                        } else if (descs[0].equals("no")) {
+                            node.no = Integer.parseInt(descs[1]);
+                        } else if (descs[0].equals("missing")) {
+                            node.missing = Integer.parseInt(descs[1]);
+                        }
+                    }
+                }
+                tree.put(id, node);
+            }
+        }
+        if (tree != null && !tree.isEmpty()) {
+            model.add(tree);
+        }
+
+        LOGGER.info("Boosted tree model load over and tree number is " + model.size());
+        input.close();
+        in.close();
+        if (model != null && model.size() > 0) {
+            boosterModel = model;
+            return true;
+        } else {
+            return false;
+        }
+    }
+
+    @Override
+    public boolean loadPartitions(String modelPath, boolean isRegister) {
+        return false;
+    }
+
+    private float score(final Map<Integer, Node> tree, final boolean[] featureIsExists, final Float[] featureValues) {
+        int id = 0;
+        while (true) {
+            Node currentNode = tree.get(id);
+            if (currentNode.isLeaf) {
+                return currentNode.value;
+            }
+
+            if (currentNode.fieldType == NodeFieldType.BINARY) {
+                if (!featureIsExists[currentNode.splitFeatureId]) {
+                    id = currentNode.no;
+                } else {
+                    id = currentNode.yes;
+                }
+            } else if (currentNode.fieldType == NodeFieldType.QUANTITATIVE) {
+                if (!featureIsExists[currentNode.splitFeatureId]) {
+                    id = currentNode.missing;
+                } else {
+                    Float value = featureValues[currentNode.splitFeatureId];
+
+                    if ((currentNode.compareType == NodeCompareType.LT && value < currentNode.splitCondition)
+                            || (currentNode.compareType == NodeCompareType.LE && value <= currentNode.splitCondition)) {
+                        id = currentNode.yes;
+                    } else {
+                        id = currentNode.no;
+                    }
+                }
+            } else {
+                LOGGER.error("Reach undefined condition: {}", id);
+            }
+        }
+    }
+
+    public Float score(final Map<String, Double> features, Map<String, Double> featuresScore, final int debugLevel) {
+        Float result = 0f;
+
+        boolean[] featureIsExists = new boolean[featureCount];
+        Float[] featureValues = new Float[featureCount];
+        for (Map.Entry<String, Double> entry : features.entrySet()) {
+            int featureId = getFeatureId(entry.getKey());
+            if (featureId >= 0) {
+                featureIsExists[featureId] = true;
+                featureValues[featureId] = entry.getValue().floatValue();
+            }
+        }
+
+        for (Map<Integer, Node> tree : boosterModel) {
+            result += score(tree, featureIsExists, featureValues);
+        }
+
+        Float transformedResult = transform(result);
+
+        if (debugLevel >= 1) {
+            if (null == featuresScore) {
+                featuresScore = new HashMap<String, Double>();
+            }
+
+            for (Map.Entry<String, Double> feature : features.entrySet()) {
+                featuresScore.put(feature.getKey(), feature.getValue());
+            }
+            int existFeatureCount = 0;
+            for (boolean isExisit : featureIsExists) {
+                existFeatureCount += isExisit ? 1 : 0;
+            }
+            featuresScore.put("COUNT", (double) existFeatureCount);
+            featuresScore.put("PREDICTED", (double) result);
+            featuresScore.put("TOTAL", (double) transformedResult);
+        }
+
+        LOGGER.debug("[calc_dwelltime]features: " + Arrays.toString(features.entrySet().toArray()) + ", prediction:" + result + " , transformed:" + transformedResult);
+
+        return transformedResult;
+    }
+
+    private enum NodeFieldType {
+        BINARY, QUANTITATIVE
+    }
+
+    private enum NodeCompareType {
+        LT, LE
+    }
+
+    private class Node {
+
+        boolean isLeaf;
+        float value;
+        NodeFieldType fieldType;
+        NodeCompareType compareType;
+        int splitFeatureId;
+        Float splitCondition;
+        Integer yes, no, missing;
+    }
+}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/score/model/LRModel.java

@@ -0,0 +1,419 @@
+package com.tzld.piaoquan.recommend.server.service.score.model;
+
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+
+import it.unimi.dsi.fastutil.longs.Long2FloatMap;
+import it.unimi.dsi.fastutil.longs.Long2FloatOpenHashMap;
+import com.tzld.piaoquan.recommend.server.gen.recommend.LRSamples;
+import com.tzld.piaoquan.recommend.server.gen.recommend.GroupedFeature;
+import com.tzld.piaoquan.recommend.server.gen.recommend.BaseFeature;
+
+import org.apache.commons.lang.exception.ExceptionUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+public class LRModel extends Model {
+    protected static final int MODEL_FIRST_LOAD_COUNT = 1 << 25; // 32M
+    private static final Logger LOGGER = LoggerFactory.getLogger(LRModel.class);
+    private static final ExecutorService executorService = Executors.newCachedThreadPool();
+    private final int bucketBits = 10;  // power(2, 10) => 1024 个槽位
+    private List<Long2FloatMap> lrModel;
+    private Config config = ConfigFactory.load("hdfs_conf.properties");
+    private Configuration hdfsConf = new Configuration();
+
+    public LRModel() {
+        //配置不同环境的hdfs conf
+        String coreSiteFile = config.hasPath("hdfs.coreSiteFile") ? StringUtils.trim(config.getString("hdfs.coreSiteFile")) : "core-site.xml";
+        String hdfsSiteFile = config.hasPath("hdfs.hdfsSiteFile") ? StringUtils.trim(config.getString("hdfs.hdfsSiteFile")) : "hdfs-site.xml";
+        hdfsConf.addResource(coreSiteFile);
+        hdfsConf.addResource(hdfsSiteFile);
+        this.lrModel = constructModel();
+    }
+
+    public List<Long2FloatMap> getLrModel() {
+        return lrModel;
+    }
+
+    public List<Long2FloatMap> constructModel() {
+        List<Long2FloatMap> initModel = new ArrayList<Long2FloatMap>();
+        int buckets = (int) Math.pow(2, bucketBits);
+        for (int i = 0; i < buckets; i++) {
+            Long2FloatMap internalModel = new Long2FloatOpenHashMap();
+            internalModel.defaultReturnValue(0.0f);
+            initModel.add(internalModel);
+        }
+        return initModel;
+    }
+
+    public int getBucket(long featureHash) {
+        return (int) (((featureHash >> bucketBits) << bucketBits) ^ featureHash);
+    }
+
+    public void putFeature(List<Long2FloatMap> model, long featureHash, float weight) {
+        model.get(getBucket(featureHash)).put(featureHash, weight);
+    }
+
+    public float getWeight(List<Long2FloatMap> model, long featureHash) {
+        return model.get(getBucket(featureHash)).get(featureHash);
+    }
+
+    @Override
+    public int getModelSize() {
+        if (this.lrModel == null)
+            return 0;
+        int sum = 0;
+        for (Map<Long, Float> model : this.lrModel) {
+            sum += model.size();
+        }
+        return sum;
+    }
+
+    public void cleanModel() {
+        this.lrModel = null;
+    }
+
+    public Float score(LRSamples lrSamples) {
+        float sum = 0.0f;
+        for(int i=0; i< lrSamples.getFeaturesCount(); i++)
+        {
+            GroupedFeature gf = lrSamples.getFeatures(i);
+            if (gf != null && gf.getFeatures(i) != null) {
+                for(int j=0; j < gf.getFeaturesCount(); j++) {
+                    BaseFeature fea = gf.getFeatures(j);
+                    if (fea != null) {
+                        float tmp = getWeight(this.lrModel, fea.getIdentifier());
+                        fea.toBuilder().setWeight(tmp);
+                        sum += tmp;
+                    }
+
+                }
+            }
+        }
+
+        float pro = (float) (1.0f / (1 + Math.exp(-sum)));
+        lrSamples.toBuilder().setPredictCtr(pro);
+        return pro;
+    }
+
+    public Float getWeights(LRSamples lrSamples) {
+        float sum = 0.0f;
+
+        for(int i=0; i< lrSamples.getFeaturesCount(); i++)
+        {
+            GroupedFeature gf = lrSamples.getFeatures(i);
+            if (gf != null && gf.getFeatures(i) != null) {
+                for(int j=0; j < gf.getFeaturesCount(); j++) {
+                    BaseFeature fea = gf.getFeatures(j);
+                    if (fea != null) {
+                        float tmp = getWeight(this.lrModel, fea.getIdentifier());
+                        fea.toBuilder().setWeight(tmp);
+                        sum += tmp;
+                    }
+                }
+            }
+        }
+        lrSamples.toBuilder().setWeight(sum);
+        return sum;
+    }
+
+    /**
+     * 目前模型比较大,分两个阶段load模型
+     * (1). load 8M 模型, 并更新;
+     * (2). load 剩余的模型
+     * 中间提供一段时间有损的打分服务
+     *
+     * @param in
+     * @return
+     * @throws IOException
+     */
+    @Override
+    public boolean loadFromStream(InputStreamReader in) throws IOException {
+
+        List<Long2FloatMap> model = constructModel();
+        BufferedReader input = new BufferedReader(in);
+        String line = null;
+        int cnt = 0;
+
+        Integer curTime = new Long(System.currentTimeMillis() / 1000).intValue();
+        LOGGER.info("[MODELLOAD] before model load, key size: {}, current time: {}", lrModel.size(), curTime);
+        //first stage
+        while ((line = input.readLine()) != null) {
+            String[] items = line.split("\t");
+            if (items.length < 2) {
+                continue;
+            }
+
+            putFeature(model, new BigInteger(items[0]).longValue(), Float.valueOf(items[1]).floatValue());
+            if (cnt++ < 10) {
+                LOGGER.debug("fea: " + items[0] + ", weight: " + items[1]);
+            }
+            if (cnt > MODEL_FIRST_LOAD_COUNT) {
+                break;
+            }
+        }
+        //model update
+        this.lrModel = model;
+        LOGGER.info("[MODELLOAD] after first stage model load, key size: {}, current time: {}", lrModel.size(), curTime);
+        //final stage
+        while ((line = input.readLine()) != null) {
+            String[] items = line.split("\t");
+            if (items.length < 2) {
+                continue;
+            }
+            putFeature(model, new BigInteger(items[0]).longValue(), Float.valueOf(items[1]).floatValue());
+        }
+        LOGGER.info("[MODELLOAD] after model load, key size: {}, current time: {}", lrModel.size(), curTime);
+
+        LOGGER.info("[MODELLOAD] model load over and size " + cnt);
+        input.close();
+        in.close();
+        return true;
+    }
+
+    @Override
+    public boolean loadPartitions(final String modelPath, boolean isRegister) {
+        if (isRegister) {
+            return loadPartitionsParallel(modelPath);
+        } else {
+            return loadPartitionsSingle(modelPath);
+        }
+    }
+
+    /**
+     * 程序运行的过程中, model 太大, 为了尽可能防止full gc, 单线程加载
+     *
+     * @param modelPath
+     * @return
+     */
+
+    public boolean loadPartitionsSingle(final String modelPath) {
+        try {
+            Path path = new Path(modelPath);
+            hdfsConf.setBoolean("fs.hdfs.impl.disable.cache", true);
+
+            FileSystem fs = path.getFileSystem(hdfsConf);
+            FileStatus[] listStatus = fs.listStatus(path);
+
+            //judge null and empty
+            if (listStatus == null || listStatus.length == 0) {
+                LOGGER.error("model path is dir, but hdfs patition path is null");
+                return false;
+            }
+
+            long startTime = System.currentTimeMillis();
+
+            // 初始化model大小,直接分配到old heap
+            List<Long2FloatMap> currLrModel = constructModel();
+
+            //multi thread load hdfs news info files
+            int failedPartitionNum = 0;
+            int partitionsNum = listStatus.length;
+            for (final FileStatus file : listStatus) {
+                String absPath = String.format("%s/%s", modelPath, file.getPath().getName());
+                InputStreamReader fin = null;
+                try {
+                    Path tmpPath = new Path(absPath);
+                    FileSystem tmpFs = tmpPath.getFileSystem(hdfsConf);
+                    FSDataInputStream inputStream = tmpFs.open(tmpPath);
+                    fin = new InputStreamReader(inputStream);
+
+                    BufferedReader input = new BufferedReader(fin);
+                    String line = null;
+                    int cnt = 0;
+
+                    //first stage
+                    while ((line = input.readLine()) != null) {
+
+                        String[] items = line.split("\t");
+                        if (items.length < 2) {
+                            continue;
+                        }
+
+                        // write sync
+                        putFeature(currLrModel, new BigInteger(items[0]).longValue(), Float.valueOf(items[1]).floatValue());
+                        cnt++;
+                    }
+                    LOGGER.info("load model [SUCCESS] , file path [{}], load item number [{}]", absPath, cnt);
+                } catch (Exception e) {
+                    failedPartitionNum++;
+                    LOGGER.error("load model file from hdfs occur error [FAILED], path: [{}], [{}]",
+                            absPath, ExceptionUtils.getFullStackTrace(e));
+                } finally {
+                    if (fin != null) {
+                        try {
+                            fin.close();
+                        } catch (IOException e) {
+                            LOGGER.error("close [{}] fail: [{}]", absPath, ExceptionUtils.getFullStackTrace(e));
+                        }
+                    }
+                }
+            }
+
+            if (failedPartitionNum == 0) {
+                this.lrModel = currLrModel;
+                LOGGER.info("[end] load model data from hdfs, spend time: [{}ms] model size: [{}], " +
+                        "total partition number [{}], failed partition numbers: [{}], model path [{}]", new Object[]{
+                        (System.currentTimeMillis() - startTime), getModelSize(), partitionsNum, failedPartitionNum, modelPath});
+                return true;
+            } else {
+                LOGGER.error("load model failed parts [{}]", failedPartitionNum);
+                return false;
+            }
+        } catch (Exception e) {
+            LOGGER.error("load model partitions occur error, model path [{}], error: [{}]",
+                    modelPath, ExceptionUtils.getFullStackTrace(e));
+            return false;
+        }
+    }
+
+    /**
+     * concurrency load model from modelpath
+     * put map must be sync
+     * if partitions not 0 && load success: return true
+     * exceptions || 0 partitions || any partitions failed : return false
+     *
+     * @param modelPath
+     * @return
+     */
+    public boolean loadPartitionsParallel(final String modelPath) {
+        try {
+            Path path = new Path(modelPath);
+            FileSystem fs = path.getFileSystem(hdfsConf);
+            FileStatus[] listStatus = fs.listStatus(path);
+
+            //judge null and empty
+            if (listStatus == null || listStatus.length == 0) {
+                LOGGER.error("model path is dir, but hdfs patition path is null");
+                return false;
+            }
+
+            long startTime = System.currentTimeMillis();
+            List<Callable<Long2FloatMap>> callables = new ArrayList<Callable<Long2FloatMap>>();
+            //multi thread load hdfs news info files
+            for (final FileStatus file : listStatus) {
+                callables.add(new Callable<Long2FloatMap>() {
+                    @Override
+                    public Long2FloatMap call() {
+                        // LOGGER.debug("load model file path [{}]", file.getPath().getName());
+                        String abspath = String.format("%s/%s", modelPath, file.getPath().getName());
+                        InputStreamReader fin = null;
+                        Long2FloatMap partModel = new Long2FloatOpenHashMap();
+                        try {
+
+                            Path path = new Path(abspath);
+                            FileSystem fs = path.getFileSystem(hdfsConf);
+                            FSDataInputStream inputStream = fs.open(path);
+                            fin = new InputStreamReader(inputStream);
+
+                            BufferedReader input = new BufferedReader(fin);
+                            String line = null;
+                            int cnt = 0;
+
+                            //first stage
+                            while ((line = input.readLine()) != null) {
+
+                                String[] items = line.split("\t");
+                                if (items.length < 2) {
+                                    continue;
+                                }
+
+                                // write sync
+                                partModel.put(new BigInteger(items[0]).longValue(), Float.valueOf(items[1]).floatValue());
+                                cnt++;
+                            }
+                            LOGGER.info("load model [SUCCESS] , file path [{}], load item number [{}]", abspath, cnt);
+                            return partModel;
+                        } catch (Exception e) {
+                            LOGGER.error("load model file from hdfs occur error [FAILED], path: [{}], [{}]",
+                                    abspath, ExceptionUtils.getFullStackTrace(e));
+                            return null;
+                        } finally {
+                            if (fin != null) {
+                                try {
+                                    fin.close();
+                                } catch (IOException e) {
+                                    LOGGER.error("close [{}] fail: [{}]", abspath, ExceptionUtils.getFullStackTrace(e));
+                                    return null;
+                                }
+                            }
+                            return partModel;
+                        }
+                    }
+                });
+            }
+
+            //invoke callable if failed return
+            List<Future<Long2FloatMap>> futures = null;
+            try {
+                futures = executorService.invokeAll(callables, 10, TimeUnit.MINUTES);
+            } catch (InterruptedException e) {
+                LOGGER.error("execute invoke fail: {}", ExceptionUtils.getFullStackTrace(e));
+            }
+
+            //wait for task complete
+            int failedPartitionNum = 0;
+            int partitionsNum = listStatus.length;
+            List<Long2FloatMap> currLrModel = constructModel();
+            for (Future<Long2FloatMap> future : futures) {
+                try {
+                    if (future.isDone() && !future.isCancelled()) {
+
+                        Long2FloatMap ret = future.get();
+                        if (ret == null) {
+                            failedPartitionNum++;
+                        }
+                        for (Map.Entry<Long, Float> entry : ret.entrySet()) {
+                            putFeature(currLrModel, entry.getKey(), entry.getValue());
+                        }
+                        ret = null; // gc
+                        continue;
+                    }
+                } catch (InterruptedException e) {
+                    failedPartitionNum++;
+                    LOGGER.error("InterruptedException {},{}", ExceptionUtils.getFullStackTrace(e));
+                } catch (ExecutionException e) {
+                    failedPartitionNum++;
+                    LOGGER.error("ExecutionException [{}]", ExceptionUtils.getFullStackTrace(e));
+                }
+            }
+
+            long endTime = System.currentTimeMillis();
+            // check all load success
+            if (failedPartitionNum == 0) {
+                this.lrModel = currLrModel;
+                // counter for alarm
+                LOGGER.info("[end] load model data from hdfs, spend time: [{}ms] model size: [{}], " +
+                        "total partition number [{}], failed partition numbers: [{}], model path [{}]", new Object[]{
+                        (endTime - startTime), getModelSize(), partitionsNum, failedPartitionNum, modelPath});
+                return true;
+            }
+
+            return false;
+        } catch (Exception e) {
+            LOGGER.error("load model partitions occur error, model path [{}], error: [{}]",
+                    modelPath, ExceptionUtils.getFullStackTrace(e));
+            return false;
+        }
+    }
+}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/score/model/Model.java

@@ -0,0 +1,13 @@
+package com.tzld.piaoquan.recommend.server.service.score.model;
+
+
+import java.io.InputStreamReader;
+
+abstract public class Model {
+    public abstract int getModelSize();
+
+    public abstract boolean loadFromStream(InputStreamReader in) throws Exception;
+
+    public abstract boolean loadPartitions(String modelPath, boolean isRegister);
+}
+

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/score/model/ModelManager.java

@@ -0,0 +1,274 @@
+package com.tzld.piaoquan.recommend.server.service.score.model;
+
+
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import org.apache.commons.lang.exception.ExceptionUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+
+
+public class ModelManager {
+    private static final int SCHEDULE_PERIOD = 10;
+    private static final Logger LOGGER = LoggerFactory.getLogger(ModelManager.class);
+    private static final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
+    private static final String SUCCESS = "_SUCCESS";
+    private static ModelManager instance;
+    Map<String, ModelLoadTask> loadTasks = new HashMap<String, ModelLoadTask>();
+    Map<String, String> modelPathMap = new HashMap<String, String>();
+    private Config config = ConfigFactory.load("hdfs_conf.properties");
+    private Configuration hdfsConf = new Configuration();
+
+    private ModelManager() {
+        //配置不同环境的hdfs conf
+        String coreSiteFile = config.hasPath("hdfs.coreSiteFile") ? StringUtils.trim(config.getString("hdfs.coreSiteFile")) : "core-site.xml";
+        String hdfsSiteFile = config.hasPath("hdfs.hdfsSiteFile") ? StringUtils.trim(config.getString("hdfs.hdfsSiteFile")) : "hdfs-site.xml";
+        hdfsConf.addResource(coreSiteFile);
+        hdfsConf.addResource(hdfsSiteFile);
+
+        start(SCHEDULE_PERIOD);
+    }
+
+    public static ModelManager getInstance() {
+        if (instance == null) {
+            synchronized (ModelManager.class) {
+                if (instance == null) {
+                    instance = new ModelManager();
+                }
+            }
+        }
+        return instance;
+    }
+
+    /**
+     * 添加一个加载任务到管理器
+     *
+     * @param modelName  Model的名字, 注册到ModelManager的不同model需要不同的名字
+     * @param path       Model在HDFS上的路径
+     * @param modelClass Model的子类型
+     */
+    public void registerModel(String modelName, String path, Class<? extends Model> modelClass) throws ModelRegisterException, IOException {
+        if (modelPathMap.containsKey(modelName)) {
+            String oldPath = modelPathMap.get(modelName);
+            if (path.equals(oldPath)) {
+                //如果模型的path没有发生改变, 不做任何操作
+                LOGGER.info("Model [{}] and Path [{}] has exist", modelName, path);
+                return;
+            } else {
+                //如果模型的path发生改变, 需要注销掉原有的任务
+                unRegisterModel(modelName);
+            }
+        }
+
+        modelPathMap.put(modelName, path);
+        if (loadTasks.containsKey(path)) {
+            ModelLoadTask loadTask = loadTasks.get(path);
+            loadTask.refCount++;
+        } else {
+            ModelLoadTask task = new ModelLoadTask(path, modelClass);
+            task.refCount++;
+            loadTasks.put(path, task);
+            loadModel(task, false, true);
+        }
+    }
+
+    /**
+     * 删除一个加载任务
+     *
+     * @param modelName Model的名字, 需要和registerModel的名字一致
+     */
+    private void unRegisterModel(String modelName) {
+        if (modelPathMap.containsKey(modelName)) {
+            String path = modelPathMap.get(modelName);
+            if (loadTasks.containsKey(path)) {
+                ModelLoadTask task = loadTasks.get(path);
+                task.refCount--;
+                if (task.refCount == 0) {
+                    loadTasks.remove(path);
+                }
+            }
+            modelPathMap.remove(modelName);
+        }
+    }
+
+    /**
+     * @param modelName
+     * @return
+     */
+    public Model getModel(String modelName) {
+        if (modelPathMap.containsKey(modelName) && loadTasks.containsKey(modelPathMap.get(modelName))) {
+            return loadTasks.get(modelPathMap.get(modelName)).model;
+        } else {
+            return null;
+        }
+    }
+
+    /**
+     * 开始调度
+     *
+     * @param period
+     */
+    protected void start(long period) {
+        final Runnable task = new Runnable() {
+            public void run() {
+                // 模型更新开关
+                // boolean modelUpdateSwitch = Configuration.getBoolean("recommend-service-framework.model-update-switch", true);
+                boolean modelUpdateSwitch =  true;
+                LOGGER.info("model update switch [{}]", modelUpdateSwitch);
+                if (modelUpdateSwitch) {
+                    updateModels(false);
+                }
+            }
+        };
+        scheduler.scheduleAtFixedRate(task, 10, period, TimeUnit.MINUTES); // 10分钟
+    }
+
+    /**
+     * 更新模型
+     */
+    public void updateModels(final boolean isForceLoads) {
+        LOGGER.info("begin to update: [{}]", loadTasks.keySet().size());
+        for (String modelPath : loadTasks.keySet()) {
+            LOGGER.debug("loadtask model path [{}]", modelPath);
+            ModelLoadTask task = loadTasks.get(modelPath);
+            loadModel(task, isForceLoads, false);
+        }
+    }
+
+    /**
+     * 检查并加载模型
+     *
+     * @param loadTask
+     */
+    private void loadModel(final ModelLoadTask loadTask, final boolean isForceLoads, final boolean isRegister) {
+        if (loadTask.isLoading) {
+            return;
+        }
+        FSDataInputStream fin = null;
+        try {
+            loadTask.isLoading = true;
+            Path path = new Path(loadTask.path);
+            hdfsConf.setBoolean("fs.hdfs.impl.disable.cache", true);
+
+            //根据path选择相应的filesystem
+            FileSystem fs = path.getFileSystem(hdfsConf);
+            FileStatus fileStatus = fs.getFileStatus(path);
+
+            // model is file, modify sign is file ModificationTime
+            if (fileStatus.isFile()) {
+
+                long timeStamp = fileStatus.getModificationTime();
+                LOGGER.debug("model [{}] last update time: [{}]", loadTask.path, loadTask.lastModifyTime + ":" + timeStamp);
+                if (loadTask.lastModifyTime != timeStamp || isForceLoads) {
+                    LOGGER.info("find model [{}] file changed, ready to update, last modify: [{}]", loadTask.path, loadTask.lastModifyTime);
+
+                    Path modelFile = new Path(loadTask.path);
+                    fin = fs.open(modelFile);
+                    Model model = loadTask.modelClass.newInstance();
+                    if (model.loadFromStream(new InputStreamReader(fin))) {
+                        loadTask.model = model;
+                        loadTask.lastModifyTime = timeStamp;
+                        //LOGGER.info("finish load model: [{}], [{}]", loadTask.path);
+                        LOGGER.info("update model: [{}], [{}]", loadTask.path, loadTask.lastModifyTime);
+                    } else {
+                        LOGGER.error("load model fail: [{}]", loadTask.path);
+                    }
+                }
+                return;
+            }
+
+            // model is dir, modify sign is file _SUCCESS file ModificationTime
+            if (fileStatus.isDirectory()) {
+
+                String successFilePath = String.format("%s/%s", loadTask.path, SUCCESS);
+                Path successPath = new Path(successFilePath);
+                long currSuccFileModifyTime = 0;
+                if (!isForceLoads) {
+                    try {
+                        FileSystem fsSucc = successPath.getFileSystem(hdfsConf);
+                        FileStatus fileStatusSucc = fsSucc.getFileStatus(successPath);
+                        currSuccFileModifyTime = fileStatusSucc.getModificationTime();
+
+                        //check model whether update
+                        if (loadTask.lastModifyTime == currSuccFileModifyTime) {
+                            LOGGER.info("[INFO MESSAGE] _SUCCESS modifytime is not changed, last update time [{}]," +
+                                    "current modify time [{}], model path: [{}]", new Object[]{loadTask.lastModifyTime,
+                                    currSuccFileModifyTime, loadTask.path});
+                            return;
+                        }
+                    } catch (Exception e) {
+                        LOGGER.error("check _SUCCESS update time occur error, _SUCCESS path [{}], Exception: [{}]",
+                                successPath, ExceptionUtils.getFullStackTrace(e));
+                        return;
+                    }
+                }
+
+                // success has modify, multi thread load model
+                Model model = loadTask.modelClass.newInstance();
+                if (model.loadPartitions(loadTask.path, isRegister)) {
+                    loadTask.model = model;
+                    loadTask.lastModifyTime = currSuccFileModifyTime;
+
+                    LOGGER.info("[SUCCESS] model load, update time [{}], model size [{}], model path [{}]",
+                            new Object[]{loadTask.lastModifyTime, model.getModelSize(), loadTask.path});
+                } else {
+                    LOGGER.error("load model [failed], model path [{}]", loadTask.path);
+                }
+            }
+
+        } catch (Exception e) {
+            LOGGER.error("update model fail:[{}], [{}]", loadTask.path, ExceptionUtils.getFullStackTrace(e));
+        } finally {
+            loadTask.isLoading = false;
+            if (fin != null) {
+                try {
+                    fin.close();
+                } catch (IOException e) {
+                    LOGGER.error("close [{}] fail: [{}]", loadTask.path, ExceptionUtils.getFullStackTrace(e));
+                }
+            }
+        }
+    }
+
+    public class ModelRegisterException extends Exception {
+
+        public ModelRegisterException(String s) {
+            super(s);
+        }
+    }
+
+    /**
+     * 调度的任务单元
+     */
+    private class ModelLoadTask {
+
+        private int refCount;
+        private String path;
+        private long lastModifyTime;
+        private boolean isLoading;
+        private Class<? extends Model> modelClass;
+        private Model model;
+
+        ModelLoadTask(String path, Class<? extends Model> modelClass) {
+            this.refCount = 0;
+            this.path = path;
+            this.lastModifyTime = 0;
+            this.modelClass = modelClass;
+        }
+    }
+}

recommend-server-service/src/main/resources/application-dev.yml

@@ -69,4 +69,4 @@ aliyun:
 
 logging:
   file:
-    path: /Users/dingyunpeng/Desktop/code/pq/recommend-server/logs/${spring.application.name}/
+    path: /Users/sunmingze/Documents/logs/${spring.application.name}/

recommend-server-service/src/main/resources/feeds_score_config_baseline.conf

@@ -0,0 +1,7 @@
+scorer-config = {
+  related-score-config = {
+    scorer-name = "com.tzld.piaoquan.recommend.server.service.score.FeedsShareLTRScorer"
+    scorer-priority = 99
+    model-path = ".v1.xgb"
+  }
+}