巧用 Redis，實現微博 Feed 流功能！
作者：技術老男孩
最近接到一個需求，用一句話來說就是：展示關注人發布的動態，這個涉及到 feed 流系統的設計。本文主要介紹一個一般企業可用的 Feed 流解決方案。
一、背景
最近接到一個需求，用一句話來說就是：展示關注人發布的動態，這個涉及到 feed 流系統的設計。本文主要介紹一個一般企業可用的 Feed 流解決方案。
二、相關概念
下面先介紹一下關于 Feed 流的簡單概念。
1.什么是 feed 流
- Feed：Feed 流中的每一條狀態或者消息都是 Feed，比如微博中的一條微博就是一個 Feed。
- Feed流：持續更新并呈現給用戶內容的信息流。每個人的朋友圈，微博關注頁等等都是一個 Feed 流。
2.feed 流分類
Feed 流常見的分類有兩種：
- Timeline：按發布的時間順序排序，產品如果選擇 Timeline 類型，那么就是認為 Feed 流中的 Feed 不多，但是每個 Feed 都很重要，都需要用戶看到。類似于微信朋友圈，微博等。
- Rank：按某個非時間的因子排序，一般是按照用戶的喜好度排序，一般用于新聞推薦類、商品推薦等。
三、設計
設計一個 Feed 流系統，兩個關鍵步驟，一個是 Feed 流的 初始化，一個是 推送。關于 Feed 流的存儲其實也是一個核心的點，但是筆主持久化使用的還是 MySQL，后續可以考慮優化。
1.Feed 流初始化
Feed 流【關注頁 Feed 流】的初始化指的是，當用戶的 Feed 流還不存在的時候，為該用戶創建一個屬于他自己的關注頁 Feed 流，具體怎么做呢？其實很簡單，遍歷一遍關注列表，取出所有關注用戶的 feed，將 feedId 存放到 redis 的 sortSet 中即可。這里面有幾個關鍵點：
- 初始化數據：初始化的數據需要從數據庫中加載出來。
- key 值：sortSet 的 key 值需要使用當前用戶的 id 做標識。
- score 值：如果是 Timeline 類型，直接取 feed 創建的時間戳即可。如果是 rank 類型，則把你的業務對應的權重值設進去。
2.推送
經過上面的初始化，已經把 feed 流放在了 redis 緩存中了。接下來就是需要更新 feed 流了，在下面四種情況需要進行更新：
- 關注的用戶發布新的 feed：
- 關注的用戶刪除 feed。
- 用戶新增關注。
- 用戶取消關注。
3.發布/刪除 Feed 流程
上面四步具體怎么操作，會在下面的實現步驟中詳細描述，在這里先我們重點討論一下第一、二種情況。因為在處理 大V 【千萬級別粉絲】的時候，我們是需要對 大V 的所有粉絲的 feed 流進行處理的，這時候涉及到的量就會非常巨大，需要多加斟酌。關于推送，一般有兩種 推/拉。
- 推：A用戶發布新的動態時，要往 A用戶所有的粉絲 feed 流中推。
- 拉：A用戶發布新的動態時，先不進行推送，而是等 粉絲進來的時候，才主動到 A用戶的個人頁TimeLine 拉取最新的 feed，然后進行一個 merge。如果關注了多個大V，可以并發的向多個大V 個人頁TimeLine 中拉取。
4.推拉結合模式
當用戶發布一條新的 Feed 時，處理流程如下：
- 先從關注列表中讀取到自己的粉絲列表，以及判斷自己是否是大V。
- 將自己的Feed消息寫入個人頁Timeline。如果是大V，寫入流程到此就結束了。
- 如果是普通用戶，還需要將自己的Feed消息寫給自己的粉絲，如果有100個粉絲，那么就要寫給100個用戶。
當刷新自己的Feed流的時候，處理流程如下：
- 先去讀取自己關注的大V列表
- 去讀取自己的 Feed 流。
- 如果有關注的大V，則再次并發讀取每一個大V的個人頁Timeline，如果關注了10個大V，那么則需要10次訪問。
- 合并2和3步的結果，然后按時間排序，返回給用戶。
至此，使用推拉結合方式的發布，讀取Feed流的流程都結束了。
5.推模式
如果只是用推模式了，則會變的比較簡單：
- 「發布Feed：」不用區分是否大V，所有用戶的流程都一樣，都是三步。
- 「讀取Feed流：」不需要第一步，也不需要第三步，只需要第二步即可，將之前的2 + N(N是關注的大V個數) 次網絡開銷減少為 1 次網絡開銷。讀取延時大幅降級。
6.兩種模式總結
推拉結合存在一個弊端，就是刷新自己的Feed流時，大V的個人頁Timeline 的讀壓力會很大。
如何解決：
- 不使用大V/普通用戶的優化方式，使用對活躍粉絲采用推模式，非活躍粉絲采用拉模式。
- 完全使用推模式就可以徹底解決這個問題，但是會帶來存儲量增大，大V Feed 發送總時間增大，從發給第一個粉絲到發給最后一個粉絲可能要幾分鐘時間。
四、實現
筆主主要采用純推模式實現了一個普通企業基本可用的 Feed 流系統，下面介紹一下具體的實現代碼，主要包括3大個部分：
- 初始化 Feed 流。
- 關注的用戶發布/刪除 feed，該用戶的粉絲更新自己的Feed流。
- 用戶新增/取消關注，更新自己的Feed流。
1.初始化 Feed 流
當用戶第一進來刷新Feed 流，且 Feed 流還不存在時，我們需要進行初始化，初始化的具體代碼如下：核心思想就是從數據庫中load出 feed 信息，塞到 zSet 中，然后分頁返回。
/**
* 獲取關注的人的信息流
*/
public List<FeedDto> listFocusFeed(Long userId, Integer page, Integer size) {
String focusFeedKey = "focusFeedKey" + userId;
// 如果 zset 為空，先初始化
if (!zSetRedisTemplate.exists(focusFeedKey)) {
initFocusIdeaSet(userId);
}
// 如果 zset 存在，但是存在 0 值
Double zscore = zSetRedisTemplate.zscore(focusFeedKey, "0");
if (zscore != null && zscore > 0) {
return null;
}
//分頁
int offset = (page - 1) * size;
long score = System.currentTimeMillis();
// 按 score 值從大到小從 zSet 中取出 FeedId 集合
List<String> list = zSetRedisTemplate.zrevrangeByScore(focusFeedKey, score, 0, offset, size);
List<FeedDto> result = new ArrayList<>();
if (QlchatUtil.isNotEmpty(list)) {
for (String s : list) {
// 根據 feedId 從緩存中 load 出 feed
FeedDto feedDto = this.loadCache(Long.valueOf(s));
if (feedDto != null) {
result.add(feedDto);
}
}
}
return result;
}
/**
* 初始化關注的人的信息流 zSet
*/
private void initFocusFeedSet( Long userId) {
String focusFeedKey = "focusFeedKey" + userId;
zSetRedisTemplate.del(focusIdeaKey);
// 從數據庫中加載當前用戶關注的人發布過的 Feed
List<Feed> list = this.feedMapper.listFocusFeed(userId);
if (QlchatUtil.isEmpty(list)) {
//保存0，避免空數據頻繁查庫
zSetRedisTemplate.zadd(focusFeedKey, 1, "0");
zSetRedisTemplate.expire(focusFeedKey, RedisKeyConstants.ONE_MINUTE * 5);
return;
}
// 遍歷 FeedList，把 FeedId 存到 zSet 中
for (Feed feed : list) {
zSetRedisTemplate.zadd(focusFeedKey, feed.getCreateTime().getTime(), feed.getId().toString());
}
zSetRedisTemplate.expire(focusFeedKey, 60 * 60 * 60);
}2.關注的用戶發布/刪除新的 feed
每當用戶發布/刪除新的 feed，我們需要更新該用戶所有的粉絲的 Feed流，該步驟一般比較耗時，所以建議異步處理，為了避免一次性load出太多的粉絲數據，這里采用循環分頁查詢。為了避免粉絲的 Feed流過大，我們會限制 Feed 流的長度為1000，當Feed流長度超過1000時，會移除最舊的 Feed。
/**
* 新增/刪除 feed時，處理粉絲 feed 流
*
* @param userId 新增/刪除 feed的用戶id
* @param feedId 新增/刪除 的feedId
* @param type feed_add = 新增feed feed_sub = 刪除feed
*/
public void handleFeed(Long userId, Long feedId, String type) {
Integer currentPage = 1;
Integer size = 1000;
List<FansDto> fansDtos;
while (true) {
Page page = new Page();
page.setSize(size);
page.setPage(currentPage);
fansDtos = this.fansService.listFans(userId, page);
for (FansDto fansDto : fansDtos) {
String focusFeedKey = "focusFeedKey" + userId;
// 如果粉絲 zSet 不存在，退出
if (!this.zSetRedisTemplate.exists(focusFeedKey)) {
continue;
}
// 新增Feed
if ("feed_add".equals(type)) {
this.removeOldestZset(focusFeedKey);
zSetRedisTemplate.zadd(focusFeedKey, System.currentTimeMillis(), feedId);
}
// 刪除Feed
else if ("feed_sub".equals(type)) {
zSetRedisTemplate.zrem(focusFeedKey, feedId);
}
}
if (fansDtos.size() < size) {
break;
}
currentPage++;
}
}
/**
* 刪除 zSet 中最舊的數據
*/
private void removeOldestZset(String focusFeedKey){
// 如果當前 zSet 大于1000，刪除最舊的數據
if (this.zSetRedisTemplate.zcard(focusFeedKey) >= 1000) {
// 獲取當前 zSet 中 score 值最小的
List<String> zrevrange = this.zSetRedisTemplate.zrevrange(focusFeedKey, -1, -1, String.class);
if (QlchatUtil.isNotEmpty(zrevrange)) {
this.zSetRedisTemplate.zrem(focusFeedKey, zrevrange.get(0));
}
}
}3.用戶新增關注/取消關注
這里比較簡單，新增/取消關注，把新關注的 Feed 往自己的 Feed流中增加/刪除 即可，但是同樣需要異步處理。
/**
* 關注/取關 時，處理followerId的zSet
*
* @param followId 被關注的人
* @param followerId 當前用戶
* @param type focus = 關注 unfocus = 取關
*/
public void handleFocus( Long followId, Long followerId, String type) {
String focusFeedKey = "focusFeedKey" + userId;
// 如果粉絲 zSet 不存在，退出
if (!this.zSetRedisTemplate.exists(focusFeedKey)) {
return;
}
List<FeedDto> FeedDtos = this.listFeedByFollowId(source, followId);
for (FeedDto feedDto : FeedDtos) {
// 關注
if ("focus".equals(type)) {
this.removeOldestZset(focusFeedKey);
this.zSetRedisTemplate.zadd(focusFeedKey, feedDto.getCreateTime().getTime(), feedDto.getId());
}
// 取關
else if ("unfocus".equals(type)) {
this.zSetRedisTemplate.zrem(focusFeedKey, feedDto.getId());
}
}
}上面展示的是核心代碼，僅僅是為大家提供一個實現思路，并不是直接可運行的代碼，畢竟真正實現還會涉及到很多其他的無關要緊的類。
責任編輯：趙寧寧
來源：
技術老男孩
