【基于Mahout的个性化推荐系统】3、显微镜模块设计

蔚落 2021-11-02 01:31 338阅读 0赞

总纲传送门：[基于Mahout的个性化推荐系统][Mahout]  
GitHub传送门：[基于Mahout的个性化推荐系统][Mahout 1]

# 一、前言 #

在推荐系统中，往往会有很多数据来源，比如初始数据集录入（学习使用）、第三方平台数据（类似广告联盟）、用户行为数据（在平台上的操作数据如点击）、用户属性分析数据（年龄性别等）等。

本文主要讲解推荐系统的数据来源模块——数据采集模块、执行器模块。数据采集主要用于采集及写入，执行器主要用于计算刷新用户偏好。

在本项目中，数据采集模块主要采集用户行为数据和推荐商品数据。用户行为信息主要用于计算用户偏好，推荐商品信息主要用于构建排行榜，进行热门推荐。

### 取名，讲究 ###

显微镜：借鉴于曾经实习的时候，接触过前端项目名为显微镜——主要做前端埋点上报用户行为信息所用。

嗅探器：在采集信息的时候，除了前端上报信息外，均是采用异步计算方式进行采集、计算写入，类似于网络攻防中的“嗅探器”，故名之。

执行器：主要用于进行执行计算用户偏好信息，将用户行为信息进行计算并写入用户偏好表中，起到执行器的作用。

# 二、异步 #

在此模块中，用户信息采集、计算写入均是采用的异步执行。在本项目中，异步的实现主要是通过基于线程池进行。通过配置线程池，当异步方法发生调用时，通过到线程池中获取线程进行方法执行调用。

**线程池配置示例：**

/**
     * 基础业务逻辑
     */
    @Bean
    public Executor asyncServiceExecutor() {
        
        log.info("ExecutorConfiguration#asyncServiceExecutor start...");
        return getDefaultExecutor();
    }
    
    /**
     * 获取默认执行器
     */
    private synchronized Executor getDefaultExecutor() {
        
        ThreadPoolTaskExecutor executor = new VisibleThreadPoolTaskExecutor();
        executor.setCorePoolSize(corePoolSize);
        executor.setMaxPoolSize(maxPoolSize);
        executor.setQueueCapacity(queueCapacity);
        executor.setThreadNamePrefix(threadNamePrefix);
        executor.setKeepAliveSeconds(keepAliveSeconds);
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //执行初始化
        executor.initialize();
        return executor;
    }

其中，方法名asyncServiceExecutor即为线程池名称。同时，在配置过程中，需要注意线程池策略问题，以下为各个策略说明：

*  ThreadPoolExecutor.AbortPolicy: 丢弃任务并抛出RejectedExecutionException异常。
 *  ThreadPoolExecutor.DiscardPolicy： 也是丢弃任务，但是不抛出异常。
 *  ThreadPoolExecutor.DiscardOldestPolicy： 丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
 *  ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务

在使用中，仅需在需要异步执行的方法头中，添加注解@Async即可，代码示例：

@Async("asyncServiceExecutor")
    public void asyncFunction(int data) {
        log.info("asyncFunction is running. data={}", data);
        try {
            Thread.sleep(5000L);
        } catch (InterruptedException e) {
            log.error("asyncFunction throws exception. e={}", e);
        }
        log.info("asyncFunction is ended. data={}", data);
    }

# 三、嗅探器-用户行为信息 #

在用户行为信息采集模块中，主要分为采集用户行为信息和客户端上报。

## 1、采集用户行为信息 ##

对于异步采集来说，主要是当用户对商品进行浏览（点击事件，对应获取商品详情）、添加购物车（新增购物车记录）、购买商品（购买事件、新增订单记录）等事件的时候，请求后台接口对应获取商品详情、新增购物车记录、新增订单记录等，在controller接口中，调用异步采集方法，进行用户行为数据上报。

上边这段话说的不怎么清楚哈，对比如下表所示。

<table> 
 <thead> 
  <tr> 
   <th>用户行为</th> 
   <th>行为事件</th> 
   <th>对应接口说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>浏览商品</td> 
   <td>点击事件</td> 
   <td>根据商品ID获取商品详情</td> 
  </tr> 
  <tr> 
   <td>添加购物车</td> 
   <td>购物车事件</td> 
   <td>新增购物车记录</td> 
  </tr> 
  <tr> 
   <td>删除购物车</td> 
   <td>购物车事件</td> 
   <td>删除购物车记录</td> 
  </tr> 
  <tr> 
   <td>购买商品</td> 
   <td>购买事件</td> 
   <td>新增订单记录</td> 
  </tr> 
 </tbody> 
</table>

**异步采集信息流程：**

1.  用户请求相应接口并正常返回数据
2.  在接口中，调用异步方法，进行数据采集
3.  异步方法中，对于数据进行封装、计算等处理
4.  调用执行器进行持久化写入

**请求接口埋点采集代码示例：**

/**
     * 根据ID获取商品信息
     */
    @GetMapping(value = "/goods/{id}")
    public ResultVo getGoodsById(@PathVariable(value = "id") Long id, HttpServletRequest request) {
        
        // TODO 其它业务处理
        /**
         * 采集用户点击记录，异步
         */
        userGoodsBehaviorSniffer.click(id, request);
        // 返回结果
        return ResultUtil.success(goodsService.getById(id));
    }

**对用户行为信息进行计算、写入代码示例：**

public void click(long goodsId, HttpServletRequest request) {
        
        // TODO 其它业务处理
        long userId = sessionUtil.getUserIdNoCheck(request);
        if (userId <= 0) {
        
            // 用户未登录，暂不做任何处理
            log.error("UserGoodsSniffer#click: user not logged in. goodsId={}, requst={}. ", goodsId, request);
            /**
             * TODO 其它处理
             */
            throw new AsyncBusinessException(ErrorEnum.NO_USER);
        }
        /**
         *  记录用户浏览该商品记录
         */
        int insertResult = userBehaviorService.insert(goodsId, userId, UserBehaviorEnum.CLICK);
        /**
         * 刷新商品推荐排行榜
         */
        pushRanking(goodsId);
        /**
         * TODO 后期对返回结果处理
         */
    }

## 2、客户端上报 ##

客户端上报，主要是针对“采集用户行为信息”的盲点区域，比如客户端实现过程中，从列表页到详情页时并未从后端获取数据，而是通过列表页传递数据。再如当进行消息客服时，用户是对于该商品感兴趣并且有购买的欲望，由于在消息客服的实现中，只进行了websocket的实时通信实现，并未进行商品信息的记录，所以该行为也是通过接口方式无法采集到的。

同样的，还是简单罗列下（啊哈哈哈哈哈哈哈哈哈）：

<table> 
 <thead> 
  <tr> 
   <th>用户行为</th> 
   <th>行为事件</th> 
   <th>对应接口说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>浏览商品</td> 
   <td>点击事件</td> 
   <td>根据商品ID获取商品详情</td> 
  </tr> 
  <tr> 
   <td>消息客服</td> 
   <td>消息客服事件</td> 
   <td>联系客服</td> 
  </tr> 
 </tbody> 
</table>

总的来说，客户端上报主要是对于采集用户行为信息的一个补充。

# 四、嗅探器-推荐商品信息 #

对于推荐商品信息的采集，主要是对于通过推荐算法推荐出来的商品信息，进行采集其推荐次数，从而构建推荐排行榜，实现热门推荐效果。

此模块的实现，类似于用户行为信息采集，也是进行的异步计算写入，同样的也是采用的面向接口编程。

其不一样的点在于其不只是对于数据库进行了写入数据，同时也进行缓存服务器redis的写入。对于Redis部分，一方面是实时进行数据的更新，另一方面是当项目启动过程中，会从DB中拉取数据进行刷新数据至redis。

**代码示例：**

@Slf4j
    @Service
    public class RecommendRankingGoodsSnifferImpl implements RecommendRankingGoodsSniffer {
        
        @Autowired
        private MyRedisUtil redisUtil;
        @Autowired
        private RecommendRankingGoodsService rankingGoodsService;
    
        /**
         * 初始化
         */
        @PostConstruct
        public void init() {
        
            /**
             * 1、将DB中的ranking数据写入redis
             */
            initRedis();
        }
    
        @Override
        @Async(value = AsyncNameConstant.MICROSCOPE)
        public void pushRanking(long goodsId) {
        
            /**
             * 写入redis
             */
            double redisResult = redisUtil.incrementScoreToRankginGoods(goodsId);
            /**
             * 写入数据库
             */
            int dbResult = rankingGoodsService.incrementRank(goodsId);
            /**
             * TODO 处理结果
             */
        }
    
        /**
         * 将DB中的数据写入redis
         */
        private void initRedis() {
        
            long total = rankingGoodsService.count();
            PageDto<RecommendRankingGoods> rankingGoodsList = rankingGoodsService.listAll(NumericConstant.ONE, total);
            redisUtil.putRecommendRankingGoods(rankingGoodsList.getData());
        }
    }

# 五、执行器 #

对于执行器的作用，前文中有介绍主要是进行用户偏好信息的计算及持久化写入。在用户行为信息采集后，进行持久化写入成功后，便会调用执行器的refresh方法，去刷新用户偏好信息。

**执行器代码示例：**

注：此代码为进行批量计算并刷新用户偏好的方法，单个记录刷新与此类似。

public Boolean refresh(long userId, Date startTime) {
        
        /**
         * 1、获取所有用户行为
         */
        PageDto behaviorDto = behaviorService.listByUserAndStartTime(userId, startTime);
        if (behaviorDto == null || behaviorDto.getTotalCount() <= 0) {
        
            // 为获取到数据
            log.error("UserGoodsPreferenceActuator#refresh: no data. userId={}, startTime={}.", userId, startTime);
            throw new AsyncBusinessException(ErrorEnum.BEHAVIOR_DATA_NOT_EXISTS);
        }
        /**
         * 2、计算用户对于每个商品的偏好
         */
        List<UserPreference> preferenceList = preferenceCalculation.calculation(behaviorDto.getData());
        /**
         * 3、将商品偏好分为两大类，已存在用户偏好记录和尚不存在
         */
        List<UserPreference> alivePreferenceList = Lists.newArrayList();
        List<UserPreference> notAlivePreferenceList = Lists.newArrayList();
        for (UserPreference item : preferenceList) {
        
            if (preferenceService.isAliveUserPreference(item)) {
        
                alivePreferenceList.add(item);
            } else {
        
                notAlivePreferenceList.add(item);
            }
        }
        /**
         * 4、对已存在的用户偏好记录，执行批量更新操作
         */
        if (alivePreferenceList.size() > 0) {
        
            int updateResult = preferenceService.update(alivePreferenceList);
            /**
             * TODO 校验结果
             */
        }
        /**
         * 5、对不存在的用户偏好记录，执行批量插入操作
         */
        if (notAlivePreferenceList.size() > 0) {
        
            int insertResult = preferenceService.insert(notAlivePreferenceList);
            /**
             * TODO 校验结果
             */
        }
        /**
         * TODO 处理结果
         */
        return true;
    }

在此方法的实现过程中，主要是进行刷新用户的偏好信息，对于用户偏好信息的计算，主要又是对于用户的行为信息进行计算。为提高用户偏好信息的精度（用户会随着时间的推移，改变自己的喜好），在拉取用户行为信息的时候采用了时间窗口，默认时间窗口为三个月，用户可自定义其时间窗口。

同时，在写入用户偏好过程中，首先会进行偏好信息存在性检查，如果存在则进行更新，如果不存在则进行插入处理。

# 六、小结 #

在本文中，主要进行采集信息模块的介绍。对于介绍来说，很为简要甚至某些地方可能描述得较为不明确，包括代码部分。在后期中，会对该项目的代码进行整理并免费开源，欢迎大家阅读源码交流，谢谢。

# 参考 #

[Mahout]: https://blog.csdn.net/chensanwa/article/details/90522346
[Mahout 1]: https://github.com/chenlinghong/graduation