Redis實戰：20w緩存數據如何保持熱度不減退？

言：在處理大規模數據集時，保持熱數據的高性能和可用性是一個關鍵問題。特別是在擁有 200 萬數據中，有 20w 數據存儲在 Redis 中的情況下，我們需要采取一些策略和技術來確保這些熱數據的熱度不減退。本文將介紹一些應對挑戰的方法，幫助我們保持數據的熱度，使其永不減退。

數據熱度的重要性

數據熱度指的是數據被頻繁訪問和使用的程度。熱數據通常是應用程序的關鍵部分，對性能和用戶體驗有重要影響。因此，保持熱數據的熱度對于應用程序的成功至關重要。

保持熱度不減退

處理大規模數據集帶來了存儲、查詢和更新的效率挑戰。而在這 200 萬數據中，我們需要特別關注的是存儲在 Redis 中的 20w 熱數據。這些熱數據需要保持高性能和可用性，避免熱度減退。下面將介紹一些策略和技術來解決這些挑戰。

策略和技術

1. 數據分析和預測

通過利用數據分析工具和算法，我們可以識別和預測熱數據。通過分析歷史數據的訪問模式和趨勢，我們可以預測未來的熱數據。這樣可以幫助我們更好地針對這些數據做出決策，例如優先存儲在高性能的 Redis 中。

2. 緩存策略

利用緩存技術，我們可以將熱數據存儲在高性能的內存中，例如 Redis。通過使用 LRU（最近最少使用）或 LFU（最不常用）等緩存替換算法，我們可以保證熱數據的高命中率，從而提高性能和響應時間。以下是一個使用 Redis 緩存的示例代碼：

import redis

# 創建Redis連接
r = redis.Redis(host='localhost', port=6379, db=0)

# 設置熱數據
r.set('hot_data_1', 'value_1')
r.set('hot_data_2', 'value_2')
r.set('hot_data_3', 'value_3')

# 獲取熱數據
data_1 = r.get('hot_data_1')
data_2 = r.get('hot_data_2')

3. 數據預加載

在系統啟動或低峰期，我們可以預先加載熱數據到緩存中。這樣可以避免熱數據第一次訪問時的延遲和性能問題，提前準備好熱數據的訪問路徑。以下是一個簡單的數據預加載示例代碼：

import redis

# 創建Redis連接
r = redis.Redis(host='localhost', port=6379, db=0)

# 預加載熱數據
def preload_hot_data():
    hot_data = ['hot_data_1', 'hot_data_2', 'hot_data_3']
    for data in hot_data:
        value = fetch_data_from_database(data)  # 從數據庫獲取數據
        r.set(data, value)

# 執行數據預加載
preload_hot_data()

4. 數據更新和同步

保持熱數據與底層數據的一致性非常重要。我們可以采用實時或異步的方式將底層數據的變更同步到熱數據中。一種常見的方法是使用發布-訂閱模式（Pub-Sub）來實現數據更新的實時通知機制。當底層數據發生變化時，即時更新熱數據，以保持其與底層數據的一致性。以下是一個使用 Pub-Sub 模式進行數據更新和同步的示例代碼：

import redis
import json

# 創建Redis連接
r = redis.Redis(host='localhost', port=6379, db=0)

# 訂閱底層數據的更新通知
def update_hot_data(channel, message):
    data = json.loads(message)
    hot_data = data['hot_data']
    value = fetch_data_from_database(hot_data)  # 從數據庫獲取最新數據
    r.set(hot_data, value)

# 創建訂閱者
pubsub = r.pubsub()

# 訂閱底層數據的更新通知
pubsub.subscribe('data_updates')

# 監聽數據更新通知
for message in pubsub.listen():
    if message['type'] == 'message':
        update_hot_data(message['channel'].decode('utf-8'), message['data'].decode('utf-8'))

5. 水平擴展和負載均衡

隨著數據規模的增長，我們可能需要進行水平擴展和負載均衡，以應對高并發和大規模數據的挑戰。通過將熱數據分片存儲在多個 Redis 節點上，并使用負載均衡算法將請求分發到這些節點上，我們可以提高系統的性能和可伸縮性。以下是一個使用 Redis Cluster 進行水平擴展和負載均衡的示例代碼：

from rediscluster import RedisCluster

# Redis Cluster節點配置
startup_nodes = [
    {'host': 'node1', 'port': 6379},
    {'host': 'node2', 'port': 6379},
    {'host': 'node3', 'port': 6379}
]

# 創建Redis Cluster連接
r = RedisCluster(startup_nodes=startup_nodes, decode_responses=True)

# 設置熱數據
r.set('hot_data_1', 'value_1')
r.set('hot_data_2', 'value_2')
r.set('hot_data_3', 'value_3')

# 獲取熱數據
data_1 = r.get('hot_data_1')
data_2 = r.get('hot_data_2')

6. 監控和調優：

對于處理大規模數據集的系統，監控和調優是保持熱數據熱度的關鍵。我們可以使用監控工具來實時監測熱數據的訪問情況、性能指標和系統負載。根據監控數據，我們可以進行性能優化和資源調配，以保持熱數據的高性能。以下是一個使用 Redis 的監控和調優示例代碼：

import redis

# 創建Redis連接
r = redis.Redis(host='localhost', port=6379, db=0)

# 獲取熱數據的訪問次數
def get_hot_data_access_count():
    hot_data_access_count = {}
    keys = r.keys('hot_data_*')
    for key in keys:
        count = r.get(key)
        hot_data_access_count[key] = count
    return hot_data_access_count

# 監控熱數據的訪問情況
def monitor_hot_data():
    while True:
        hot_data_access_count = get_hot_data_access_count()
        # 打印熱數據的訪問次數
        print(hot_data_access_count)
        # 做一些性能優化和資源調配的決策

結論

通過采取數據分析和預測、緩存策略、數據預加載、數據更新和同步、水平擴展和負載均衡以及監控和調優等策略和技術，我們可以有效應對保持熱度不減退的挑戰。這些方法可以幫助我們保持熱數據的高性能和可用性，提供出色的用戶體驗和系統性能。

參考文獻：

Smith, J. (2020). Scaling Redis at Twitter. Retrieved from https://blog.twitter.com/engineering/en_us/topics/infrastructure/2020/scaling-redis-at-twitter.html
Redis Labs. (n.d.). Redis as a Caching Layer. Retrieved from https://redislabs.com/solutions/use-cases/caching/

以上是一份關于保持熱度不減退的詳細文章，其中包含了更多的解釋和代碼示例。如果您有任何進一步的問題或需要更多幫助，請隨時提問。

在前面

我一直以來使用redis的時候，很多低烈度需求(并發要求不是很高)需要用到消息隊列的時候，在項目本身已經使用了Redis的情況下都想直接用Redis來做消息隊列，而不想引入新的服務，kafka和RabbitMQ等；

奈何這兄弟一直不給力；

雖然 Redis 的Pub/Sub 是實現了發布/訂閱的，但這家伙最坑的是：丟數據

由于Pub/Sub 只是簡單的實現了發布訂閱模式，簡單的溝通起生產者和消費者，當接收生產者的數據后并立即推送或者說轉發給訂閱消費者，并不會做任何的持久化、存儲操作。由此：

消費者（客戶端）掉線；
? 消費者未訂閱（所以使用的時候一定記得先訂閱再生產）；
? 服務端宕機；
? 消費者消費不過來，消息堆積（生產數據受數據緩沖區限制）；

以上情況都會導致生產數據的丟失，基于上坑，據我所知大家很少使用Pub/Sub ；

不過官方的哨兵集群通信的時候就是用的Pub/Sub；

然后，各路大佬結合隊列、阻塞等等實現了各種各樣的方案，主要是使用：BLPOP+LPUSH 的實現

這里就不一一展開了，有興趣請看葉老板文章；

可能是各種實現都會帶來各種的問題，redis的官方也看到了社區的掙扎。終于，到了Redis5.0，官方帶來了消息隊列的實現：Stream。

Redis Stream介紹

簡單來說Redis Stream 就是想用Redis 做消息隊列的最佳推薦；

XADD--發布消息

XADD stream1 * name hei age 18
XADD stream1 * name zhangshan age 19 #再發一條

127.0.0.1:6379> XADD stream1 * name hei age 18
"1631628884174-0"
127.0.0.1:6379> XADD stream1 * name zhangshan age 19 
"1631628890025-0"

其中的'*'表示讓 Redis 自動生成唯一的消息 ID,格式是「時間戳-自增序號」

XREAD--訂閱消息

訂閱消息

XREAD COUNT 5 STREAMS stream1 0-0

127.0.0.1:6379> XREAD COUNT 5 STREAMS stream1 0-0 
1) 1) "stream1"
   2) 1) 1) "1631628884174-0"
         2) 1) "name"
            2) "hei"
            3) "age"
            4) "18"
      2) 1) "1631628890025-0"
         2) 1) "name"
            2) "zhangshan"
            3) "age"
            4) "19"

'0-0' 表示從開頭讀取
如果需繼續拉取下一條，需傳入上一條消息的id

阻塞等待消息

XREAD COUNT 5 BLOCK 50000 STREAMS stream1 1631628890025-0

阻塞等待消息id ‘1631628890025-0’ 后的消息
50000 阻塞時間（毫秒） ‘0’ 表示無限期阻塞

從到這里就可以看出 Pub/Sub多端訂閱的最大優點，Stream也是支持的。有的同學很快就發現問題了，這里多端訂閱后，沒有消息確認ACK機制。

沒錯，因為現在所有的消費者都是訂閱共同的消息，多端訂閱，如果某個客戶端ACK某條消息后，其他端消費不了，就實現不了多端消費了。

由此，引出分組：GROUP

GROUP--訂閱分組消息(多端訂閱)

同樣先發布消息

XADD stream1 * name hei age 18
XADD stream1 * name zhangshan age 19

127.0.0.1:6379> XADD stream1 * name hei age 18
"1631629080208-0"
127.0.0.1:6379> XADD stream1 * name zhangshan age 19 
"1631629084083-0"

XGROUP CREATE 創建分組

創建分組1

XGROUP CREATE stream1 group1 0-0

127.0.0.1:6379> XGROUP CREATE stream1 group1 0-0  
OK

‘0-0’ 表示從開頭讀取
'>' 表示讀取最新，未被消費過的消息

XREADGROUP--分組讀取

分組 group1

XREADGROUP GROUP group1 consumer1 COUNT 5 STREAMS stream1 >

consumer1 消費者名稱， redis服務器會記住第一次使用的消費者名稱；


127.0.0.1:6379> XREADGROUP GROUP group1 consumer1 COUNT 5 STREAMS stream1 >  
1) 1) "stream1"
   2) 1) 1) "1631628884174-0"
         2) 1) "name"
            2) "hei"
            3) "age"
            4) "18"
      2) 1) "1631628890025-0"
         2) 1) "name"
            2) "zhangshan"
            3) "age"
            4) "19"
      3) 1) "1631629080208-0"
         2) 1) "name"
            2) "hei"
            3) "age"
            4) "18"
      4) 1) "1631629084083-0"
         2) 1) "name"
            2) "zhangshan"
            3) "age"
            4) "19"
127.0.0.1:6379> XREADGROUP GROUP group1 consumer1 COUNT 5 STREAMS stream1 >  
(nil)

同樣
‘0-0’ 表示從開頭讀取
'>' 表示讀取最新，未被消費過的消息（可以看到命令執行第二遍已經讀不到新消息了）

分組 group2

127.0.0.1:6379> XGROUP CREATE stream1 group2 0-0  
OK
127.0.0.1:6379> XREADGROUP GROUP group2 consumer1 COUNT 5 STREAMS stream1 >  
1) 1) "stream1"
   2) 1) 1) "1631628884174-0"
         2) 1) "name"
            2) "hei"
            3) "age"
            4) "18"
      2) 1) "1631628890025-0"
         2) 1) "name"
            2) "zhangshan"
            3) "age"
            4) "19"
      3) 1) "1631629080208-0"
         2) 1) "name"
            2) "hei"
            3) "age"
            4) "18"
      4) 1) "1631629084083-0"
         2) 1) "name"
            2) "zhangshan"
            3) "age"
            4) "19

可以看到可以讀到同樣的消息，多端訂閱沒有問題；

當然分組也支持阻塞讀取：

#和XREAD一樣
XREAD COUNT 5 BLOCK 0 STREAMS queue 1618469127777-0 

#分組阻塞
XREADGROUP GROUP group2 consumer1 COUNT 5 BLOCK 0 STREAMS stream1 >

‘0’ 表示無限期阻塞，單位（毫秒）

XPENDING--待處理消息

消息使用XREADGROUP 讀取后會進入待處理條目列表（PEL）；

我們看看：

 XPENDING stream1 group2

127.0.0.1:6379>  XPENDING stream1 group2
1) (integer) 4
2) "1631628884174-0"
3) "1631629084083-0"
4) 1) 1) "consumer1"
      2) "4"

表示：

(integer) 4 //表示當前消費者組的待處理消息的數量
"1631628884174-0" //消息最大id
"1631629084083-0" //最小id

"consumer1" // 消費者名稱
"4" //消費者待處理消息數量

XACK--刪除已處理消息（消息確認機制）

我們已經知道group2待處理消息有4條，我們從頭讀取看看：

XREADGROUP GROUP group2 consumer1 COUNT 5 STREAMS stream1 0-0

127.0.0.1:6379> XREADGROUP GROUP group2 consumer1 COUNT 5 STREAMS stream1 0-0
1) 1) "stream1"
   2) 1) 1) "1631628884174-0"
         2) 1) "name"
            2) "hei"
            3) "age"
            4) "18"
      2) 1) "1631628890025-0"
         2) 1) "name"
            2) "zhangshan"
            3) "age"
            4) "19"
      3) 1) "1631629080208-0"
         2) 1) "name"
            2) "hei"
            3) "age"
            4) "18"
      4) 1) "1631629084083-0"
         2) 1) "name"
            2) "zhangshan"
            3) "age"
            4) "19"

假設最后一條消息 ‘1631629084083-0’ 我已處理完成

127.0.0.1:6379> XACK stream1 group2 1631629084083-0
(integer) 1

再看：

127.0.0.1:6379> XREADGROUP GROUP group2 consumer1 COUNT 5 STREAMS stream1 0-0
1) 1) "stream1"
   2) 1) 1) "1631628884174-0"
         2) 1) "name"
            2) "hei"
            3) "age"
            4) "18"
      2) 1) "1631628890025-0"
         2) 1) "name"
            2) "zhangshan"
            3) "age"
            4) "19"
      3) 1) "1631629080208-0"
         2) 1) "name"
            2) "hei"
            3) "age"
            4) "18"

127.0.0.1:6379>  XPENDING stream1 group2
1) (integer) 3
2) "1631628884174-0"
3) "1631629080208-0"
4) 1) 1) "consumer1"
      2) "3"

可以清楚看到goroup2 待處理消息剩下3條；

這時 Redis 已經把這條消息標記為「處理完成」不再追蹤；

Stream在Asp.net Core中的使用

private static string _connstr = "172.16.3.119:6379";
private static string _keyStream = "stream1";
private static string _nameGrourp = "group1";
private static string _nameConsumer = "consumer1";

發布：

csRedis.XAdd(_keyStream, "*", ("name", "message1"));

static async Task CsRedisStreamConsumer()
{
    Console.WriteLine("CsRedis StreamConsumer start!");

    var csRedis = new CSRedis.CSRedisClient(_connstr);
    csRedis.XAdd(_keyStream, "*", ("name", "message1"));

    try
    {
        csRedis.XGroupCreate(_keyStream, _nameGrourp);
    }
    catch { }

    (string key, (string id, string[] items)[] data)[] product;
    (string Pid, string Platform, string Time) data = (null, null, null);

    while (true)
    {
        try
        {
            product = csRedis.XReadGroup(_nameGrourp, _nameConsumer, 1, 10000, (_keyStream, ">"));
            if (product?.Length > 0 == true && product[0].data?.Length > 0 == true)
            {
                Console.WriteLine($"message-id:{product.FirstOrDefault().data.FirstOrDefault().id}");

                product.FirstOrDefault().data.FirstOrDefault().items.ToList().ForEach(value =>
                {
                    Console.WriteLine($"    {value}");
                });

                //csRedis.XAck(_keyStream, _nameGrourp, product[0].data[0].id);
            }
        }
        catch (Exception)
        {
            //throw;
        }
    }
}

CSRedisCore

這里的超時報錯可通過修改連接參數：syncTimeout 解決
CSRedisCore支持阻塞讀取；

StackExchange.Redis

發布：

db.StreamAdd(_keyStream, "name", "message1", "*");

static async Task StackExchangeRedisStreamConsumer()
{
    Console.WriteLine("StackExchangeRedis StreamConsumer start!");

    var redis = ConnectionMultiplexer.Connect(_connstr);
    var db = redis.GetDatabase();

    try
    {
        ///初始化方式1
        //db.StreamAdd(_keyStream, "name", "message1", "*");
        //db.StreamCreateConsumerGroup(_keyStream, _nameGrourp);

        //方式2
        db.StreamCreateConsumerGroup(_keyStream, _nameGrourp, StreamPosition.NewMessages);
    }
    catch { }

    StreamEntry[] data = null;

    while (true)
    {
        data = db.StreamReadGroup(_keyStream, _nameGrourp, _nameConsumer, ">", count: 1, noAck: true);

        if (data?.Length > 0 == true)
        {
            Console.WriteLine($"message-id:{data.FirstOrDefault().Id}");

            data.FirstOrDefault().Values.ToList().ForEach(c =>
            {
                Console.WriteLine($"    {c.Name}:{c.Value}");
            });

            db.StreamAcknowledge(_keyStream, _nameGrourp, data.FirstOrDefault().Id);
        }
    }
}

StackExchange.Redis 有點比較坑的是不存在阻塞讀取；理由：https://stackexchange.github.io/StackExchange.Redis/PipelinesMultiplexers.html#multiplexing

QA

Q：Stream是否支持AOF、RDB持久化？

A：支持，其它數據類型一樣，每個寫操作，也都會寫入到 RDB 和 AOF 中。

Q：Stream是否還是會丟數據？若是，何種情況下？；

A：會；1、AOF是定時寫盤的，如果數據還在內存中時redis服務宕機就會；2、主從切換時（從庫還未同步完成主庫發來的數據，就被提成主庫）

總結

技術中有的時候沒有“銀彈”，只有更適合的技術，汝之蜜糖彼之砒霜；

很多時候的技術選型都是個比較麻煩的東西，對選型人的要求很高；你可能不是只需要熟悉其中的一種路線，而是要踩過各種各樣的坑，再根據當前受限的環境，選擇比較適合目前需求/團隊的；

回到Stream上，我認為目前Stream能滿足挺大部分隊列需求；

特別是“在項目本身已經使用了Redis的情況下都想直接用Redis來做消息隊列，而不想引入新的更專業的mq，比如kafka和RabbitMQ的時候”

當然，最終決定需要用更專業的mq與否的，還是需求；

# 揭秘神器：Node.js開發者必備的Redis客戶端可視化管理工具

### 引言：Redis與Node.js的不解之緣

在現代Web開發領域，尤其是Node.js生態中，Redis憑借其高性能、內存存儲、豐富的數據結構以及實時性等特性，成為了開發者手中不可或缺的緩存和數據存儲利器。然而，如何高效、直觀地管理和操作Redis數據庫，對開發者來說至關重要。本文將為你揭示一款專為Node.js開發者設計的Redis客戶端可視化管理工具——Redis Desktop Manager（簡稱RDM），助你在開發過程中如虎添翼。

### 一、Redis Desktop Manager：功能強大且易用

. 界面友好，操作直觀

Redis Desktop Manager以其簡潔明了的界面設計和人性化的交互體驗，讓開發者能輕松實現Redis數據的可視化管理。通過圖形化界面，無論是查看鍵值對、執行命令，還是監控Redis服務器狀態，都變得簡單易行。

2. 支持多平臺及多種Redis版本

RDM支持Windows、macOS和Linux三大主流操作系統，并兼容Redis 2.x至最新版本，確保無論在哪種開發環境下，都能提供一致且高效的使用體驗。

3. 多數據庫連接與分組管理

RDM允許用戶同時連接并管理多個Redis實例，支持自定義數據庫分組，方便在不同項目間快速切換，極大提高了工作效率。

### 二、實戰演練：使用Redis Desktop Manager進行數據操作

1. 連接Redis服務

javascript
// 假設你的Redis服務器配置如下：
{
  host: 'localhost',
  port: 6379,
  password: 'your_password'
}

首先，打開RDM，點擊“Add New Connection”，輸入Redis服務器地址、端口、密碼等相關信息，即可成功連接到Redis服務。

2. 數據庫操作示例

- **插入鍵值對**：在RDM左側鍵空間列表中選擇一個數據庫，然后在右側操作區域直接輸入`SET key value`，例如`SET myKey myValue`，點擊執行按鈕或按回車即可。

- **查詢鍵值**：查詢`myKey`對應的值，只需輸入`GET myKey`并執行。

- **其他高級操作**：如集合（Set）、有序集合（ZSet）、哈希表（Hash）等復雜數據類型的增刪查改，均可在RDM的命令行界面便捷完成。

### 三、進階功能：深度探索與性能監控

1. 數據備份與恢復

RDM提供了便捷的數據導出導入功能，可以將Redis中的數據以RDB或AOF格式備份，并在需要時恢復，保障數據安全。

2. 實時監控Redis性能

在線咨詢

上一篇：HTML常用標簽概述（可用于網店裝修）
下一篇：JavaScript命名沖突不可避免？

您的項目需求

*請認真填寫需求信息，我們會在24小時內與您取得聯系。

整合營銷服務商