Redis 實戰 —— 03. Redis 簡單實踐 – Web應用

• 2021 年 1 月 22 日
• 筆記
• Redis, Redis 實戰

需求

功能： P23

• 登錄 cookie
• 購物車 cookie
• 快取生成的網頁
• 快取資料庫行
• 分析網頁訪問記錄

高層次角度下的 Web 應用 P23

從高層次的角度來看， Web 應用就是通過 HTTP 協議對網頁瀏覽器發送的請求進行響應的伺服器或者服務（service）。 Web 請求一般是無狀態的（stateless），即伺服器本身不會記錄與過往請求有關的任何資訊，使得失效的伺服器可以很容易地被替換掉。

Web 伺服器對請求進行響應的典型步驟：

1. 伺服器對客戶端發來對請求（request）進行解析
2. 請求被轉發給一個預定義的處理器（handler）
3. 處理器可能會從資料庫中取數據
4. 處理器根據數據對模板（templete）進行渲染（render）
5. 處理器向客戶端返回渲染後的內容作為對請求的響應（response）

基礎數值量 P24

本次實踐所有內容均圍繞著發現並解決一個虛構的大型網上商店來展開的，一些基礎數據量如下：

• 每天有 500 萬名不同的用戶
• 每天有 1 億次點擊
• 每天從網站購買超過 10 萬件商品

實現

登錄和 cookie 快取 P24

有兩種常見的方法可以將登錄資訊存儲在 cookie 裡面：

• 簽名（signed） cookie：通常會存儲用戶名，可能還會有其他網站覺得游泳的資訊，例如：最後一次成功登錄時間、用戶 id 等。還會有簽名，用伺服器驗證 cookie 中的資訊是否被修改。
• 令牌（token） cookie：存儲遺傳隨機位元組作為令牌，伺服器根據令牌在資料庫中查找令牌的擁有著。隨著時間的推移，舊令牌會被新令牌取代。

簽名 cookie 和令牌 cookie 的優點與缺點 P24

本次實踐採用令牌 cookie 來引用存儲的用戶登錄資訊的條目。除登錄資訊外，還需要將用戶訪問時長和已瀏覽商品的數量等資訊存儲到資料庫裡面，便於未來通過分析這些資訊來學習如何更好地向用戶推銷商品。

使用一個哈希表來存儲登錄 cookie 令牌與已登錄用戶之間的映射，根據給定的令牌查找對應的用戶 id。 P24

// redis key
type RedisKey string
const (
	// 登錄用戶 哈希表（field：token；value：userId）
	LOGIN_USER RedisKey = "loginUser"

	// 用戶最近操作時間 有序集合
	USER_LATEST_ACTION RedisKey = "userLatestAction"

	// 用戶最近瀏覽商品時間 有序集合 前綴（存儲 itemId 及瀏覽的時間戳）
	VIEWED_ITEM_PREFIX RedisKey = "viewedItem:"

	// 用戶購物車 哈希表 前綴（存儲 itemId 及其加車數量）
	CART_PREFIX RedisKey = "cart:"

	// 請求返回值快取 字元串 前綴（存儲 請求對應返回值的 序列化串）
	REQUEST_PREFIX RedisKey = "request:"

	// 快取數據間隔（單位：ms） 字元串
	ITEM_INTERVAL RedisKey = "itemInterval"

	// 數據快取時間（精確到毫秒） 有序集合
	ITEM_CACHED_TIME RedisKey = "itemCachedTime"

	// 數據（商品）的json 字元串 前綴（存儲 itemId 的相關資訊）
	ITEM_PREFIX RedisKey = "item:"

	// 商品瀏覽次數 有序集合（存儲 itemId 及瀏覽次數）
	ITEM_VIEWED_NUM RedisKey = "itemViewedNum"
)

// 根據 token 獲取 userId（err 不為 nil 時，用戶已登錄且 userId 有效）
func GetUserId(conn redis.Conn, token string) (userId int, err error) {
	return redis.Int(conn.Do("HGET", LOGIN_USER, token))
}

此時我們已經能通過 token 獲取用戶 id 了，還需要相應的設置方法，即用戶每次操作時都會進行相關資訊設置，並更新 token 的最近操作時間戳。如果用戶正在瀏覽一個商品，則還需要將該商品添加到瀏覽商品歷史有序集合中，且限制一個用戶最多記錄最新的 25 個商品瀏覽記錄。 P25

// 一個用戶瀏覽的商品最多記錄最新的 25 個
const MAX_VIEWED_ITEM_COUNT = 25

// 更新令牌相關資訊（用戶有操作是就會更新，如果當前操作是瀏覽商詳頁，則傳入 itemId，否則 itemId <= 0）
func UpdateToken(conn redis.Conn, token string, userId int, itemId int) {
	currentTime := time.Now().Unix() + int64(itemId)

	// 更新令牌及相應 userId 對應關係
	_, _ = conn.Do("HSET", LOGIN_USER, token, userId)
	// 最近操作時間 記錄 token 的時間戳
	// （不能記錄 userId 的時間戳，userId 不會變，所以即使 token 更新了， userId 對應的時間戳還是會更新，沒法判斷當前 token 是否過期）
	_, _ = conn.Do("ZADD", USER_LATEST_ACTION, currentTime, token)

	// 當前瀏覽的時商品詳情頁時，會傳入 itemId，否則 itemId <= 0
	if itemId > 0 {
		// 決定用 userId 當做後綴：token 可能會改變，而 userId 是唯一確定的
		viewedItemKey := VIEWED_ITEM_PREFIX + RedisKey(strconv.Itoa(userId))
		// 添加（更新）最近瀏覽商品資訊
		_, _ = conn.Do("ZADD", viewedItemKey, currentTime, itemId)
		// 移除時間戳升序狀態下 [0, 倒數第 MAX_VIEWED_ITEM_COUNT + 1 個] 內的所有元素，留下最近的 MAX_VIEWED_ITEM_COUNT 個
		_, _ = conn.Do("ZREMRANGEBYRANK", viewedItemKey, 0, -(MAX_VIEWED_ITEM_COUNT + 1))
	}
}

存儲的數據會越來越多，且登錄用戶不會一直操作，所以可以設定最多支援的登錄用戶數，並定期刪除超過的那些最久為操作的用戶登錄資訊。 P26

書上同時也將用戶的歷史訪問記錄刪除了，我這裡沒做刪除，把存儲歷史訪問記錄的有序集合當作資料庫用，與用戶登錄狀態無關，即使用戶的登錄資訊被刪除了，仍舊可以分析相應的數據，比較符合實際使用情況。

清理方法內部死循環可能不太優雅，但是使用 go 關鍵字以協程運行，可以在一定程度上達到定時任務的效果，且和大部分情況下定時任務一樣，會隨著主程式的退出而退出。

// 登錄用戶最多記錄 1000w 條最新資訊（其實這時候早已經是大 key 了，不過當前場景不需要太過於考慮這個）
const MAX_LOGIN_USER_COUNT = 10000000

// 清理 session 實際間隔 10s 運行一次
const CLEAN_SESSIONS_INTERVAL = 10 * time.Second

// 每次清理的個數
const CLEAN_COUNT = 1000


// 求 兩個 int 的最小值
func min(a, b int) int {
	if a < b {
		return a
	}
	return b
}

// 合併 RedisKey 與 []string，返回一個 []interface{}
func merge(redisKey RedisKey, strs ...string) []interface{} {
	result := make([]interface{}, 1 + len(strs))
	result[0] = redisKey
	for i, item := range strs {
		result[i + 1] = item
	}

	return result
}

// 清理 session （由於大部分用戶不會一直操作，所以需要定期清理 最舊的登錄資訊）
// 內部死循環，可用 go 調用，當作定時任務
func CleanSessions(conn redis.Conn) {
	for ; ;  {
		loginUserCount, _ := redis.Int(conn.Do("ZCARD", LOGIN_USER))
		// 超過最大記錄數，則清理
		if loginUserCount > MAX_VIEWED_ITEM_COUNT {
			// 獲取最舊的記錄的 token ，最多不超過 CLEAN_COUNT 個（多執行緒/分散式情況下會有並發問題，重點不在這裡，暫不考慮）
			cleanCount := min(loginUserCount - MAX_LOGIN_USER_COUNT, CLEAN_COUNT)
			tokens, _ := redis.Strings(conn.Do("ZRANGE", USER_LATEST_ACTION, 0, cleanCount - 1))

			// 不支援 []string 直接轉 []interface{} （對字元串數組使用 ... 無法對應參數 ...interface{}）
			// 只有數組的元素類型 Type 相同才能使用 ... 傳參到相應的 ...Type
			_, _ = conn.Do("HDEL", merge(LOGIN_USER, tokens...)...)
			_, _ = conn.Do("ZREM", merge(USER_LATEST_ACTION, tokens...)...)

			// 不刪除用戶的歷史訪問記錄，當作資料庫用
		}

		// 每次執行完，等待 CLEAN_SESSIONS_INTERVAL 長時間
		time.Sleep(CLEAN_SESSIONS_INTERVAL)
	}
}

購物車 P28

每個用戶的購物車是一個哈希表，存儲了 itemId 與 商品加車數量之間的關係。此處購物車僅提供最基礎的數量設置，加減數量的邏輯由調用者進行相關處理。

// 更新購物車商品數量（不考慮並發問題）
func UpdateCartItem(conn redis.Conn, userId int, itemId int, count int) {
	cartKey := CART_PREFIX + RedisKey(strconv.Itoa(userId))
	if count <= 0 {
		// 刪除該商品
		_, _ = conn.Do("HREM", cartKey, itemId)
	} else {
		// 更新商品數量
		_, _ = conn.Do("HSET", cartKey, itemId, count)
	}
}

購物車也和歷史訪問記錄一樣，當作資料庫使用，與用戶登錄態無關，不隨登錄態退出而刪除。因此定期清理登錄數據的程式碼不需要修改，也不用添加新函數。

網頁快取 P29

假設網站的 95% 頁面每天最多只會改變一次，那麼沒有必要每次都進行全部操作，可以直接在實際處理請求前將快取的數據返回，既能降低伺服器壓力，又能提升用戶體驗。

Java 中可以使用註解方式對特定的服務進行攔截快取，實現快取的可插拔，避免修改核心程式碼。

Go 的話，不知道如何去實現可插拔的方式，感覺可以利用 Java 中攔截器的方式，在請求分發到具體的方法前進行判斷及快取。我這裡只進行簡單的業務邏輯處理展示大致流程，不關心如何實現可插拔，讓用戶無感知。

// 判斷當前請求是否可以快取（隨實際業務場景處理，此處不關心，默認都可以快取）
func canCache(conn redis.Conn, request http.Request) bool {
	return true
}

// 對請求進行哈希，得到一個標識串（隨實際業務場景處理，此處不關心，默認為 url）
func hashRequest(request http.Request) string {
	return request.URL.Path
}

// 對返回值進行序列化，得到字元串（隨實際業務場景處理，此處不關心，默認為 序列化狀態碼）
func serializeResponse(response http.Response) string {
	return strconv.Itoa(response.StatusCode)
}

// 對快取的結果進行反序列化，得到返回值（隨實際業務場景處理，此處不關心，默認 狀態碼為 200）
func deserializeResponse(str string) http.Response {
	return http.Response{StatusCode: 200}
}

// 返回值快取時長為 5 分鐘
const CACHE_EXPIRE = 5 * time.Minute

// 快取請求返回值
func CacheRequest(conn redis.Conn, request http.Request, handle func(http.Request) http.Response) http.Response {
	// 如果當前請求不能快取，則直接調用方法返回
	if !canCache(conn, request) {
		return handle(request)
	}

	// 從快取中獲取快取的返回值
	cacheKey := REQUEST_PREFIX + RedisKey(hashRequest(request))
	responseStr, _ := redis.String(conn.Do("GET", cacheKey))
	// 序列化時，所有有效的快取都有狀態嗎，所以必定不為 ""
	if responseStr != "" {
		return deserializeResponse(responseStr)
	}

	// 快取中沒有，則重新執行一遍
	response := handle(request)
	// 先進行快取，再返回結果
	responseStr = serializeResponse(response)
	_, _ = conn.Do("SET", cacheKey, responseStr, "EX", CACHE_EXPIRE)
	return response
}

數據快取 P30

我們不能快取的經常變動的頁面，但是可以快取這些頁面上的數據，例如：促銷商品、熱賣商品等。 P30

現在網站需要進行促銷，促銷商品數量固定，賣完即止。為了保證用戶近實時看到促銷商品及數量，又要保證不給資料庫帶來巨大壓力，所以需要對促銷商品的數據進行快取。

可以用定時任務定期將需要快取的數據更新到 Redis 中（其實對於促銷等商品直接在快取中進行相關庫存扣減，既能保證實時數量，也能降低資料庫壓力，不過會存在熱 key 問題）。由於不同的商品可能對實時性要求不一樣，所以需要記錄每個商品的更新周期和更新時間，分別存成哈希表和有序集合。 P31

為了讓定時任務定期快取數據，需要提供一個函數，以設置更新周期和更新時間。

// 毫秒轉納秒所需的倍數
const MILLI_2_NANO int64 = 1e6

// 更新數據快取的更新周期（單位：ms）和更新時間（精確到毫秒）
func UpdateItemCachedInfo(conn redis.Conn, itemId int, interval int) {
	_, _ = conn.Do("HSET", ITEM_INTERVAL, itemId, interval)
	_, _ = conn.Do("ZADD", ITEM_CACHED_TIME, time.Now().UnixNano() / MILLI_2_NANO, itemId)
}

定時任務定時獲取第一個需要更新的商品，若更新時間還未到，則等待下次執行。當更新周期不存在或者小於等於 0 時，表示不需要快取，刪除相關快取；當更新周期大於等於 0 時，獲取商品數據，並更新相關快取。 P32

// 定時任務每 50ms 執行一次
const CACHE_ITEM_INTERVAL = 50

// 獲取商品數據的 json串（隨實際業務場景處理，此處不關心，默認 只含有itemId）
func getItemJson(itemId int) string {
	return fmt.Sprintf("{\"id\":%d}", itemId)
}

// 快取數據
// 內部死循環，可用 go 調用，當作定時任務
func CacheItem(conn redis.Conn) {
	for ; ;  {
		// 獲取第一個需要更新的商品（不考慮沒有商品的情況）
		result, _ := redis.Ints(conn.Do("ZRANGE", ITEM_CACHED_TIME, 0, 0, "WITHSCORES"))
		itemId, itemCachedTime := result[0], result[1]
		// 如果當前時間還沒到，則等下次執行
		if int64(itemCachedTime) * MILLI_2_NANO > time.Now().UnixNano() {
			time.Sleep(CACHE_ITEM_INTERVAL * time.Millisecond)
			continue
		}

		// 獲取更新周期
		interval, _ := redis.Int(conn.Do("HGET", ITEM_INTERVAL, itemId))
		itemKey := ITEM_PREFIX + RedisKey(strconv.Itoa(itemId))
		// 如果更新周期 小於等於0，則移除相關快取資訊
		if interval <= 0 {
			_, _ = conn.Do("HREM", ITEM_INTERVAL, itemId)
			_, _ = conn.Do("ZREM", ITEM_CACHED_TIME, itemId)
			_, _ = conn.Do("DELETE", itemKey)
			continue
		}

		// 如果更新周期 大於0，則還獲取數據需要進行快取
		itemJson := getItemJson(itemId)
		_, _ = conn.Do("SET", itemKey, itemJson)
		_, _ = conn.Do("ZADD", ITEM_CACHED_TIME, time.Now().UnixNano() / MILLI_2_NANO + int64 (interval), itemId)
	}
}

網頁分析 P33

現在網站只想將 100 000 件商品中用戶最經常瀏覽的 10 000 件商品快取，所以需要記錄每件商品的總瀏覽次數，並能夠獲取到瀏覽次數最多的 10 000 件商品，所以需要用有序集合進行存儲記錄。同時需要在 UpdateToken 加入增加次數的語句，更改後 UpdateToken 如下： P33

// 更新令牌相關資訊（用戶有操作是就會更新，如果當前操作是瀏覽商詳頁，則傳入 itemId，否則 itemId <= 0）
func UpdateToken(conn redis.Conn, token string, userId int, itemId int) {
	currentTime := time.Now().Unix() + int64(itemId)

	// 更新令牌及相應 userId 對應關係
	_, _ = conn.Do("HSET", LOGIN_USER, token, userId)
	// 最近操作時間 記錄 token 的時間戳
	// （不能記錄 userId 的時間戳，userId 不會變，所以即使 token 更新了， userId 對應的時間戳還是會更新，沒法判斷當前 token 是否過期）
	_, _ = conn.Do("ZADD", USER_LATEST_ACTION, currentTime, token)

	// 當前瀏覽的時商品詳情頁時，會傳入 itemId，否則 itemId <= 0
	if itemId > 0 {
		// 決定用 userId 當做後綴：token 可能會改變，而 userId 是唯一確定的
		viewedItemKey := VIEWED_ITEM_PREFIX + RedisKey(strconv.Itoa(userId))
		// 添加（更新）最近瀏覽商品資訊
		_, _ = conn.Do("ZADD", viewedItemKey, currentTime, itemId)
		// 移除時間戳升序狀態下 [0, 倒數第 MAX_VIEWED_ITEM_COUNT + 1 個] 內的所有元素，留下最近的 MAX_VIEWED_ITEM_COUNT 個
		_, _ = conn.Do("ZREMRANGEBYRANK", viewedItemKey, 0, -(MAX_VIEWED_ITEM_COUNT + 1))
		// 每次瀏覽商詳頁是，都要增加當前商品的瀏覽量
		_, _ = conn.Do("ZINCRBY", ITEM_VIEWED_NUM, 1, itemId) //【改動點】
	}
}

此時，可以定時刪除瀏覽量不在前 10 000 的商品快取，同時為了保證新的熱點商品能夠不被已有的熱點商品影響，所以在刪除商品快取後，要對沒刪除的商品次數進行折半處理。可以使用 ZINTERSTORE ，並配置WEIGHTS 選項可以對所有商品分值乘以相同的權重（當只有一個有序集合時，ZUNIONSTORE 效果一樣） P33

// 刪除非熱點商品快取，重設熱點商品瀏覽量 執行周期，每 5分鐘一次
const RESCALE_ITEM_VIEWED_NUM_INTERVAL = 300
// 熱點商品瀏覽量權重
const ITEM_VIEWED_NUM_WEIGHT = 0.5
// 最大快取商品數量
const MAX_ITEM_CACHED_NUM = 10000

// 刪除非熱點商品快取，重設降權熱點商品瀏覽量
// 內部死循環，可用 go 調用，當作定時任務
func RescaleItemViewedNum(conn redis.Conn) {
	for ; ;  {
		// 刪除瀏覽量最小的 [0, 倒數 20 001] 商品，留下瀏覽量最大的 20 000 件商品
		// 此處留下的瀏覽量記錄是最大快取商品數量的 2 倍，可以讓新熱點數據不被刪掉
		_, _ = conn.Do("ZREMRANGEBYRANK", ITEM_VIEWED_NUM, "0", -((MAX_ITEM_CACHED_NUM << 1) + 1))
		// 瀏覽量折半，保證新熱點數據不被影響太多
		_, _ = conn.Do("ZINTERSTORE", ITEM_VIEWED_NUM, "1", ITEM_VIEWED_NUM, "WEIGHTS", ITEM_VIEWED_NUM_WEIGHT)
		// 等待 5min 後，再執行下一次操作
		time.Sleep(RESCALE_ITEM_VIEWED_NUM_INTERVAL * time.Second)
	}
}

此處書中疑點

P33 倒數第二段：

新添加的程式碼記錄了所有商品的瀏覽次數，並根據瀏覽次數對商品進行了排序，被瀏覽得最多的商品將被放到有序集合的索引 0 位置上，並且具有整個有序集合最少的分值。

相應 Python 程式碼片段：

conn.zincrby('viewed:', item, -1)

而進行刪除排名在 20 000 名之後的商品操作時如下：

conn.zremrangebyrank('viewed:', 0, -20001)

Redis 命令 ZREMRANGEBYRANK 移除有序集合中排名在區間內的所有元素（按升序排序），按此理解，上述程式碼的結果會留下瀏覽量最小的 20 000 個商品，與實際需求不符。

後來看到其他命令使用方式不同，想到可能是 Python 的命令有點不一樣，網上一搜發現兩種結果都有，還是要自己實踐證明。

經過在 Python 中進行實踐，上述刪除操作的確是按升序排序，刪除分值（瀏覽量）最低的部分，留下分值（瀏覽量）最高的部分。（當然，實踐比較簡單，有可能有其他配置會影響結果，就不再探究了）

接下來書上通過以下程式碼獲取瀏覽量排名，並進行排名判斷： P34

rank = conn.zrank('viewed:', item_id)
return rank is not None and rank < 10000

可以看到作者在此處的是符合負數瀏覽量的設置方式的，但是正數瀏覽量可以通過 ZREVRANK 獲取降序排名，可以猜測作者在刪除排名在 20 000 後的商品時沒有考慮清楚。

按照負數瀏覽量時，可以使用以下程式碼正確刪除排名在 20 000 後的商品：

conn.zremrangebyrank('viewed:', 20000, -1)

至此，我們就可以實現前面提到過的 canCache 函數，只有能被快取的商品頁面，並且排名在 10 000 內的商品頁面才能被快取。 P34

// 從請求從獲取 itemId，不存在則返回 錯誤（隨實際業務場景處理，此處不關心，默認都是 1）
func getItemId(request http.Request) (int, error) {
	return 1, nil
}

// 判斷當前請求是否是動態的（隨實際業務場景處理，此處不關心，默認都不是）
func isDynamic(request http.Request) bool {
	return false
}

// 判斷當前請求是否可以快取（隨實際業務場景處理，此處不關心，默認都可以快取）
func canCache(conn redis.Conn, request http.Request) bool {
	itemId, err := getItemId(request)
	// 如果沒有 itemId（即不是商品頁面） 或者 結果是動態的，則不能快取
	if err == nil || isDynamic(request) {
		return false
	}

	// 獲取請求的商品頁面的瀏覽量排名
	itemRank, err := redis.Int(conn.Do("ZREVRANK", ITEM_VIEWED_NUM, itemId))
	// 如果 不存在 或者 排名大於 10000 名，則不快取
	if err != nil || itemRank >= MAX_ITEM_CACHED_NUM {
		return false
	}
	return true
}

小結

• 要隨著需求的改變重構已有的程式碼

所思所想

• 上一次實踐很多時間都花在考慮處理異常流程的業務邏輯上了，無用程式碼篇幅較大，且與專心學習 Redis 及熟悉 go 的初衷不符合，所以本次開始就基本專註於如何去實現功能，不太考慮一些參數校驗和異常流程了。
• 實踐是檢驗真理的唯一標準。記得在讀《Head First 設計模式》時也遇到了書本中存疑的地方，不能盡信書，還是要敢於質疑，用實踐去驗證各種可能性。

本文首發於公眾號：滿賦諸機（點擊查看原文） 開源在 GitHub ：reading-notes/redis-in-action