Redis实现朋友圈，微博等Feed流功能,实现Feed流微服务（代码实现）

共饮一杯无

发布于 2022-12-18 18:23:47

6060

发布于 2022-12-18 18:23:47

上篇博客讲述了Redis实现朋友圈，微博等Feed流功能,实现Feed流微服务的业务场景、实现思路和环境搭建，本文继续讲解具体的代码实现内容。

文章目

- [添加 Feed 信息](https://cloud.tencent.com/developer)
    - [FeedsController](https://cloud.tencent.com/developer)
    - [FeedsService](https://cloud.tencent.com/developer)
    - [FeedsMapper](https://cloud.tencent.com/developer)
    - [ms-follow 服务新增获取粉丝列表](https://cloud.tencent.com/developer)
    - [ms-gateway 服务配置网关路由](https://cloud.tencent.com/developer)
    - [启动项目测试](https://cloud.tencent.com/developer)
- [删除 Feed 信息](https://cloud.tencent.com/developer)
    - [FeedsController](https://cloud.tencent.com/developer)
    - [FeedsService](https://cloud.tencent.com/developer)
    - [FeedsMapper](https://cloud.tencent.com/developer)
    - [启动项目测试](https://cloud.tencent.com/developer)
- [关注/取关时处理用户 Feed](https://cloud.tencent.com/developer)
    - [FeedsController](https://cloud.tencent.com/developer)
    - [FeedsService](https://cloud.tencent.com/developer)
    - [FeedsMapper](https://cloud.tencent.com/developer)
    - [ms-follow 服务关注取关时变更 Feed](https://cloud.tencent.com/developer)
        - [FollowService新增关注/取关时Feed逻辑](https://cloud.tencent.com/developer)
    - [启动项目测试](https://cloud.tencent.com/developer)
        - [用户8，9，10都关注了用户7](https://cloud.tencent.com/developer)
        - [用户10取消关注用户7](https://cloud.tencent.com/developer)
        - [用户11关注用户7](https://cloud.tencent.com/developer)
- [分页获取关注的 Feed 数据](https://cloud.tencent.com/developer)
    - [构建返回的FeedsVO](https://cloud.tencent.com/developer)
    - [FeedsController](https://cloud.tencent.com/developer)
    - [FeedsService](https://cloud.tencent.com/developer)
    - [FeedsMapper](https://cloud.tencent.com/developer)
    - [启动项目测试](https://cloud.tencent.com/developer)

添加 Feed 信息

FeedsController

    /**
     * 添加 Feed
     *
     * @param feeds
     * @param access_token
     * @return
     */
    @PostMapping
    public ResultInfo<String> create(@RequestBody Feeds feeds, String access_token) {
        feedsService.create(feeds, access_token);
        return ResultInfoUtil.buildSuccess(request.getServletPath(), "添加成功");
    }

FeedsService

    /**
     * 添加 Feed
     *
     * @param feeds
     * @param accessToken
     */
    @Transactional(rollbackFor = Exception.class)
    public void create(Feeds feeds, String accessToken) {
        // 校验 Feed 内容不能为空，不能太长
        AssertUtil.isNotEmpty(feeds.getContent(), "请输入内容");
        AssertUtil.isTrue(feeds.getContent().length() > 255, "输入内容太多，请重新输入");
        // 获取登录用户信息
        SignInUserInfo userInfo = loadSignInUserInfo(accessToken);
        // Feed 关联用户信息
        feeds.setFkUserId(userInfo.getId());
        // 添加 Feed
        int count = feedsMapper.save(feeds);
        AssertUtil.isTrue(count == 0, "添加失败");
        // 推送到粉丝的列表中 -- 后续这里应该采用异步消息队列解决性能问题
        // 先获取粉丝 id 集合
        List<Integer> followers = findFollowers(userInfo.getId());
        // 推送 Feed
        long now = System.currentTimeMillis();
        followers.forEach(follower -> {
            String key = RedisKeyConstant.following_feeds.getKey() + follower;
            redisTemplate.opsForZSet().add(key, feeds.getId(), now);
        });
    }

    /**
     * 获取粉丝 id 集合
     *
     * @param userId
     * @return
     */
    private List<Integer> findFollowers(Integer userId) {
        String url = followServerName + "followers/" + userId;
        ResultInfo resultInfo = restTemplate.getForObject(url, ResultInfo.class);
        if (resultInfo.getCode() != ApiConstant.SUCCESS_CODE) {
            throw new ParameterException(resultInfo.getCode(), resultInfo.getMessage());
        }
        List<Integer> followers = (List<Integer>) resultInfo.getData();
        return followers;
    }

FeedsMapper

    /**
     * 添加 Feed
     * @param feeds 
     * @return
     */
    @Insert("insert into t_feeds (content, fk_user_id, praise_amount, " +
            " comment_amount, fk_restaurant_id, create_date, update_date, is_valid) " +
            " values (#{content}, #{fkUserId}, #{praiseAmount}, #{commentAmount}, #{fkRestaurantId}, " +
            " now(), now(), 1)")
    @Options(useGeneratedKeys = true, keyProperty = "id")
    int save(Feeds feeds);

ms-follow 服务新增获取粉丝列表

FollowController

    /**
     * 获取粉丝列表
     *
     * @param userId
     * @return
     */
    @GetMapping("followers/{userId}")
    public ResultInfo findFollowers(@PathVariable Integer userId) {
        return ResultInfoUtil.buildSuccess(request.getServletPath(),
                followService.findFollowers(userId));
    }

FollowService

    /**
     * 获取粉丝列表
     *
     * @param userId
     * @return
     */
    public Set<Integer> findFollowers(Integer userId) {
        AssertUtil.isNotNull(userId, "请选择要查看的用户");
        Set<Integer> followers = redisTemplate.opsForSet()
                .members(RedisKeyConstant.followers.getKey() + userId);
        return followers;
    }

ms-gateway 服务配置网关路由

spring:
  application:
    name: ms-gateway
  cloud:
    gateway:
      discovery:
        locator:
          enabled: true # 开启配置注册中心进行路由功能
          lower-case-service-id: true # 将服务名称转小写
      routes:
          # Feed服务路由             
        - id: ms-feeds
          uri: lb://ms-feeds
          predicates:
            - Path=/feeds/**
          filters:
            - StripPrefix=1

启动项目测试

先让id等于10、9、8的用户关注id等于7的用户。
id等于7的用户登录后发布一条动态。

让 id=10 的用户关注 id=7 的用户：

让 id=9 的用户关注 id=7 的用户：

让 id=8 的用户关注 id=7 的用户：

id=7 的用户登录系统并发送一条动态：

http://localhost/feeds?access_token=48781f97-1c3a-4737-ae55-984c0944649e

查看数据库 feeds 信息：

查看 redis 中粉丝的 feeds 信息：

可以看到用户id为8、9、10的用户都收到了这条Feed。

删除 Feed 信息

FeedsController

    /**
     * 删除 Feed
     *
     * @param id
     * @param access_token
     * @return
     */
    @DeleteMapping("{id}")
    public ResultInfo delete(@PathVariable Integer id, String access_token) {
        feedsService.delete(id, access_token);
        return ResultInfoUtil.buildSuccess(request.getServletPath(), "删除成功");
    }

FeedsService

    /**
     * 删除 Feed
     *
     * @param id
     * @param accessToken
     */
    @Transactional(rollbackFor = Exception.class)
    public void delete(Integer id, String accessToken) {
        // 请选择要删除的 Feed
        AssertUtil.isTrue(id == null || id < 1, "请选择要删除的Feed");
        // 获取登录用户
        SignInUserInfo userInfo = loadSignInUserInfo(accessToken);
        // 获取 Feed 内容
        Feeds feeds = feedsMapper.findById(id);
        // 判断 Feed 是否已经被删除且只能删除自己的 Feed
        AssertUtil.isTrue(feeds == null, "该Feed已被删除");
        AssertUtil.isTrue(!feeds.getFkUserId().equals(userInfo.getId()),
                "只能删除自己的Feed");
        // 删除
        int count = feedsMapper.delete(id);
        if (count == 0) {
            return;
        }
        // 将内容从粉丝的集合中删除 -- 异步消息队列优化
        // 先获取我的粉丝
        List<Integer> followers = findFollowers(userInfo.getId());
        // 移除 Feed
        followers.forEach(follower -> {
            String key = RedisKeyConstant.following_feeds.getKey() + follower;
            redisTemplate.opsForZSet().remove(key, feeds.getId());
        });
    }

FeedsMapper

启动项目测试

数据库中的feeds：

用户只能删除自己创建的Feed，测试用id为6的用户删除id为14的Feed（该Feed是id为7的用户创建的）：

用id为7的用户登陆后，逻辑删除id=14的feeds：

删除后再次删除：

查看数据库中的feeds已经逻辑删除：

查看redis相关Feed也不存在了：

关注/取关时处理用户 Feed

当A用户关注B用户时，那么要实时的将B的所有Feed推送到A用户的Feed集合中，同样如果A用户取关B用户，那么要将B用户所有的Feed从A用户的Feed集合中移除。

FeedsController

    /**
     * 变更 Feed
     *
     * @return
     */
    @PostMapping("updateFollowingFeeds/{followingDinerId}")
    public ResultInfo addFollowingFeeds(@PathVariable Integer followingDinerId,
                                        String access_token, @RequestParam int type) {
        feedsService.addFollowingFeed(followingDinerId, access_token, type);
        return ResultInfoUtil.buildSuccess(request.getServletPath(), "操作成功");
    }

FeedsService

    /**
     * 变更 Feed
     *
     * @param followinguserId 关注的好友 ID
     * @param accessToken      登录用户token
     * @param type             1 关注 0 取关
     */
    @Transactional(rollbackFor = Exception.class)
    public void addFollowingFeed(Integer followinguserId, String accessToken, int type) {
        // 请选择关注的好友
        AssertUtil.isTrue(followinguserId == null || followinguserId < 1,
                "请选择关注的好友");
        // 获取登录用户信息
        SignInUserInfo userInfo = loadSignInUserInfo(accessToken);
        // 获取关注/取关的用户的所有 Feed
        List<Feeds> feedsList = feedsMapper.findByUserId(followinguserId);
        String key = RedisKeyConstant.following_feeds.getKey() + userInfo.getId();
        if (type == 0) {
            // 取关
            List<Integer> feedIds = feedsList.stream()
                    .map(feed -> feed.getId())
                    .collect(Collectors.toList());
            redisTemplate.opsForZSet().remove(key, feedIds.toArray(new Integer[]{}));
        } else {
            // 关注
            Set<ZSetOperations.TypedTuple> typedTuples =
                    feedsList.stream()
                            .map(feed -> new DefaultTypedTuple<>(feed.getId(), (double) feed.getUpdateDate().getTime()))
                            .collect(Collectors.toSet());
            redisTemplate.opsForZSet().add(key, typedTuples);
        }
    }

FeedsMapper

/**
* 根据用户 ID 查询 Feed
* @param userId
* @return
*/
@Select("select id, content, update_date from t_feeds " +
        " where fk_user_id = #{userId} and is_valid = 1")
    List<Feeds> findByUserId(@Param("userId") Integer userId);

ms-follow 服务关注取关时变更 Feed

添加调用ms-feeds服务的请求地址项目路径

FollowService新增关注/取关时Feed逻辑

    /**
     * 发送请求添加或者移除关注人的Feed列表
     *
     * @param followUserId 关注好友的ID
     * @param accessToken   当前登录用户token
     * @param type          0=取关 1=关注
     */
    private void sendSaveOrRemoveFeed(Integer followUserId, String accessToken, int type) {
        String feedsUpdateUrl = feedsServerName + "updateFollowingFeeds/"
                + followUserId + "?access_token=" + accessToken;
        // 构建请求头
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_FORM_URLENCODED);
        // 构建请求体（请求参数）
        MultiValueMap<String, Object> body = new LinkedMultiValueMap<>();
        body.add("type", type);
        HttpEntity<MultiValueMap<String, Object>> entity = new HttpEntity<>(body, headers);
        restTemplate.postForEntity(feedsUpdateUrl, entity, ResultInfo.class);
    }

启动项目测试

用户8，9，10都关注了用户7

那么用户7推送一条feeds(朋友圈) 的时候，他的粉丝用户8，9,10应该都可以看到，测试用户7发送feed：

查看数据库和redis：

用户10取消关注用户7

用户10的feeds集合中存储了关注用户的feeds :

让用户10取消关注用户7：

用户7的所有feeds(朋友圈) 应该从用户10的feeds集合中移除：

只剩下用户8、9相关的。

用户11关注用户7

用户7的所有feeds(朋友圈) 应该都添加到用户11的feeds集合中：

分页获取关注的 Feed 数据

当前数据库用户7发布了

构建返回的FeedsVO

/**
 *
 * Feed显示信息
 * @author zjq
 */
@Getter
@Setter
@ApiModel(description = "Feed显示信息")
public class FeedsVO implements Serializable {

    @ApiModelProperty("主键")
    private Integer id;
    @ApiModelProperty("内容")
    private String content;
    @ApiModelProperty("点赞数")
    private int praiseAmount;
    @ApiModelProperty("评论数")
    private int commentAmount;
    @ApiModelProperty("餐厅id")
    private Integer fkRestaurantId;
    @ApiModelProperty("用户ID")
    private Integer fkUserId;
    @ApiModelProperty("用户信息")
    private ShortUserInfo userInfo;
    @ApiModelProperty("显示时间")
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm")
    public Date createDate;

}

FeedsController

/**
* 分页获取关注的 Feed 数据
*
* @param page
* @param access_token
* @return
*/
@GetMapping("{page}")
    public ResultInfo selectForPage(@PathVariable Integer page, String access_token) {
    List<FeedsVO> feedsVOS = feedsService.selectForPage(page, access_token);
return ResultInfoUtil.buildSuccess(request.getServletPath(), feedsVOS);
}

FeedsService

登录用户每次发送朋友圈都会向粉丝的feeds集合中推送这条朋友圈，那么当粉丝就可以获取关注的人的所有feeds。

比如用户8,9,11关注了用户7，那么用户7发的5条朋友圈，用户8,9,11都能看到,用户8同时还跟用户6是好友，那么用户8可以同时看到用户7和用户6发送的3条朋友圈。

实现逻辑如下：

获取登录用户信息
构建分页查询的参数start，end
从Redis的sorted sets中按照score的降序进行读取Feed的id
从数据库中获取Feed的信息
构建Feed关联的用户信息（不是循环逐条读取，而是批量获取）

    /**
     * 根据时间由近至远，每次查询 6 条 Feed
     *
     * @param page
     * @param accessToken
     * @return
     */
    public List<FeedsVO> selectForPage(Integer page, String accessToken) {
        if (page == null) {
            page = 1;
        }
        // 获取登录用户
        SignInUserInfo userInfo = loadSignInUserInfo(accessToken);
        // 我关注的好友的 Feedkey
        String key = RedisKeyConstant.following_feeds.getKey() + userInfo.getId();
        // SortedSet 的 ZREVRANGE 命令是闭区间
        long start = (page - 1) * ApiConstant.PAGE_SIZE;
        long end = page * ApiConstant.PAGE_SIZE - 1;
        Set<Integer> feedIds = redisTemplate.opsForZSet().reverseRange(key, start, end);
        if (feedIds == null || feedIds.isEmpty()) {
            return Lists.newArrayList();
        }
        // 根据多主键查询 Feed
        List<Feeds> feeds = feedsMapper.findFeedsByIds(feedIds);
        // 初始化关注好友 ID 集合
        List<Integer> followinguserIds = new ArrayList<>();
        // 添加用户 ID 至集合，顺带将 Feeds 转为 Vo 对象
        List<FeedsVO> feedsVOS = feeds.stream().map(feed -> {
            FeedsVO feedsVO = new FeedsVO();
            BeanUtil.copyProperties(feed, feedsVO);
            // 添加用户 ID
            followinguserIds.add(feed.getFkUserId());
            return feedsVO;
        }).collect(Collectors.toList());
        // 远程调用获取 Feed 中用户信息
        ResultInfo resultInfo = restTemplate.getForObject(usersServerName + "findByIds?access_token=${accessToken}&ids={ids}",
                ResultInfo.class, accessToken, followinguserIds);
        if (resultInfo.getCode() != ApiConstant.SUCCESS_CODE) {
            throw new ParameterException(resultInfo.getCode(), resultInfo.getMessage());
        }
        List<LinkedHashMap> userInfoMaps = (ArrayList) resultInfo.getData();
        // 构建一个 key 为用户 ID，value 为 ShortuserInfo 的 Map
        Map<Integer, ShortUserInfo> userInfos = userInfoMaps.stream()
                .collect(Collectors.toMap(
                        // key
                        diner -> (Integer) diner.get("id"),
                        // value
                        diner -> BeanUtil.fillBeanWithMap(diner, new ShortUserInfo(), true)
                ));
        // 循环 VO 集合，根据用户 ID 从 Map 中获取用户信息并设置至 VO 对象
        feedsVOS.forEach(feedsVO -> {
            feedsVO.setUserInfo(userInfos.get(feedsVO.getFkUserId()));
        });
        return feedsVOS;
    }

FeedsMapper

    /**
     * 根据多主键查询 Feed
     * @param feedIds
     * @return
     */
    @Select("<script> " +
            " select id, content, fk_user_id, praise_amount, " +
            " comment_amount, fk_restaurant_id, create_date, update_date, is_valid " +
            " from t_feeds where is_valid = 1 and id in " +
            " <foreach item=\"id\" collection=\"feedIds\" open=\"(\" separator=\",\" close=\")\">" +
            "   #{id}" +
            " </foreach> order by id desc" +
            " </script>")
    List<Feeds> findFeedsByIds(@Param("feedIds") Set<Integer> feedIds);