一. 用 Redis 保存時間序列類型數據方案

我們日常開發中，有很多這種類似的場景，記錄某一個時刻下，某個目標的相關屬性或者狀態。 那么常規的，我們可以用時間戳作為Key，而這個目標的相關屬性我們可以將其轉化為JSON串或者通過字符串拼接的方式來保存為String類型，然后作為Value。

但是，這樣的存儲是針對于一個時間戳而言的，而實際環境中，往往需要記錄非常多的這樣的數據，甚至可能需要對這種數據進行統計、聚合、范圍查找等操作。那么可想而知，Redis中的String類型，雖然可以提供存儲功能，但是卻難以提供統計、聚合、范圍查找等這些復雜操作。 除此之外，String類型我在String內存開銷問題以及基本/擴展數據類型的使用這篇文章有提到，如果存儲的數據量太大，那么內存的占用是非常龐大的。耗內存。

同時用時間戳作為Key的數據，往往也有以下特點：

• 數據的插入比較頻繁。
• 讀操作的查詢模式的種類比較多（統計、聚合…）。

1.1 內存和范圍查找的支持性問題

場景：按照一定的時間間隔記錄某一個設備集群中每臺機器的溫度。

針對內存占用問題，我們可以選擇Hash類型去替代String。

# 傳統的String類型存儲
set 1659947381000 31
set 1659947392000 35
set 1659947413000 28
# 改為Hash存儲
set temperature 1659947381000 31 1659947392000 35 1659947413000 28

雖然Hash結構彌補了String類型在內存開銷上的短板。但是僅僅這樣是無法滿足數據的一些范圍查詢或者操作的。

那么針對范圍查找問題：因此我們可以再使用Sorted Set去保存相同的一份數據。這里做個解釋，為何要同時用兩個數據結構來存儲相同的數據：

1. Hash結構：用來提供單值查詢。
2. Sorted Set：提供范圍查詢。雖然將范圍的前后指定為一個相同的值，看起來像是單個值查詢，但是本質上依舊是范圍查詢，效率不如人家Hash來的快。

那么此時就可以用這樣的命令來進行查找：

zrangebyscore temperature 1659947381000 1659947413000 

此外，我們既然使用了兩種數據結構來保存相同的數據，就應該保證數據一致性。我們應該保證兩個數據結構中的數據是完全一樣的，不能出現哪個結構中的數據有少或者不一致的情況。因此我們可以在進行數據插入的時候，保障兩個操作的原子性。即使用簡單的事務操作：

• multi：事務開始。之后的操作將會放入一個隊列中，而不會真正的去執行。
• exec：事務結束，開始執行隊列中的一系列命令。

例如：

multi
hset temperature 1659947381000 25
zadd temperature 1659947381000 25
exec

那么Java對應的操作就是：

Transaction multi = jedis.multi();
multi.hset("temperature", "1659947381000", "25");
System.out.println("事務執行中：hash：" + multi.hget("temperature", "1659947381000"));
multi.zadd("temperatureZSet", 1659947381000L, "25");
System.out.println("事務執行中：sorted set：" + multi.zrangeByScore("temperatureZSet", 1659947381000L, 1659947381000L));
multi.exec();
System.out.println("**************事務執行完畢******************");
System.out.println("事務執行中：hash：" + jedis.hget("temperature", "1659947381000"));
System.out.println("事務執行中：sorted set：" + jedis.zrangeByScore("temperatureZSet", 1659947341000L, 1659947392000L));

注意：

• hash和sorted set兩個集合使用的key不能是同一個。
• 并且事務中不能使用普通的jedis對象。multi 對象擁有和jedis對象同樣的API操作。因為Redis中事務開啟后，執行的操作是放到隊列中的，并不是馬上執行的，因此需要做區分。

結果如下：

到這里，內存問題和范圍查找的問題已經解決了。雖然我們用了兩個數據類型來保存相同的一份數據，但是整體的內存消耗，是比全部用String類型存儲要節省的。那么接下來要解決的，就是聚合操作問題。

備注：

1. 到這里為止，如果業務上只涉及到時序的范圍查找，是可以同時用Hash和Sorted Set去替代傳統的String的。如果僅僅限于此，我個人建議1.2節可以不看。
2. Redis中對于事務的使用，在文章中提到的原子性問題也是有一定缺陷的。因為Redis中的事務并不像Mysql那樣，倘若在一個事務中，先后執行了A和B操作，但是在執行C操作的時候發生了錯誤，A和B的操作是不會回滾的。
3. Redis主要還是拿來做緩存比較多，這種專門的時序數據處理最好交給專門的時序數據庫處理，例如influxDB。
4. 1.2節內容僅供參考，并且實用性和實際操作起來是否簡單這個問題上，有待商榷，因為并不容易實現。（至少我寫這篇文章的時候，關于RedisTimeSeries的JavaAPI操作沒有找到）

1.2 聚合操作的支持性問題（僅供參考）

首先，我們當然可以在客戶端將相關的數據全部讀取過來，然后再客戶端自行完成聚合操作。但是倘若有這么幾個點：

1. 數據量很大。
2. 聚合操作的頻率很高。

那么這種情況下，就會有很多請求（包含了大量數據）在Redis和客戶端之間來回穿梭，就會造成資源的競爭，降低Redis的性能。

那么針對聚合操作問題：我們可以使用RedisTimeSeries，它支持在Redis實例上對時間維度進行聚合計算。

但是使用這個，卻比較麻煩，需要了解這么幾個點：

1. RedisTimeSeries是Redis的擴展模塊，原生Redis并不支持。
2. 使用的時候需要將Redis源碼單獨編譯成動態鏈接庫 redistimeseries.so，再使用 loadmodule 命令進行加載。

loadmodule redistimeseries.so

那么針對上述的聚合場景，使用RedisTimeSeries的大致流程如下：

# 創建時間序列數據集合,創建一個key為temperature ，數據的有效期為800s。（過后會自動刪除），同時這個集合的標簽屬性uid為1
ts.create temperature retention 800000 labels uid 1
# 插入數據
ts.add temperature 1659947381000 25
# 最新數據的獲取 ts.get只能返回最新的數據
ts.get temperature 
# 按照標簽過濾查詢
ts.mget FILTER uid = xxx
# 聚合計算 在[1659947371000 ,1659947381000]范圍內，按照每180s的時間間隔，對這個時間窗口內的數據做均值計算
ts.range temperature 1659947371000 1659947381000 AGGREGATION avg 180000

二. 用 Redis 實現消息隊列

前言：Redis是可以做消息隊列的，但是對于一些不允許出現消息丟失的情形下，例如金融支付操作。不要用Redis作為中間件，請使用專門的中間件去做存儲。例如Kafka、ActiveMQ、RabbitMQ等。具體原因下面分析。

首先消息隊列需要解決三個問題：

• 消息保序。
• 消息的重復消費問題。
• 消息的可靠性保證。

2.1 消息保序的實現

那么如何用Redis作為消息隊列呢？利用Redis中的List數據結構。

1. List這個數據結構本身就是FIFO，先進先出的順序對數據進行存儲的。
2. 實際操作上，生產者通過lpush命令將數據寫入List中。消費者端則通過rpop命令將其彈出。

這是一般的操作。但是光憑這樣的操作并不滿足一個合格的消息中間件具備的條件。因為在生產者向Redis中寫入數據的時候，Redis并不會主動地通知消費者有新消息寫入了。此時消費者只能通過這樣的偽代碼來實現輪詢：

while(true){
	String json = jedis.rpop('key');
	process(json);
}

問題：這樣的無限循環，會導致CPU一直消耗在這里執行rpop命令。造成性能損失。

解決：建議使用brpop命令，即阻塞式讀取，客戶端在沒有讀到隊列數據時，自動阻塞，直到有新的數據寫入隊列，再開始讀取新數據。

2.2 重復消費問題解決

對于消息的重復消費問題，我們只需要提供一個唯一標識，然后消費的時候做判斷即可。

1. 生產者端：發送消息的時候，給消息里面塞一個唯一標識。
2. 消費者端：將消費完成的消息的唯一標識記錄下來。在后續消費的時候，都要反查一遍先。

# 唯一標識:主題:內容
lpush key 1000001:title:helloworld

2.3 消息可靠性保證

背景：當消費者程序從Redis中讀取一條消息并做處理，但是還沒處理完成的時候就發生了宕機，那么Redis中這條數據已經被剔除，但這個數據并沒有被真正的消費掉。怎么辦？

解決：生產者在推消息給Redis的時候，使用 BRPOPLPUSH 命令，其作用如下：

1. 在生產者推消息的時候，Redis 會把這個消息再插入到另一個 List 留存。
2. 這樣一來，如果消費者程序讀了消息但沒能正常處理，等它重啟后，就可以從備份 List 中重新讀取消息并處理。

綜上所述，常規情況下：

1. 生產者端使用BRPOPLPUSH 命令往Redis中推數據，同時塞入唯一標識。
2. 消費者端使用brpop命令。防止無限循環調用rpop()命令。將消費過的消息的唯一標識做數據存儲。
3. 消費者倘若消費某個消息成功，由于生產者端往兩個List都插入了數據，此時最好將備份隊列中的消息刪除，避免備份隊列中存儲過多過期數據，造成內存浪費。

2.4 Redis 做中間件的優劣勢

先來說下Redis做中間件的優勢：

1. 用Redis作為消息隊列，由于Redis的特性，在內存上操作，因此性能高。
2. API操作起來非常方便，沒有復雜的操作，部署輕量。 Kafka的操作相比之下就會復雜許多。維護成本也要更高點。

Redis做中間件的劣勢：可能出現數據丟失。 有這么個幾個場景：

1. AOF策略為每秒寫盤。該過程為異步，若Redis發生宕機，會丟失1秒的數據。若改為同步寫盤，則會導致性能下降。
2. 在主從集群下，倘若寫操作的頻率非常大，那么主從的數據同步就會存在延遲，那么在進行主從切換的時候，也可能存在數據丟失問題。詳細可以看Redis - Redis主從數據一致性和哨兵機制。
3. 無法保證數據的完整性，而像Kafka這樣的專業中間件，副本等機制保證了數據的可靠性。哪怕集群的某個節點掛掉了，也不會丟失數據。詳細可以參考Kafka復習計劃 - Kafka基礎知識以及集群參方案和參數。