课程说明

  • 圈子推荐功能说明
  • 圈子推荐功能流程
  • 圈子推荐功能的实现
  • 小视频推荐功能的实现

1、圈子推荐

1.1、功能说明

在圈子功能中，针对于用户发布的动态信息，系统可以根据用户的发布、浏览、点赞等操作，对动态信息做计算，然后对每个用户进行不同的推荐。

1.2、流程说明

day07-圈子推荐功能实现 - 图2

流程说明：

  • 用户对圈子的动态操作，如：发布、浏览、点赞、喜欢等，就会给RocketMQ进行发送消息；
  • 推荐系统接收消息，并且处理消息数据，处理之后将结果数据写入到MongoDB中；
  • Spark系统拉取数据，然后进行推荐计算；
  • 计算之后的结果数据写入到Redis中，为每个用户都进行个性化推荐；
  • 如果有用户没有数据的，查询MongoDB中的默认数据；

1.3、动态增加自增id

由于我们使用的推荐模型中，动态id需要是Long类型的，而我们之前使用的ObjectId类型的，所以需要增加Long类型的id。

1.3.1、修改Publish对象

  1. package com.tanhua.dubbo.server.pojo;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  6. import org.bson.types.ObjectId;
  7. import org.springframework.data.mongodb.core.mapping.Document;
  9. import java.util.Date;
  10. import java.util.List;
  12. /**
  13.  * 发布表，动态内容
  14.  */
  15. @Data
  16. @NoArgsConstructor
  17. @AllArgsConstructor
  18. @Document(collection = "quanzi_publish")
  19. public class Publish implements java.io.Serializable {
  21.     private static final long serialVersionUID = 8732308321082804771L;
  23.     private ObjectId id; //主键id
  24.     private Long pid; //Long类型的id，用于推荐引擎使用
  25.     private Long userId;
  26.     private String text; //文字
  27.     private List<String> medias; //媒体数据，图片或小视频 url
  28.     private Integer seeType; // 谁可以看，1-公开，2-私密，3-部分可见，4-不给谁看
  29.     private List<Long> seeList; //部分可见的列表
  30.     private List<Long> notSeeList; //不给谁看的列表
  31.     private String longitude; //经度
  32.     private String latitude; //纬度
  33.     private String locationName; //位置名称
  34.     private Long created; //发布时间
  36. }

1.3.2、修改发布逻辑

  1. @Override
  2.     public String savePublish(Publish publish) {
  4.         // 校验
  5.         if (publish.getUserId() == null) {
  6.             return null;
  7.         }
  9.         try {
  10.             publish.setCreated(System.currentTimeMillis()); //设置创建时间
  11.             publish.setId(ObjectId.get()); //设置id
  12.             publish.setPid(this.idService.createId("publish", publish.getId().toHexString()));
  14.             this.mongoTemplate.save(publish); //保存发布
  16.             Album album = new Album(); // 构建相册对象
  17.             album.setPublishId(publish.getId());
  18.             album.setCreated(System.currentTimeMillis());
  19.             album.setId(ObjectId.get());
  20.             this.mongoTemplate.save(album, "quanzi_album_" + publish.getUserId());
  22.             //写入好友的时间线中
  23.             Criteria criteria = Criteria.where("userId").is(publish.getUserId());
  24.             List<Users> users = this.mongoTemplate.find(Query.query(criteria), Users.class);
  25.             for (Users user : users) {
  26.                 TimeLine timeLine = new TimeLine();
  27.                 timeLine.setId(ObjectId.get());
  28.                 timeLine.setPublishId(publish.getId());
  29.                 timeLine.setUserId(user.getUserId());
  30.                 timeLine.setDate(System.currentTimeMillis());
  31.                 this.mongoTemplate.save(timeLine, "quanzi_time_line_" + user.getFriendId());
  32.             }
  34.             return publish.getId().toHexString();
  35.         } catch (Exception e) {
  36.             e.printStackTrace();
  37.             //TODO 出错的事务回滚
  38.         }
  40.         return null;
  41.     }
  1. package com.tanhua.dubbo.server.service;
  3. import org.apache.commons.lang3.StringUtils;
  4. import org.springframework.beans.factory.annotation.Autowired;
  5. import org.springframework.data.redis.core.RedisTemplate;
  6. import org.springframework.stereotype.Service;
  8. //生成自增长的id，原理：使用redis的自增长值
  9. @Service
  10. public class IdService {
  12.     @Autowired
  13.     private RedisTemplate<String, String> redisTemplate;
  15.     public Long createId(String type, String strId) {
  16.         type = StringUtils.upperCase(type);
  18.         String idHashKey = "TANHUA_ID_HASH_" + type;
  19.         if (this.redisTemplate.opsForHash().hasKey(idHashKey, strId)) {
  20.             return Long.valueOf(this.redisTemplate.opsForHash().get(idHashKey, strId).toString());
  21.         }
  23.         String idKey = "TANHUA_ID_" + type;
  24.         Long id = this.redisTemplate.opsForValue().increment(idKey);
  26.         this.redisTemplate.opsForHash().put(idHashKey, strId, id.toString());
  28.         return id;
  29.     }
  31. }

itcast-tanhua-dubbo-service需要增加Redis依赖：

  1. <dependency>
  2.     <groupId>org.springframework.boot</groupId>
  3.     <artifactId>spring-boot-starter-data-redis</artifactId>
  4. </dependency>

1.4、动态计分规则

  • 浏览 +1
  • 点赞 +5
  • 喜欢 +8
  • 评论 + 10
  • 文字长度：50以内1分，50~100之间2分，100以上3分
  • 图片个数：每个图片一分

1.5、发送消息

1.5.1、QuanziMQService

itcast-tanhua-server增加依赖：

  1. <!--RocketMQ相关-->
  2. <dependency>
  3.     <groupId>org.apache.rocketmq</groupId>
  4.     <artifactId>rocketmq-spring-boot-starter</artifactId>
  5.     <version>2.0.3</version>
  6. </dependency>
  7. <dependency>
  8.     <groupId>org.apache.rocketmq</groupId>
  9.     <artifactId>rocketmq-client</artifactId>
  10.     <version>4.6.0</version>
  11. </dependency>

配置文件：

  1. # RocketMQ相关配置
  2. rocketmq.name-server=172.16.55.155:9876
  3. rocketmq.producer.group=tanhua
  1. package com.tanhua.server.service;
  3. import com.alibaba.dubbo.config.annotation.Reference;
  4. import com.tanhua.dubbo.server.api.QuanZiApi;
  5. import com.tanhua.dubbo.server.pojo.Publish;
  6. import com.tanhua.server.pojo.User;
  7. import com.tanhua.server.utils.UserThreadLocal;
  8. import org.apache.rocketmq.spring.core.RocketMQTemplate;
  9. import org.slf4j.Logger;
  10. import org.slf4j.LoggerFactory;
  11. import org.springframework.beans.factory.annotation.Autowired;
  12. import org.springframework.stereotype.Service;
  14. import java.util.Date;
  15. import java.util.HashMap;
  16. import java.util.Map;
  18. @Service
  19. public class QuanziMQService {
  21.     private static final Logger LOGGER = LoggerFactory.getLogger(QuanziMQService.class);
  23.     @Autowired
  24.     private RocketMQTemplate rocketMQTemplate;
  26.     @Reference(version = "1.0.0")
  27.     private QuanZiApi quanZiApi;
  29.     /**
  30.      * 发布动态消息
  31.      *
  32.      * @param publishId
  33.      * @return
  34.      */
  35.     public Boolean publishMsg(String publishId) {
  36.         return this.sendMsg(publishId, 1);
  37.     }
  39.     /**
  40.      * 浏览动态消息
  41.      *
  42.      * @param publishId
  43.      * @return
  44.      */
  45.     public Boolean queryPublishMsg(String publishId) {
  46.         return this.sendMsg(publishId, 2);
  47.     }
  49.     /**
  50.      * 点赞动态消息
  51.      *
  52.      * @param publishId
  53.      * @return
  54.      */
  55.     public Boolean likePublishMsg(String publishId) {
  56.         return this.sendMsg(publishId, 3);
  57.     }
  59.     /**
  60.      * 取消点赞动态消息
  61.      *
  62.      * @param publishId
  63.      * @return
  64.      */
  65.     public Boolean disLikePublishMsg(String publishId) {
  66.         return this.sendMsg(publishId, 6);
  67.     }
  69.     /**
  70.      * 喜欢动态消息
  71.      *
  72.      * @param publishId
  73.      * @return
  74.      */
  75.     public Boolean lovePublishMsg(String publishId) {
  76.         return this.sendMsg(publishId, 4);
  77.     }
  79.     /**
  80.      * 取消喜欢动态消息
  81.      *
  82.      * @param publishId
  83.      * @return
  84.      */
  85.     public Boolean disLovePublishMsg(String publishId) {
  86.         return this.sendMsg(publishId, 7);
  87.     }
  89.     /**
  90.      * 评论动态消息
  91.      *
  92.      * @param publishId
  93.      * @return
  94.      */
  95.     public Boolean commentPublishMsg(String publishId) {
  96.         return this.sendMsg(publishId, 5);
  97.     }
  99.     /**
  100.      * 发送圈子操作相关的消息
  101.      *
  102.      * @param publishId
  103.      * @param type      1-发动态，2-浏览动态， 3-点赞， 4-喜欢， 5-评论，6-取消点赞，7-取消喜欢
  104.      * @return
  105.      */
  106.     private Boolean sendMsg(String publishId, Integer type) {
  107.         try {
  108.             User user = UserThreadLocal.get();
  110.             Publish publish = this.quanZiApi.queryPublishById(publishId);
  112.             //构建消息
  113.             Map<String, Object> msg = new HashMap<>();
  114.             msg.put("userId", user.getId());
  115.             msg.put("date", System.currentTimeMillis());
  116.             msg.put("publishId", publishId);
  117.             msg.put("pid", publish.getPid());
  118.             msg.put("type", type);
  120.             this.rocketMQTemplate.convertAndSend("tanhua-quanzi", msg);
  121.         } catch (Exception e) {
  122.             LOGGER.error("发送消息失败! publishId = " + publishId + ", type = " + type, e);
  123.             return false;
  124.         }
  126.         return true;
  127.     }
  128. }

1.5.2、修改MovementsController

  1. package com.tanhua.server.controller;
  3. import com.tanhua.server.service.MovementsService;
  4. import com.tanhua.server.service.QuanziMQService;
  5. import com.tanhua.server.vo.Movements;
  6. import com.tanhua.server.vo.PageResult;
  7. import org.apache.commons.lang3.StringUtils;
  8. import org.springframework.beans.factory.annotation.Autowired;
  9. import org.springframework.http.HttpStatus;
  10. import org.springframework.http.ResponseEntity;
  11. import org.springframework.web.bind.annotation.*;
  12. import org.springframework.web.multipart.MultipartFile;
  14. @RestController
  15. @RequestMapping("movements")
  16. public class MovementsController {
  18.     @Autowired
  19.     private MovementsService movementsService;
  21.     @Autowired
  22.     private QuanziMQService quanziMQService;
  24.     /**
  25.      * 发送动态
  26.      *
  27.      * @param textContent
  28.      * @param location
  29.      * @param multipartFile
  30.      * @return
  31.      */
  32.     @PostMapping()
  33.     public ResponseEntity<Void> savePublish(@RequestParam("textContent") String textContent,
  34.                                             @RequestParam("location") String location,
  35.                                             @RequestParam("longitude") String longitude,
  36.                                             @RequestParam("latitude") String latitude,
  37.                                             @RequestParam(value = "imageContent", required = false) MultipartFile[] multipartFile) {
  38.         try {
  39.             String publishId = this.movementsService.savePublish(textContent, location, longitude, latitude, multipartFile);
  40.             if (StringUtils.isNotEmpty(publishId)) {
  41.                 // 发送消息
  42.                 this.quanziMQService.publishMsg(publishId);
  43.                 return ResponseEntity.ok(null);
  44.             }
  45.         } catch (Exception e) {
  46.             e.printStackTrace();
  47.         }
  48.         return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
  49.     }
  51.     /**
  52.      * 查询好友动态
  53.      *
  54.      * @param page
  55.      * @param pageSize
  56.      * @return
  57.      */
  58.     @GetMapping
  59.     public PageResult queryPublishList(@RequestParam(value = "page", defaultValue = "1") Integer page,
  60.                                        @RequestParam(value = "pagesize", defaultValue = "10") Integer pageSize) {
  61.         return this.movementsService.queryPublishList(page, pageSize, false);
  62.     }
  64.     /**
  65.      * 查询推荐动态
  66.      *
  67.      * @param page
  68.      * @param pageSize
  69.      * @return
  70.      */
  71.     @GetMapping("recommend")
  72.     public PageResult queryRecommendPublishList(@RequestParam(value = "page", defaultValue = "1") Integer page,
  73.                                                 @RequestParam(value = "pagesize", defaultValue = "10") Integer pageSize) {
  74.         return this.movementsService.queryPublishList(page, pageSize, true);
  75.     }
  77.     /**
  78.      * 点赞
  79.      *
  80.      * @param publishId
  81.      * @return
  82.      */
  83.     @GetMapping("/{id}/like")
  84.     public ResponseEntity<Long> likeComment(@PathVariable("id") String publishId) {
  85.         try {
  86.             Long likeCount = this.movementsService.likeComment(publishId);
  87.             if (likeCount != null) {
  89.                 //发送点赞消息
  90.                 this.quanziMQService.likePublishMsg(publishId);
  92.                 return ResponseEntity.ok(likeCount);
  93.             }
  94.         } catch (Exception e) {
  95.             e.printStackTrace();
  96.         }
  97.         return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
  98.     }
  100.     /**
  101.      * 取消点赞
  102.      *
  103.      * @param publishId
  104.      * @return
  105.      */
  106.     @GetMapping("/{id}/dislike")
  107.     public ResponseEntity<Long> disLikeComment(@PathVariable("id") String publishId) {
  108.         try {
  109.             Long likeCount = this.movementsService.cancelLikeComment(publishId);
  110.             if (null != likeCount) {
  112.                 //发送取消点赞消息
  113.                 this.quanziMQService.disLikePublishMsg(publishId);
  115.                 return ResponseEntity.ok(likeCount);
  116.             }
  117.         } catch (Exception e) {
  118.             e.printStackTrace();
  119.         }
  120.         return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
  121.     }
  123.     /**
  124.      * 喜欢
  125.      *
  126.      * @param publishId
  127.      * @return
  128.      */
  129.     @GetMapping("/{id}/love")
  130.     public ResponseEntity<Long> loveComment(@PathVariable("id") String publishId) {
  131.         try {
  132.             Long loveCount = this.movementsService.loveComment(publishId);
  133.             if (null != loveCount) {
  135.                 //发送喜欢消息
  136.                 this.quanziMQService.lovePublishMsg(publishId);
  138.                 return ResponseEntity.ok(loveCount);
  139.             }
  140.         } catch (Exception e) {
  141.             e.printStackTrace();
  142.         }
  143.         return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
  144.     }
  146.     /**
  147.      * 取消喜欢
  148.      *
  149.      * @param publishId
  150.      * @return
  151.      */
  152.     @GetMapping("/{id}/unlove")
  153.     public ResponseEntity<Long> disLoveComment(@PathVariable("id") String publishId) {
  154.         try {
  155.             Long loveCount = this.movementsService.cancelLoveComment(publishId);
  156.             if (null != loveCount) {
  158.                 //发送取消喜欢消息
  159.                 this.quanziMQService.disLovePublishMsg(publishId);
  161.                 return ResponseEntity.ok(loveCount);
  162.             }
  163.         } catch (Exception e) {
  164.             e.printStackTrace();
  165.         }
  166.         return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
  167.     }
  169.     /**
  170.      * 查询单条动态信息
  171.      *
  172.      * @param publishId
  173.      * @return
  174.      */
  175.     @GetMapping("/{id}")
  176.     public ResponseEntity<Movements> queryById(@PathVariable("id") String publishId) {
  177.         try {
  178.             Movements movements = this.movementsService.queryById(publishId);
  179.             if (null != movements) {
  181.                 //发送消息
  182.                 this.quanziMQService.queryPublishMsg(publishId);
  184.                 return ResponseEntity.ok(movements);
  185.             }
  186.         } catch (Exception e) {
  187.             e.printStackTrace();
  188.         }
  189.         return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
  190.     }
  192. }

CommentsController：

  1.  /**
  2.      * 发表评论
  3.      *
  4.      * @param param
  5.      * @return
  6.      */
  7.     @PostMapping
  8.     public ResponseEntity<Void> saveComments(@RequestBody Map<String, String> param) {
  9.         try {
  10.             String publishId = param.get("movementId");
  11.             String content = param.get("comment");
  12.             Boolean bool = this.commentsService.saveComments(publishId, content);
  13.             if (bool) {
  15.                 //发送消息
  16.                 this.quanziMQService.sendCommentPublishMsg(publishId);
  18.                 return ResponseEntity.ok(null);
  19.             }
  20.         } catch (Exception e) {
  21.             e.printStackTrace();
  22.         }
  23.         return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
  24.     }

1.6、接收消息

1.6.1、创建itcast-tanhua-recommend工程

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <parent>
  6.         <artifactId>itcast-tanhua</artifactId>
  7.         <groupId>cn.itcast.tanhua</groupId>
  8.         <version>1.0-SNAPSHOT</version>
  9.     </parent>
  10.     <modelVersion>4.0.0</modelVersion>
  12.     <artifactId>itcast-tanhua-recommend</artifactId>
  14.     <dependencies>
  15.         <dependency>
  16.             <groupId>org.springframework.boot</groupId>
  17.             <artifactId>spring-boot-starter-web</artifactId>
  18.         </dependency>
  19.         <dependency>
  20.             <groupId>org.springframework.boot</groupId>
  21.             <artifactId>spring-boot-starter-test</artifactId>
  22.             <scope>test</scope>
  23.         </dependency>
  24.         <dependency>
  25.             <groupId>org.springframework.boot</groupId>
  26.             <artifactId>spring-boot-starter-data-redis</artifactId>
  27.         </dependency>
  28.         <dependency>
  29.             <groupId>org.springframework.boot</groupId>
  30.             <artifactId>spring-boot-starter-data-mongodb</artifactId>
  31.         </dependency>
  32.         <!--其他工具包依赖-->
  33.         <dependency>
  34.             <groupId>org.apache.commons</groupId>
  35.             <artifactId>commons-lang3</artifactId>
  36.         </dependency>
  37.         <dependency>
  38.             <groupId>commons-io</groupId>
  39.             <artifactId>commons-io</artifactId>
  40.             <version>2.6</version>
  41.         </dependency>
  42.         <!--RocketMQ相关-->
  43.         <dependency>
  44.             <groupId>org.apache.rocketmq</groupId>
  45.             <artifactId>rocketmq-spring-boot-starter</artifactId>
  46.             <version>2.0.3</version>
  47.         </dependency>
  48.         <dependency>
  49.             <groupId>org.apache.rocketmq</groupId>
  50.             <artifactId>rocketmq-client</artifactId>
  51.             <version>4.6.0</version>
  52.         </dependency>
  53.         <dependency>
  54.             <groupId>org.projectlombok</groupId>
  55.             <artifactId>lombok</artifactId>
  56.         </dependency>
  57.         <dependency>
  58.             <groupId>joda-time</groupId>
  59.             <artifactId>joda-time</artifactId>
  60.         </dependency>
  61.     </dependencies>
  63. </project>

1.6.2、配置文件

application.properties

  1. spring.application.name = itcast-rocketmq
  2. server.port = 18082
  4. # RocketMQ相关配置
  5. rocketmq.name-server=172.16.55.155:9876
  6. rocketmq.producer.group=tanhua
  8. # mongodb相关配置
  9. #spring.data.mongodb.uri=mongodb://192.168.31.81:27017/tanhua
  11. #设置了密码的mongodb配置方式
  12. spring.data.mongodb.username=tanhua
  13. spring.data.mongodb.password=l3SCjl0HvmSkTtiSbN0Swv40spYnHhDV
  14. spring.data.mongodb.authentication-database=admin
  15. spring.data.mongodb.database=tanhua
  16. spring.data.mongodb.port=27017
  17. spring.data.mongodb.host=192.168.31.81

log4j.properties

  1. log4j.rootLogger=DEBUG,A1
  3. log4j.appender.A1=org.apache.log4j.ConsoleAppender
  4. log4j.appender.A1.layout=org.apache.log4j.PatternLayout
  5. log4j.appender.A1.layout.ConversionPattern=[%t] [%c]-[%p] %m%n

1.6.3、RecommendQuanZi

存储到MongoDB的中的实体结构。

  1. package com.tanhua.recommend.pojo;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  6. import org.bson.types.ObjectId;
  8. @Data
  9. @NoArgsConstructor
  10. @AllArgsConstructor
  11. public class RecommendQuanZi {
  13.     private ObjectId id;
  14.     private Long userId;// 用户id
  15.     private Long publishId; //动态id，需要转化为Long类型
  16.     private Double score; //得分
  17.     private Long date; //时间戳
  18. }

1.6.4、QuanZiMsgConsumer

  1. package com.tanhua.recommend.msg;
  3. import com.fasterxml.jackson.databind.JsonNode;
  4. import com.fasterxml.jackson.databind.ObjectMapper;
  5. import com.tanhua.dubbo.server.pojo.Publish;
  6. import com.tanhua.recommend.pojo.RecommendQuanZi;
  7. import org.apache.commons.lang3.StringUtils;
  8. import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
  9. import org.apache.rocketmq.spring.core.RocketMQListener;
  10. import org.bson.types.ObjectId;
  11. import org.joda.time.DateTime;
  12. import org.slf4j.Logger;
  13. import org.slf4j.LoggerFactory;
  14. import org.springframework.beans.factory.annotation.Autowired;
  15. import org.springframework.data.mongodb.core.MongoTemplate;
  16. import org.springframework.stereotype.Component;
  17. import org.springframework.util.CollectionUtils;
  19. @Component
  20. @RocketMQMessageListener(topic = "tanhua-quanzi",
  21.         consumerGroup = "tanhua-quanzi-consumer")
  22. public class QuanZiMsgConsumer implements RocketMQListener<String> {
  24.     private static final ObjectMapper MAPPER = new ObjectMapper();
  26.     private static final Logger LOGGER = LoggerFactory.getLogger(QuanZiMsgConsumer.class);
  28.     @Autowired
  29.     private MongoTemplate mongoTemplate;
  31.     @Override
  32.     public void onMessage(String msg) {
  33.         try {
  34.             JsonNode jsonNode = MAPPER.readTree(msg);
  36.             Long userId = jsonNode.get("userId").asLong();
  37.             Long pid = jsonNode.get("pid").asLong();
  38.             String publishId = jsonNode.get("publishId").asText();
  39.             Integer type = jsonNode.get("type").asInt();
  41.             //1-发动态，2-浏览动态， 3-点赞， 4-喜欢， 5-评论，6-取消点赞，7-取消喜欢
  42.             RecommendQuanZi recommendQuanZi = new RecommendQuanZi();
  43.             recommendQuanZi.setUserId(userId);
  44.             recommendQuanZi.setId(ObjectId.get());
  45.             recommendQuanZi.setDate(System.currentTimeMillis());
  46.             recommendQuanZi.setPublishId(pid);
  48.             switch (type) {
  49.                 case 1: {
  50.                     int score = 0;
  51.                     Publish publish = this.mongoTemplate.findById(new ObjectId(publishId), Publish.class);
  52.                     int count = StringUtils.length(publish.getText());
  53.                     if (count >= 0 && count <= 50) {
  54.                         score += 1;
  55.                     } else if (count <= 100) {
  56.                         score += 2;
  57.                     } else {
  58.                         score += 3;
  59.                     }
  61.                     if (!CollectionUtils.isEmpty(publish.getMedias())) {
  62.                         score += publish.getMedias().size();
  63.                     }
  65.                     recommendQuanZi.setScore(Double.valueOf(score));
  67.                     break;
  68.                 }
  69.                 case 2: {
  70.                     recommendQuanZi.setScore(1d);
  71.                     break;
  72.                 }
  73.                 case 3: {
  74.                     recommendQuanZi.setScore(5d);
  75.                     break;
  76.                 }
  77.                 case 4: {
  78.                     recommendQuanZi.setScore(8d);
  79.                     break;
  80.                 }
  81.                 case 5: {
  82.                     recommendQuanZi.setScore(10d);
  83.                     break;
  84.                 }
  85.                 case 6: {
  86.                     recommendQuanZi.setScore(-5d);
  87.                     break;
  88.                 }
  89.                 case 7: {
  90.                     recommendQuanZi.setScore(-8d);
  91.                     break;
  92.                 }
  93.                 default: {
  94.                     recommendQuanZi.setScore(0d);
  95.                     break;
  96.                 }
  97.             }
  99. //            String collectionName = "recommend_quanzi_" + new DateTime().toString("yyyyMMdd");
  100.             //为了方便测试，将数据写到一张表
  101.             String collectionName = "recommend_quanzi";
  102.             this.mongoTemplate.save(recommendQuanZi, collectionName);
  104.         } catch (Exception e) {
  105.             LOGGER.error("处理消息失败~" + msg, e);
  106.         }
  107.     }
  108. }

1.7、测试

1.7.1、发布动态

day07-圈子推荐功能实现 - 图3

发布4张图片：

day07-圈子推荐功能实现 - 图4

数据：

day07-圈子推荐功能实现 - 图5

消息处理：

day07-圈子推荐功能实现 - 图6

1.7.2、浏览动态

day07-圈子推荐功能实现 - 图7

消息处理：

day07-圈子推荐功能实现 - 图8

1.7.3、点赞

day07-圈子推荐功能实现 - 图9

消息处理：

day07-圈子推荐功能实现 - 图10

1.7.4、取消点赞

day07-圈子推荐功能实现 - 图11

消息处理：

day07-圈子推荐功能实现 - 图12

1.7.5、喜欢

day07-圈子推荐功能实现 - 图13

消息处理：

day07-圈子推荐功能实现 - 图14

1.7.6、取消喜欢

day07-圈子推荐功能实现 - 图15

消息处理：

day07-圈子推荐功能实现 - 图16

1.7.7、评论

day07-圈子推荐功能实现 - 图17

消息处理：

day07-圈子推荐功能实现 - 图18

2、推荐系统

在推荐系统中，我们将基于前面写入到推荐表中的数据通过Spark进行计算，在Spark计算完成后将结果写入到Redis中，以供在业务系统中进行查询。

2.1、导入数据

使用资料中提供的centos7镜像或将资料中的mongodb数据导入到现有的mongodb服务中。

导入说明：

  • 将资料目录中的mongodb.tar.gz上传至mongodb所在服务器
  • 将其解压到/var/lib/docker/volumes目录下，原有mongodb目录要删除

  • 重启mongodb服务

    • docker restart mongodb
  • 使用客户端连接到mongodb服务进行查看
  • 数据中所关联的用户数据在tanhua.sql文件中，将其替换原有的数据

mongodb：

day07-圈子推荐功能实现 - 图19

mysql数据：

day07-圈子推荐功能实现 - 图20

2.2、部署圈子推荐服务

推荐服务我们将基于docker的形式进行部署：

  1. #拉取镜像
  2. docker pull registry.cn-hangzhou.aliyuncs.com/itcast/tanhua-spark-quanzi:1.0
  4. #创建容器
  5. docker create --name tanhua-spark-quanzi --restart=always \
  6. --env MONGODB_HOST=192.168.31.81 \
  7. --env MONGODB_PORT=27017 \
  8. --env MONGODB_USERNAME=tanhua \
  9. --env MONGODB_PASSWORD=l3SCjl0HvmSkTtiSbN0Swv40spYnHhDV \
  10. --env MONGODB_DATABASE=tanhua \
  11. --env MONGODB_COLLECTION=recommend_quanzi \
  12. --env SCHEDULE_PERIOD=3 \
  13. --env REDIS_NODES="192.168.31.81:6379,192.168.31.81:6380,192.168.31.81:6381" \
  14. registry.cn-hangzhou.aliyuncs.com/itcast/tanhua-spark-quanzi:1.0
  16. #参数说明
  17. #MONGODB_HOST mongodb服务的地址
  18. #MONGODB_PORT mongodb服务的端口
  19. #MONGODB_USERNAME mongodb服务的认证用户名
  20. #MONGODB_PASSWORD mongodb服务的认证密码
  21. #MONGODB_DATABASE mongodb连接的数据库
  22. #MONGODB_COLLECTION 操作表
  23. #SCHEDULE_PERIOD  下次执行时间间隔，但是为分，默认为10分钟
  24. #REDIS_NODES  redis集群地址，也可以使用单节点
  26. #mongodb开启认证服务
  27. #docker create --name mongodb --restart=always -p 27017:27017 -v mongodb:/data/db mongo:4.0.3 --auth
  29. #启动服务，启动之后就会进行执行，在SCHEDULE_PERIOD时间后再次执行
  30. docker start tanhua-spark-quanzi
  32. #查看日志
  33. docker logs -f tanhua-spark-quanzi
  35. #执行完成后会将数据写入到redis中

2.3、测试

进入redis查看是否已经有数据：

day07-圈子推荐功能实现 - 图21

3、修改查询逻辑

之前是通过MongoDB直接查询，而现在需要先从Redis进行命中，如果未命中则需要进行MongoDB查询。

修改server工程中的MovementsService类型：

  1.  /**
  2.      * 查询动态
  3.      *
  4.      * @param page
  5.      * @param pageSize
  6.      * @return
  7.      */
  8.     public PageResult queryPublishList(Integer page, Integer pageSize, boolean isRecommend) {
  9.         PageResult pageResult = new PageResult();
  10.         //获取当前的登录信息
  11.         User user = UserThreadLocal.get();
  14.         PageInfo<Publish> pageInfo = null;
  15.         if (isRecommend) { //推荐动态逻辑处理
  16.             // 查询Redis
  17.             String value = this.redisTemplate.opsForValue().get("QUANZI_PUBLISH_RECOMMEND_" + user.getId());
  18.             if (StringUtils.isNotEmpty(value)) {
  19.                 String[] pids = StringUtils.split(value, ',');
  20.                 int startIndex = (page - 1) * pageSize;
  21.                 if(startIndex < pids.length){
  22.                     int endIndex = startIndex + pageSize - 1;
  23.                     if (endIndex >= pids.length) {
  24.                         endIndex = pids.length - 1;
  25.                     }
  27.                     List<Long> pidList = new ArrayList<>();
  28.                     for (int i = startIndex; i <= endIndex; i++) {
  29.                         pidList.add(Long.valueOf(pids[i]));
  30.                     }
  32.                     List<Publish> publishList = this.quanZiApi.queryPublishByPids(pidList);
  33.                     pageInfo = new PageInfo<>();
  34.                     pageInfo.setRecords(publishList);
  35.                 }
  36.             }
  37.         }
  39.         if (null == pageInfo) {
  40.             Long userId = isRecommend ? null : user.getId();
  41.             pageInfo = this.quanZiApi.queryPublishList(userId, page, pageSize);
  42.         }
  44.         pageResult.setPagesize(pageSize);
  45.         pageResult.setPage(page);
  46.         pageResult.setCounts(0);
  47.         pageResult.setPages(0);
  49.         List<Publish> records = pageInfo.getRecords();
  51.         if (CollectionUtils.isEmpty(records)) {
  52.             //没有动态信息
  53.             return pageResult;
  54.         }
  56.         List<Movements> movementsList = new ArrayList<>();
  57.         for (Publish record : records) {
  58.             Movements movements = new Movements();
  60.             movements.setId(record.getId().toHexString());
  61.             movements.setImageContent(record.getMedias().toArray(new String[]{}));
  62.             movements.setTextContent(record.getText());
  63.             movements.setUserId(record.getUserId());
  64.             movements.setCreateDate(RelativeDateFormat.format(new Date(record.getCreated())));
  66.             movementsList.add(movements);
  67.         }
  69.         List<Long> userIds = new ArrayList<>();
  70.         for (Movements movements : movementsList) {
  71.             if (!userIds.contains(movements.getUserId())) {
  72.                 userIds.add(movements.getUserId());
  73.             }
  75.         }
  77.         QueryWrapper<UserInfo> queryWrapper = new QueryWrapper<>();
  78.         queryWrapper.in("user_id", userIds);
  79.         List<UserInfo> userInfos = this.userInfoService.queryList(queryWrapper);
  80.         for (Movements movements : movementsList) {
  81.             for (UserInfo userInfo : userInfos) {
  82.                 if (movements.getUserId().longValue() == userInfo.getUserId().longValue()) {
  83.                     this.fillValueToMovements(movements, userInfo);
  84.                     break;
  85.                 }
  86.             }
  87.         }
  89.         pageResult.setItems(movementsList);
  90.         return pageResult;
  91.     }

测试： day07-圈子推荐功能实现 - 图22

day07-圈子推荐功能实现 - 图23

可以看到，已经查询到了动态数据。

4、小视频推荐

小视频的推荐和动态推荐的实现逻辑非常的类似。

4.1、增加自增id

  1. package com.tanhua.dubbo.server.pojo;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  6. import org.bson.types.ObjectId;
  7. import org.springframework.data.mongodb.core.mapping.Document;
  9. import java.util.List;
  11. @Data
  12. @NoArgsConstructor
  13. @AllArgsConstructor
  14. @Document(collection = "video")
  15. public class Video implements java.io.Serializable {
  17.     private static final long serialVersionUID = -3136732836884933873L;
  19.     private ObjectId id; //主键id
  20.     private Long vid;
  21.     private Long userId;
  22.     private String text; //文字
  23.     private String picUrl; //视频封面文件
  24.     private String videoUrl; //视频文件
  25.     private Long created; //创建时间
  26.     private Integer seeType; // 谁可以看，1-公开，2-私密，3-部分可见，4-不给谁看
  27.     private List<Long> seeList; //部分可见的列表
  28.     private List<Long> notSeeList; //不给谁看的列表
  29.     private String longitude; //经度
  30.     private String latitude; //纬度
  31.     private String locationName; //位置名称
  32. }

修改VideoApiImpl逻辑：

  1.     @Override
  2.     public Boolean saveVideo(Video video) {
  3.         if (video.getUserId() == null) {
  4.             return false;
  5.         }
  7.         video.setId(ObjectId.get());
  8.         video.setCreated(System.currentTimeMillis());
  10.         //生成vid
  11.         video.setVid(this.idService.createId("video", video.getId().toHexString()));
  13.         this.mongoTemplate.save(video);
  14.         return true;
  15.     }

4.2、动态计分规则

  • 发布+2
  • 点赞 +5
  • 评论 + 10

4.3、发送消息

4.3.1、VideoMQService

  1. package com.tanhua.server.service;
  3. import com.alibaba.dubbo.config.annotation.Reference;
  4. import com.tanhua.dubbo.server.api.QuanZiApi;
  5. import com.tanhua.dubbo.server.api.VideoApi;
  6. import com.tanhua.dubbo.server.pojo.Publish;
  7. import com.tanhua.dubbo.server.pojo.Video;
  8. import com.tanhua.server.pojo.User;
  9. import com.tanhua.server.utils.UserThreadLocal;
  10. import org.apache.rocketmq.spring.core.RocketMQTemplate;
  11. import org.slf4j.Logger;
  12. import org.slf4j.LoggerFactory;
  13. import org.springframework.beans.factory.annotation.Autowired;
  14. import org.springframework.stereotype.Service;
  16. import java.util.HashMap;
  17. import java.util.Map;
  19. @Service
  20. public class VideoMQService {
  22.     private static final Logger LOGGER = LoggerFactory.getLogger(VideoMQService.class);
  24.     @Autowired
  25.     private RocketMQTemplate rocketMQTemplate;
  27.     @Reference(version = "1.0.0")
  28.     private VideoApi videoApi;
  30.     /**
  31.      * 发布小视频消息
  32.      *
  33.      * @return
  34.      */
  35.     public Boolean videoMsg(String videoId) {
  36.         return this.sendMsg(videoId, 1);
  37.     }
  39.     /**
  40.      * 点赞小视频
  41.      *
  42.      * @return
  43.      */
  44.     public Boolean likeVideoMsg(String videoId) {
  45.         return this.sendMsg(videoId, 2);
  46.     }
  48.     /**
  49.      * 取消点赞小视频
  50.      *
  51.      * @return
  52.      */
  53.     public Boolean disLikeVideoMsg(String videoId) {
  54.         return this.sendMsg(videoId, 3);
  55.     }
  57.     /**
  58.      * 评论小视频
  59.      *
  60.      * @return
  61.      */
  62.     public Boolean commentVideoMsg(String videoId) {
  63.         return this.sendMsg(videoId, 4);
  64.     }
  66.     /**
  67.      * 发送小视频操作相关的消息
  68.      *
  69.      * @param videoId
  70.      * @param type     1-发动态，2-点赞， 3-取消点赞，4-评论
  71.      * @return
  72.      */
  73.     private Boolean sendMsg(String videoId, Integer type) {
  74.         try {
  75.             User user = UserThreadLocal.get();
  77.             Video video = this.videoApi.queryVideoById(videoId);
  79.             //构建消息
  80.             Map<String, Object> msg = new HashMap<>();
  81.             msg.put("userId", user.getId());
  82.             msg.put("date", System.currentTimeMillis());
  83.             msg.put("videoId", videoId);
  84.             msg.put("vid", video.getVid());
  85.             msg.put("type", type);
  87.             this.rocketMQTemplate.convertAndSend("tanhua-video", msg);
  88.         } catch (Exception e) {
  89.             LOGGER.error("发送消息失败! videoId = " + videoId + ", type = " + type, e);
  90.             return false;
  91.         }
  93.         return true;
  94.     }
  95. }

4.3.2、VideoController

  1. package com.tanhua.server.controller;
  3. import com.tanhua.server.service.MovementsService;
  4. import com.tanhua.server.service.VideoMQService;
  5. import com.tanhua.server.service.VideoService;
  6. import com.tanhua.server.vo.PageResult;
  7. import org.apache.commons.lang3.StringUtils;
  8. import org.springframework.beans.factory.annotation.Autowired;
  9. import org.springframework.http.HttpStatus;
  10. import org.springframework.http.ResponseEntity;
  11. import org.springframework.web.bind.annotation.*;
  12. import org.springframework.web.multipart.MultipartFile;
  14. import java.util.Map;
  16. @RestController
  17. @RequestMapping("smallVideos")
  18. public class VideoController {
  20.     @Autowired
  21.     private VideoService videoService;
  23.     @Autowired
  24.     private MovementsService movementsService;
  26.     @Autowired
  27.     private CommentsController commentsController;
  29.     @Autowired
  30.     private VideoMQService videoMQService;
  32.     /**
  33.      * 发布小视频
  34.      *
  35.      * @param picFile
  36.      * @param videoFile
  37.      * @return
  38.      */
  39.     @PostMapping
  40.     public ResponseEntity<Void> saveVideo(@RequestParam(value = "videoThumbnail", required = false) MultipartFile picFile,
  41.                                           @RequestParam(value = "videoFile", required = false) MultipartFile videoFile) {
  42.         try {
  43.             String id = this.videoService.saveVideo(picFile, videoFile);
  44.             if (StringUtils.isNotEmpty(id)) {
  45.                 //发送消息
  46.                 this.videoMQService.videoMsg(id);
  48.                 return ResponseEntity.ok(null);
  49.             }
  50.         } catch (Exception e) {
  51.             e.printStackTrace();
  52.         }
  53.         return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
  54.     }
  56.     /**
  57.      * 查询小视频列表
  58.      *
  59.      * @param page
  60.      * @param pageSize
  61.      * @return
  62.      */
  63.     @GetMapping
  64.     public ResponseEntity<PageResult> queryVideoList(@RequestParam(value = "page", defaultValue = "1") Integer page,
  65.                                                      @RequestParam(value = "pagesize", defaultValue = "10") Integer pageSize) {
  66.         try {
  67.             if (page <= 0) {
  68.                 page = 1;
  69.             }
  70.             PageResult pageResult = this.videoService.queryVideoList(page, pageSize);
  71.             if (null != pageResult) {
  72.                 return ResponseEntity.ok(pageResult);
  73.             }
  74.         } catch (Exception e) {
  75.             e.printStackTrace();
  76.         }
  77.         return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
  78.     }
  80.     /**
  81.      * 视频点赞
  82.      *
  83.      * @param videoId 视频id
  84.      * @return
  85.      */
  86.     @PostMapping("/{id}/like")
  87.     public ResponseEntity<Long> likeComment(@PathVariable("id") String videoId) {
  88.         try {
  89.             Long likeCount = this.movementsService.likeComment(videoId);
  90.             if (likeCount != null) {
  91.                 this.videoMQService.likeVideoMsg(videoId);
  92.                 return ResponseEntity.ok(likeCount);
  93.             }
  94.         } catch (Exception e) {
  95.             e.printStackTrace();
  96.         }
  97.         return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
  98.     }
  100.     /**
  101.      * 取消点赞
  102.      *
  103.      * @param videoId
  104.      * @return
  105.      */
  106.     @PostMapping("/{id}/dislike")
  107.     public ResponseEntity<Long> disLikeComment(@PathVariable("id") String videoId) {
  108.         try {
  109.             Long likeCount = this.movementsService.cancelLikeComment(videoId);
  110.             if (null != likeCount) {
  111.                 this.videoMQService.disLikeVideoMsg(videoId);
  112.                 return ResponseEntity.ok(likeCount);
  113.             }
  114.         } catch (Exception e) {
  115.             e.printStackTrace();
  116.         }
  117.         return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
  118.     }
  120.     /**
  121.      * 评论点赞
  122.      *
  123.      * @param publishId
  124.      * @return
  125.      */
  126.     @PostMapping("/comments/{id}/like")
  127.     public ResponseEntity<Long> commentsLikeComment(@PathVariable("id") String publishId) {
  128.         try {
  129.             Long likeCount = this.movementsService.likeComment(publishId);
  130.             if (likeCount != null) {
  131.                 return ResponseEntity.ok(likeCount);
  132.             }
  133.         } catch (Exception e) {
  134.             e.printStackTrace();
  135.         }
  136.         return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
  137.     }
  139.     /**
  140.      * 评论取消点赞
  141.      *
  142.      * @param publishId
  143.      * @return
  144.      */
  145.     @PostMapping("/comments/{id}/dislike")
  146.     public ResponseEntity<Long> disCommentsLikeComment(@PathVariable("id") String publishId) {
  147.         try {
  148.             Long likeCount = this.movementsService.cancelLikeComment(publishId);
  149.             if (null != likeCount) {
  150.                 return ResponseEntity.ok(likeCount);
  151.             }
  152.         } catch (Exception e) {
  153.             e.printStackTrace();
  154.         }
  155.         return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
  156.     }
  158.     /**
  159.      * 提交评论
  160.      *
  161.      * @param param
  162.      * @param videoId
  163.      * @return
  164.      */
  165.     @PostMapping("/{id}/comments")
  166.     public ResponseEntity<Void> saveComments(@RequestBody Map<String, String> param,
  167.                                              @PathVariable("id") String videoId) {
  168.         param.put("movementId", videoId);
  169.         ResponseEntity<Void> entity = this.commentsController.saveComments(param);
  171.         if (entity.getStatusCode().is2xxSuccessful()) {
  172.             //发送消息
  173.             this.videoMQService.commentVideoMsg(videoId);
  174.         }
  176.         return entity;
  177.     }
  179.     /**
  180.      * 评论列表
  181.      */
  182.     @GetMapping("/{id}/comments")
  183.     public ResponseEntity<PageResult> queryCommentsList(@PathVariable("id") String videoId,
  184.                                                         @RequestParam(value = "page", defaultValue = "1") Integer page,
  185.                                                         @RequestParam(value = "pagesize", defaultValue = "10") Integer pagesize) {
  186.         return this.commentsController.queryCommentsList(videoId, page, pagesize);
  187.     }
  189.     /**
  190.      * 视频用户关注
  191.      */
  192.     @PostMapping("/{id}/userFocus")
  193.     public ResponseEntity<Void> saveUserFocusComments(@PathVariable("id") Long userId) {
  194.         try {
  195.             Boolean bool = this.videoService.followUser(userId);
  196.             if (bool) {
  197.                 return ResponseEntity.ok(null);
  198.             }
  199.         } catch (Exception e) {
  200.             e.printStackTrace();
  201.         }
  202.         return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
  203.     }
  205.     /**
  206.      * 视频用户关注
  207.      */
  208.     @PostMapping("/{id}/userUnFocus")
  209.     public ResponseEntity<Void> saveUserUnFocusComments(@PathVariable("id") Long userId) {
  210.         try {
  211.             Boolean bool = this.videoService.disFollowUser(userId);
  212.             if (bool) {
  213.                 return ResponseEntity.ok(null);
  214.             }
  215.         } catch (Exception e) {
  216.             e.printStackTrace();
  217.         }
  218.         return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
  219.     }
  220. }

4.4、接收消息

4.4.1、RecommendVideo

  1. package com.tanhua.recommend.pojo;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  6. import org.bson.types.ObjectId;
  8. @Data
  9. @NoArgsConstructor
  10. @AllArgsConstructor
  11. public class RecommendVideo {
  13.     private ObjectId id;
  14.     private Long userId;// 用户id
  15.     private Long videoId; //视频id，需要转化为Long类型
  16.     private Double score; //得分
  17.     private Long date; //时间戳
  18. }

4.4.2、VideoMsgConsumer

  1. package com.tanhua.recommend.msg;
  3. import com.fasterxml.jackson.databind.JsonNode;
  4. import com.fasterxml.jackson.databind.ObjectMapper;
  5. import com.tanhua.recommend.pojo.Publish;
  6. import com.tanhua.recommend.pojo.RecommendQuanZi;
  7. import com.tanhua.recommend.pojo.RecommendVideo;
  8. import org.apache.commons.lang3.StringUtils;
  9. import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
  10. import org.apache.rocketmq.spring.core.RocketMQListener;
  11. import org.bson.types.ObjectId;
  12. import org.joda.time.DateTime;
  13. import org.slf4j.Logger;
  14. import org.slf4j.LoggerFactory;
  15. import org.springframework.beans.factory.annotation.Autowired;
  16. import org.springframework.data.mongodb.core.MongoTemplate;
  17. import org.springframework.stereotype.Component;
  18. import org.springframework.util.CollectionUtils;
  20. @Component
  21. @RocketMQMessageListener(topic = "tanhua-video",
  22.         consumerGroup = "tanhua-video-consumer")
  23. public class VideoMsgConsumer implements RocketMQListener<String> {
  25.     private static final ObjectMapper MAPPER = new ObjectMapper();
  27.     private static final Logger LOGGER = LoggerFactory.getLogger(VideoMsgConsumer.class);
  29.     @Autowired
  30.     private MongoTemplate mongoTemplate;
  32.     @Override
  33.     public void onMessage(String msg) {
  34.         try {
  35.             JsonNode jsonNode = MAPPER.readTree(msg);
  37.             Long userId = jsonNode.get("userId").asLong();
  38.             Long vid = jsonNode.get("vid").asLong();
  39.             Integer type = jsonNode.get("type").asInt();
  41.             //1-发动态，2-点赞， 3-取消点赞，4-评论
  42.             RecommendVideo recommendVideo = new RecommendVideo();
  43.             recommendVideo.setUserId(userId);
  44.             recommendVideo.setId(ObjectId.get());
  45.             recommendVideo.setDate(System.currentTimeMillis());
  46.             recommendVideo.setVideoId(vid);
  48.             switch (type) {
  49.                 case 1: {
  50.                     recommendVideo.setScore(2d);
  51.                     break;
  52.                 }
  53.                 case 2: {
  54.                     recommendVideo.setScore(5d);
  55.                     break;
  56.                 }
  57.                 case 3: {
  58.                     recommendVideo.setScore(-5d);
  59.                     break;
  60.                 }
  61.                 case 4: {
  62.                     recommendVideo.setScore(10d);
  63.                     break;
  64.                 }
  65.                 default: {
  66.                     recommendVideo.setScore(0d);
  67.                     break;
  68.                 }
  69.             }
  71.             String collectionName = "recommend_video_" + new DateTime().toString("yyyyMMdd");
  72.             this.mongoTemplate.save(recommendVideo, collectionName);
  74.         } catch (Exception e) {
  75.             LOGGER.error("处理小视频消息失败~" + msg, e);
  76.         }
  77.     }
  78. }

4.4.3、测试

day07-圈子推荐功能实现 - 图24

4.5、部署推荐服务

  1. #拉取镜像
  2. docker pull registry.cn-hangzhou.aliyuncs.com/itcast/tanhua-spark-video:1.0
  4. #创建容器
  5. docker create --name tanhua-spark-video --restart=always \
  6. --env MONGODB_HOST=192.168.31.81 \
  7. --env MONGODB_PORT=27017 \
  8. --env MONGODB_USERNAME=tanhua \
  9. --env MONGODB_PASSWORD=l3SCjl0HvmSkTtiSbN0Swv40spYnHhDV \
  10. --env MONGODB_DATABASE=tanhua \
  11. --env MONGODB_COLLECTION=recommend_video \
  12. --env SCHEDULE_PERIOD=3 \
  13. --env REDIS_NODES="192.168.31.81:6379,192.168.31.81:6380,192.168.31.81:6381" \
  14. registry.cn-hangzhou.aliyuncs.com/itcast/tanhua-spark-video:1.0
  16. #启动服务
  17. docker start tanhua-spark-video
  19. #查看日志
  20. docker logs -f tanhua-spark-video

测试：

day07-圈子推荐功能实现 - 图25

4.6、修改查询逻辑

修改VideoService的实现：

  1. public PageResult queryVideoList(Integer page, Integer pageSize) {
  3.         User user = UserThreadLocal.get();
  5.         PageResult pageResult = new PageResult();
  6.         pageResult.setPage(page);
  7.         pageResult.setPagesize(pageSize);
  8.         pageResult.setPages(0);
  9.         pageResult.setCounts(0);
  11.         PageInfo<Video> pageInfo = null;
  13.         //先从Redis进行命中，如果命中则返回推荐列表，如果未命中查询默认列表
  14.         String redisValue = this.redisTemplate.opsForValue().get("QUANZI_VIDEO_RECOMMEND_" + user.getId());
  15.         if (StringUtils.isNotEmpty(redisValue)) {
  16.             String[] pids = StringUtils.split(redisValue, ',');
  17.             int startIndex = (page - 1) * pageSize;
  18.             if (startIndex < pids.length) {
  19.                 int endIndex = startIndex + pageSize - 1;
  20.                 if (endIndex >= pids.length) {
  21.                     endIndex = pids.length - 1;
  22.                 }
  24.                 List<Long> vidList = new ArrayList<>();
  25.                 for (int i = startIndex; i <= endIndex; i++) {
  26.                     vidList.add(Long.valueOf(pids[i]));
  27.                 }
  29.                 List<Video> videoList = this.videoApi.queryVideoListByPids(vidList);
  30.                 pageInfo = new PageInfo<>();
  31.                 pageInfo.setRecords(videoList);
  32.             }
  33.         }
  35.         if(null == pageInfo){
  36.             pageInfo = this.videoApi.queryVideoList(page, pageSize);
  37.         }
  39.         List<Video> records = pageInfo.getRecords();
  40.         List<VideoVo> videoVoList = new ArrayList<>();
  41.         List<Long> userIds = new ArrayList<>();
  42.         for (Video record : records) {
  43.             VideoVo videoVo = new VideoVo();
  45.             videoVo.setUserId(record.getUserId());
  46.             videoVo.setCover(record.getPicUrl());
  47.             videoVo.setVideoUrl(record.getVideoUrl());
  48.             videoVo.setId(record.getId().toHexString());
  49.             videoVo.setSignature("我就是我~");
  51.             Long commentCount = this.quanZiApi.queryCommentCount(videoVo.getId(), 2);
  52.             videoVo.setCommentCount(commentCount == null ? 0 : commentCount.intValue()); // 评论数
  54.             String followUserKey = "VIDEO_FOLLOW_USER_" + user.getId() + "_" + videoVo.getUserId();
  55.             videoVo.setHasFocus(this.redisTemplate.hasKey(followUserKey) ? 1 : 0); //是否关注
  57.             String userKey = "QUANZI_COMMENT_LIKE_USER_" + user.getId() + "_" + videoVo.getId();
  58.             videoVo.setHasLiked(this.redisTemplate.hasKey(userKey) ? 1 : 0); //是否点赞（1是，0否）
  60.             String key = "QUANZI_COMMENT_LIKE_" + videoVo.getId();
  61.             String value = this.redisTemplate.opsForValue().get(key);
  62.             if (StringUtils.isNotEmpty(value)) {
  63.                 videoVo.setLikeCount(Integer.valueOf(value)); //点赞数
  64.             } else {
  65.                 videoVo.setLikeCount(0);
  66.             }
  68.             if (!userIds.contains(record.getUserId())) {
  69.                 userIds.add(record.getUserId());
  70.             }
  72.             videoVoList.add(videoVo);
  73.         }
  75.         QueryWrapper<UserInfo> queryWrapper = new QueryWrapper();
  76.         queryWrapper.in("user_id", userIds);
  77.         List<UserInfo> userInfos = this.userInfoService.queryList(queryWrapper);
  78.         for (VideoVo videoVo : videoVoList) {
  79.             for (UserInfo userInfo : userInfos) {
  80.                 if (videoVo.getUserId().longValue() == userInfo.getUserId().longValue()) {
  82.                     videoVo.setNickname(userInfo.getNickName());
  83.                     videoVo.setAvatar(userInfo.getLogo());
  85.                     break;
  86.                 }
  87.             }
  89.         }
  91.         pageResult.setItems(videoVoList);
  92.         return pageResult;
  93.     }