使用 Redis 進行緩存

Redis 通常被認為只是一個數據存儲，但它的速度和內存中特性使其成為緩存的絕佳選擇。緩存是一種技術，通過將經常訪問的數據存儲在快速的臨時存儲位置來提高應用程序性能。通過使用 Redis 作為緩存，您可以顯著減少主數據庫的負載并縮短用戶的響應時間。本課將探討如何有效地使用 Redis 進行緩存，涵蓋關鍵概念、策略和最佳實踐。

了解緩存概念

緩存是軟件開發中的一種基本優化技術。它涉及將數據副本存儲在緩存中，緩存是一個高速數據存儲層，以便將來可以更快地處理對該數據的請求。發出請求時，首先檢查緩存。如果在緩存中找到數據（“緩存命中”），則直接從緩存中提供數據。如果未找到數據（“緩存未命中”），則會從原始數據源（例如數據庫）中檢索數據，將其存儲在緩存中，然后提供給用戶。

緩存的好處

• 改進的性能： 緩存通過從更快的存儲層提供數據來減少延遲并縮短響應時間。
• 減少數據庫負載： 通過從緩存中提供經常訪問的數據，您可以減少主數據庫的負載，使其能夠處理更復雜的查詢和作。
• 提高可擴展性： 緩存通過減少后端系統的負載，使您的應用程序能夠處理更多的并發用戶和請求。
• 節省成本： 通過減少數據庫負載和提高資源利用率，緩存可以節省基礎設施和運營費用方面的成本。

緩存策略

Redis 可以使用多種緩存策略，每種策略都有自己的優點和缺點：

• Cache-Aside （延遲加載）： 應用程序首先檢查緩存中的數據。如果找到數據，則直接返回數據。如果沒有，應用程序將從數據庫中檢索數據，將其存儲在緩存中，然后返回它。此策略易于實施，并確保緩存僅包含已請求的數據。
• 直寫： 應用程序同時將數據寫入緩存和數據庫。這可確保緩存始終是最新的，但會增加寫入延遲。
• 回寫 （Write-Behind）： 應用程序將數據寫入緩存，緩存將數據異步寫入數據庫。此策略提供最低的寫入延遲，但如果緩存在將數據寫入數據庫之前失敗，則可能導致數據丟失。
• 通讀： 應用程序與緩存交互，而緩存又與數據庫交互。請求數據時，緩存會檢查它是否包含數據。否則，它將從數據庫中檢索數據，將其存儲在緩存中，然后將其返回給應用程序。

對于大多數使用案例，Cache-Aside 策略是 Redis 最實用且最常用的策略，因為它簡單高效。

使用 Redis 作為緩存

Redis 非常適合緩存，因為它的速度、內存數據存儲和對各種數據結構的支持。以下是將 Redis 用作緩存的方法：

設置和檢索數據

您可以使用 SET 和 GET 命令在 Redis 中存儲和檢索數據。例如：

SET user:123 '{"id": 123, "name": "John Doe", "email": "john.doe@example.com"}'
GET user:123

在此示例中，我們將一個 JSON 字符串存儲在 Redis 中，該字符串表示用戶對象，其鍵為 user：123。使用 GET user：123 檢索數據時，Redis 返回 JSON 字符串。

設置過期時間 （TTL）

為防止緩存無限增長，您可以使用 EXPIRE 命令或帶有 SET 命令的 EX 選項為緩存數據設置過期時間（TTL - 生存時間）：

SET user:123 '{"id": 123, "name": "John Doe", "email": "john.doe@example.com"}' EX 3600  # Expires in 3600 seconds (1 hour)
EXPIRE user:123 3600 # Expires in 3600 seconds (1 hour)
TTL user:123 # Check the remaining time to live

這可確保在指定時間段后自動刪除緩存的數據，從而防止提供過時的數據。

數據序列化

緩存復雜數據結構時，您需要在將數據存儲在 Redis 中之前對其進行序列化，并在檢索數據后對其進行反序列化。常見的序列化格式包括 JSON 和 Protocol Buffers。

以下是在 Python 中使用 JSON 的示例：

import redis
import json# Connect to Redis
redis_client = redis.Redis(host='localhost', port=6379, db=0)# Data to cache
user_data = {"id": 123, "name": "John Doe", "email": "john.doe@example.com"}# Serialize the data to JSON
user_data_json = json.dumps(user_data)# Store the JSON string in Redis with an expiration time
redis_client.set('user:123', user_data_json, ex=3600)# Retrieve the data from Redis
cached_user_data_json = redis_client.get('user:123')# Deserialize the JSON string back to a Python dictionary
if cached_user_data_json:cached_user_data = json.loads(cached_user_data_json)print(cached_user_data)
else:print("Data not found in cache")

緩存失效策略

緩存失效是在基礎數據更改時刪除或更新緩存數據的過程。緩存失效有幾種策略：

• 基于 TTL 的失效： 在指定的 TTL 之后，數據會自動從緩存中刪除。這是最簡單的策略，但如果基礎數據在 TTL 過期之前發生更改，則可能會導致數據過時。
• 基于事件的失效： 當發生特定事件（例如數據庫更新）時，緩存將失效。此策略可確保緩存始終是最新的，但它需要與數據源進行更復雜的集成。
• 手動失效： 緩存由管理員或應用程序代碼手動失效。此策略對緩存失效提供了最大的控制，但它需要仔細監控和管理。

示例：緩存數據庫查詢

假設您有一個從數據庫中檢索用戶數據的函數：

import redis
import json
import time# Assume this function fetches data from a database
def get_user_from_db(user_id):# Simulate a database query with a delaytime.sleep(1)user_data = {"id": user_id, "name": f"User {user_id}", "email": f"user{user_id}@example.com"}return user_datadef get_user(user_id, redis_client):"""Retrieves user data from cache if available, otherwise fetches from the database,caches it, and returns it."""cache_key = f'user:{user_id}'cached_user_data = redis_client.get(cache_key)if cached_user_data:# Cache hitprint(f"Cache hit for user {user_id}")user_data = json.loads(cached_user_data)else:# Cache missprint(f"Cache miss for user {user_id}. Fetching from DB.")user_data = get_user_from_db(user_id)user_data_json = json.dumps(user_data)redis_client.set(cache_key, user_data_json, ex=3600)  # Cache for 1 hourreturn user_data# Example usage
redis_client = redis.Redis(host='localhost', port=6379, db=0)user1 = get_user(123, redis_client)
print(user1)user1_cached = get_user(123, redis_client) #This time it will be a cache hit
print(user1_cached)user2 = get_user(456, redis_client)
print(user2)

在此示例中，get_user 函數首先檢查用戶數據在 Redis 緩存中是否可用。如果是，則從緩存中檢索數據并返回數據。否則，將從數據庫中檢索數據，將其存儲在緩存中，過期時間為 1 小時，然后返回。

高級緩存技術

緩存防盜

當大量請求同時命中緩存，并且緩存已過期或為空時，就會發生緩存加速。這可能會使數據庫過載，因為所有請求都嘗試同時檢索數據。

要防止緩存踩踏，可以使用以下技術：

• Probabilistic Early Expiration（概率提前到期）： 您可以向過期時間添加一個小的隨機延遲，而不是為所有緩存條目設置固定的過期時間。這有助于分配數據庫上的負載。
• 鎖定： 當發生緩存未命中時，您可以獲取一個鎖，以防止其他請求同時嘗試從數據庫中檢索數據。只有第一個請求會檢索數據，將其存儲在緩存中，然后釋放鎖。
• 后臺刷新： 您可以在緩存過期之前在后臺刷新緩存。這可確保緩存始終是最新的，并降低緩存被踩踏的可能性。

使用 Redis 數據結構進行緩存

Redis 提供了各種可用于緩存不同類型數據的數據結構：

• Strings： 用于緩存簡單的鍵值對，例如用戶 ID 和名稱。
• Hashes： 用于緩存具有多個字段的對象，例如用戶配置文件。
• Lists： 用于緩存有序數據，例如最近的活動源。
• Sets： 用于緩存唯一數據，例如用戶角色。
• Sorted Sets： 用于緩存排名數據，例如排行榜。

選擇正確的數據結構可以提高緩存的效率和性能。

示例：緩存博客文章列表

假設您要緩存最近的博客文章列表。您可以使用 Redis 列表來存儲帖子 ID，然后在需要時從數據庫中檢索完整的帖子數據。

import redis
import json# Connect to Redis
redis_client = redis.Redis(host='localhost', port=6379, db=0)# Assume this function fetches blog posts from a database
def get_blog_posts_from_db():# Simulate a database queryblog_posts = [{"id": 1, "title": "Redis Caching", "content": "..."},{"id": 2, "title": "NoSQL Databases", "content": "..."},{"id": 3, "title": "Python Programming", "content": "..."}]return blog_postsdef get_recent_blog_posts(redis_client, limit=10):"""Retrieves recent blog posts from cache if available, otherwise fetches from the database,caches it, and returns it."""cache_key = 'recent_blog_posts'cached_post_ids = redis_client.lrange(cache_key, 0, limit - 1)if cached_post_ids:# Cache hitprint("Cache hit for recent blog posts")post_ids = [int(post_id) for post_id in cached_post_ids]# In a real application, you would fetch the full post data from the database# based on these IDs.  Here, we just return the IDs.return post_idselse:# Cache missprint("Cache miss for recent blog posts. Fetching from DB.")blog_posts = get_blog_posts_from_db()post_ids = [post['id'] for post in blog_posts]# Store the post IDs in Redisfor post_id in reversed(post_ids):  # Add in reverse order to maintain orderredis_client.lpush(cache_key, post_id)redis_client.expire(cache_key, 3600)  # Cache for 1 hourreturn post_ids[:limit]# Example usage
recent_posts = get_recent_blog_posts(redis_client)
print(recent_posts)recent_posts_cached = get_recent_blog_posts(redis_client) #This time it will be a cache hit
print(recent_posts_cached)

實踐練習

1. 實施 Cache-Aside 策略： 創建一個使用 Redis 緩存 API 調用結果的函數。該函數應首先檢查數據在緩存中是否可用。如果是，則返回緩存的數據。如果沒有，請進行 API 調用，將結果存儲在緩存中并指定過期時間，然后返回結果。
2. 實施緩存失效： 修改前面的函數，以便在底層數據更改時使緩存失效。您可以通過更新數據庫中的值或調用其他 API 終端節點來模擬數據更改。
3. 使用 Redis 哈希來緩存對象： 創建一個使用 Redis 哈希緩存用戶配置文件的函數。該函數應將每個用戶配置文件存儲為單獨的哈希值，其中包含 name、email 和其他相關信息的字段。
4. 實施緩存踩踏防護： 修改 API 緩存功能，以防止使用概率提前過期或鎖定的緩存踩踏。