如何正確使用Redis Streams？

作者：布加迪編譯 2018-12-05 09:00:00

Redis Streams是一種新的Redis數(shù)據(jù)結(jié)構(gòu)，用于管理生產(chǎn)者和消費(fèi)者之間的數(shù)據(jù)通道。本文介紹了如何入手。

【51CTO.com快譯】 Redis是一種內(nèi)存中的多模式數(shù)據(jù)庫，適用于諸多使用場合，包括內(nèi)容緩存、會(huì)話存儲(chǔ)、實(shí)時(shí)分析、消息代理和數(shù)據(jù)流。去年，我撰文介紹了如何使用Redis Pub/Sub、Lists和Sorted Sets用于實(shí)時(shí)數(shù)據(jù)流處理，詳見https://www.infoworld.com/article/3212768/database/how-to-use-redis-for-real-time-stream-processing.html?，F(xiàn)在Redis 5.0已發(fā)布，Redis有了一種全新的數(shù)據(jù)結(jié)構(gòu)來管理數(shù)據(jù)流。

有了Redis Streams數(shù)據(jù)結(jié)構(gòu)，可以執(zhí)行比Pub/Sub、Lists和Sorted Sets豐富得多的操作。Redis Streams有眾多優(yōu)點(diǎn)，它讓你能夠執(zhí)行下列操作：

收集大量高速傳來的數(shù)據(jù)(唯一的瓶頸是你的網(wǎng)絡(luò)I/O);
在許多生產(chǎn)者和許多消費(fèi)者之間建立數(shù)據(jù)通道;
即使生產(chǎn)者和消費(fèi)者運(yùn)作的速度不一樣，也能高效地管理數(shù)據(jù)的消費(fèi);
你的消費(fèi)者離線或斷開連接時(shí)確保數(shù)據(jù)持久化;
在生產(chǎn)者和消費(fèi)者之間異步通信;
擴(kuò)大消費(fèi)者的數(shù)量;
消費(fèi)者在消費(fèi)數(shù)據(jù)過程中失效時(shí)，實(shí)現(xiàn)類似事務(wù)的數(shù)據(jù)安全性;
有效地使用主內(nèi)存。

Redis Streams的最大優(yōu)點(diǎn)是它內(nèi)置于Redis中，因此部署或管理Redis Streams無需額外的步驟。我在本文中將逐步介紹使用Redis Streams的基本方面，包括如何將數(shù)據(jù)添加到數(shù)據(jù)流、如何讀取數(shù)據(jù)(一次性讀取、異步讀取和到達(dá)時(shí)讀取等)，以滿足消費(fèi)者的不同使用場合。

一、了解Redis Streams中的數(shù)據(jù)流

Redis Streams提供了一種“只允許追加”(append only)的數(shù)據(jù)結(jié)構(gòu)，與日志類似。它提供了可以將數(shù)據(jù)源添加到數(shù)據(jù)流、使用數(shù)據(jù)流以及監(jiān)控和管理如何消費(fèi)數(shù)據(jù)的命令。 Streams數(shù)據(jù)結(jié)構(gòu)很靈活，讓你可以以幾種方式來連接生產(chǎn)者和消費(fèi)者。

圖1. Redis Streams的簡單應(yīng)用，只有一個(gè)生產(chǎn)者和一個(gè)消費(fèi)者

圖1表明了Redis Streams的基本用法。單單一個(gè)生產(chǎn)者充當(dāng)數(shù)據(jù)源，消費(fèi)者是將數(shù)據(jù)發(fā)送給相關(guān)接收者的消息傳遞應(yīng)用程序。

圖2.多個(gè)消費(fèi)者從Redis Streams讀取數(shù)據(jù)的應(yīng)用

圖2中，一個(gè)公共數(shù)據(jù)流被多個(gè)消費(fèi)者使用。使用Redis Streams，消費(fèi)者可以按照自己的節(jié)奏來讀取和分析數(shù)據(jù)。

在下一個(gè)應(yīng)用中，如圖3所示，情況變得復(fù)雜一點(diǎn)。該服務(wù)從多個(gè)生產(chǎn)者接收數(shù)據(jù)，并將所有數(shù)據(jù)存儲(chǔ)在Redis Streams數(shù)據(jù)結(jié)構(gòu)中。該應(yīng)用有多個(gè)消費(fèi)者從Redis Streams讀取數(shù)據(jù)，讀取數(shù)據(jù)的還有消費(fèi)者組(consumer group)，消費(fèi)者組支持無法與生產(chǎn)者保持同樣速度的消費(fèi)者。

圖3. Redis Streams支持多個(gè)生產(chǎn)者和消費(fèi)者

二、用Redis Streams將數(shù)據(jù)添加到數(shù)據(jù)流

圖3中的圖表只顯示了向Redis Stream添加數(shù)據(jù)的一種方法。雖然一個(gè)或多個(gè)生產(chǎn)者可以向數(shù)據(jù)結(jié)構(gòu)添加數(shù)據(jù)，但任何新數(shù)據(jù)始終追加到數(shù)據(jù)流的末尾。

1.添加數(shù)據(jù)的默認(rèn)方法

這是向Redis Streams添加數(shù)據(jù)的最簡單方法：

XADD mystream * name Anna  
XADD mystream * name Bert  
XADD mystream * name Cathy

在上述命令中，XADD是Redis命令，mystream是數(shù)據(jù)流的名稱，Anna、Bert和Cathy是每一行添加的名稱，而*操作符告訴Redis為每一行自動(dòng)生成識(shí)別符。這個(gè)命令得出三個(gè)mystream條目：

1518951481323-0 name Cathy  
1518951480723-0 name Bert  
1518951480106-0 name Anna

2.針對(duì)每個(gè)條目，為數(shù)據(jù)添加用戶管理的ID

Redis讓你可以為每個(gè)條目維護(hù)你自己的識(shí)別符(見下面)。雖然這在一些情況下很有用，但依賴自動(dòng)生成的ID通常來得更簡單：

XADD mystream 10000000 name Anna  
XADD mystream 10000001 name Bert  
XADD mystream 10000002 name Cathy

這得出下列的mystream條目：

10000002-0 name Cathy  
10000001-0 name Bert  
10000000-0 name Anna

3.為數(shù)據(jù)添加最大限制

你可以為數(shù)據(jù)流設(shè)置條目最大數(shù)：

XADD mystream MAXLEN 1000000 * name Anna  
XADD mystream MAXLEN 1000000 * name Bert  
XADD mystream MAXLEN 1000000 * name Cathy

數(shù)據(jù)流達(dá)到1000000個(gè)左右條目的長度時(shí)，該命令驅(qū)逐舊條目。

一個(gè)小貼士：Redis Streams將數(shù)據(jù)存儲(chǔ)在基樹(radix tree)的宏節(jié)點(diǎn)中。每個(gè)宏節(jié)點(diǎn)有幾個(gè)數(shù)據(jù)項(xiàng)(通常在幾十個(gè)左右)。如果添加一個(gè)近似的MAXLEN值(如下所示)，就沒必要為每次插入處理宏節(jié)點(diǎn)。如果幾十個(gè)數(shù)(比如1000000或1000050)對(duì)你來說關(guān)系不大，可以用近似字符(~)來調(diào)用命令，從而優(yōu)化性能。

XADD mystream MAXLEN ~ 1000000 * name Anna  
XADD mystream MAXLEN ~ 1000000 * name Bert  
XADD mystream MAXLEN ~ 1000000 * name Cathy

三、用Redis Streams消費(fèi)來自數(shù)據(jù)流的數(shù)據(jù)

Redis Streams結(jié)構(gòu)提供了一套豐富的命令和功能，以便以多種方式消費(fèi)數(shù)據(jù)。

1.從數(shù)據(jù)流的開頭讀取所有內(nèi)容

場景：數(shù)據(jù)流已含有你需要處理的數(shù)據(jù)，而且你想從開頭開始處理數(shù)據(jù)。

為此你要使用的命令是XREAD，它讓你可以從數(shù)據(jù)的開頭讀取所有或前N個(gè)條目。一條最佳實(shí)踐是，逐頁讀取數(shù)據(jù)始終是好主意。想從數(shù)據(jù)流的開頭讀取多達(dá)100個(gè)條目，命令是：

XREAD COUNT 100 STREAMS mystream 0

假設(shè)1518951481323-0是你在上一個(gè)命令中收到的數(shù)據(jù)項(xiàng)的最后一個(gè)ID，你可以運(yùn)行該命令，檢索下100個(gè)條目：

XREAD COUNT 100 STREAMS mystream 1518951481323-1

2.異步消費(fèi)數(shù)據(jù)(通過阻塞調(diào)用)

場景：你的消費(fèi)者消費(fèi)和處理數(shù)據(jù)的速度比數(shù)據(jù)添加到數(shù)據(jù)流的速度還快。

在許多使用場合下，消費(fèi)者讀取的速度比生產(chǎn)者向數(shù)據(jù)流添加數(shù)據(jù)的速度還快。這種情況下，你希望消費(fèi)者等待、新數(shù)據(jù)到達(dá)時(shí)接到通知。BLOCK選項(xiàng)讓你可以指定等待新數(shù)據(jù)的時(shí)長：

XREAD BLOCK 60000 STREAMS mystream 1518951123456-1

在這里，XREAD返回1518951123456-1之后的所有數(shù)據(jù)。如果之后沒有數(shù)據(jù)，查詢將等待N= 60秒，直至新數(shù)據(jù)到達(dá)，然后超時(shí)中斷。如果你想要無限期地阻止該命令，按如下方式調(diào)用XREAD：

XREAD BLOCK 0 STREAMS mystream 1518951123456-1

注意：在該示例中，你還可以使用XRANGE命令來逐頁檢索數(shù)據(jù)。

3.只讀取剛到達(dá)的新數(shù)據(jù)

場景：你只對(duì)處理從當(dāng)前時(shí)間點(diǎn)開始的新數(shù)據(jù)集有興趣。

你反復(fù)讀取數(shù)據(jù)時(shí)，從上次停下來的地方重新開始始終是個(gè)好主意。比如在前一個(gè)示例中，你進(jìn)行了阻塞調(diào)用以讀取大于1518951123456-1的數(shù)據(jù)。然而，你可能不知道最新的ID。在這種情況下，可以用$符號(hào)開始讀取數(shù)據(jù)流，該符號(hào)告訴XREAD命令只檢索新數(shù)據(jù)。由于該調(diào)用使用的BLOCK選項(xiàng)是60秒，它將等到數(shù)據(jù)流中有一些數(shù)據(jù)。

XREAD BLOCK 60000 STREAMS mystream $

這種情況下，你將開始用$選項(xiàng)讀取新數(shù)據(jù)。然而，不該用$選項(xiàng)進(jìn)行后續(xù)調(diào)用。比如說，如果1518951123456-0是之前調(diào)用中檢索的數(shù)據(jù)的ID，你的下一個(gè)調(diào)用應(yīng)該是：

XREAD BLOCK 60000 STREAMS mystream 1518951123456-1

4.迭代數(shù)據(jù)流以讀取過去的數(shù)據(jù)

場景：你的數(shù)據(jù)流已有足夠的數(shù)據(jù)，你想查詢它已分析到目前為止收集的數(shù)據(jù)。

可以分別使用XRANGE和XREVRANGE，以向前或向后的方向讀取兩個(gè)條目之間的數(shù)據(jù)。在該示例中，命令讀取1518951123450-0和1518951123460-0之間的數(shù)據(jù)：

XRANGE mystream 1518951123450-0 1518951123460-0

XRANGE還讓你可以借助COUNT選項(xiàng)，限制返回的數(shù)據(jù)項(xiàng)數(shù)量。比如說，下列查詢返回兩個(gè)間隔之間的前10個(gè)數(shù)據(jù)項(xiàng)。使用該選項(xiàng)，你可以像使用SCAN命令一樣迭代數(shù)據(jù)流：

XRANGE mystream 1518951123450-0 1518951123460-0 COUNT 10

如果你不知道查詢的上限或下限，可以將下限換成-、將上限換成+。比如說，下列查詢返回從數(shù)據(jù)開頭的前10個(gè)數(shù)據(jù)項(xiàng)：

XRANGE mystream - + COUNT 10

XREVRANGE的語法類似XRANGE，只是下限和上限的順序倒過來。比如說，下列查詢以相反的順序返回?cái)?shù)據(jù)流末尾的前10個(gè)數(shù)據(jù)項(xiàng)：

XREVRANGE mystream + - COUNT 10

5.在多個(gè)消費(fèi)者之間劃分?jǐn)?shù)據(jù)

場景：消費(fèi)者消費(fèi)數(shù)據(jù)的速度遠(yuǎn)低于生產(chǎn)者生成數(shù)據(jù)的速度。

在某些情況下，包括圖像處理、深度學(xué)習(xí)和情感分析，消費(fèi)者與生產(chǎn)者相比可能很慢。這種情況下，可以通過分散消費(fèi)者并劃分每個(gè)消費(fèi)者消耗的數(shù)據(jù)的做法，來匹配到達(dá)數(shù)據(jù)的速度與消耗數(shù)據(jù)的速度。

使用Redis Streams，你可以利用消費(fèi)者組來完成此任務(wù)。多個(gè)消費(fèi)者是消費(fèi)者組的一部分時(shí)，Redis Streams將確保每個(gè)消費(fèi)者都收到一組獨(dú)有的數(shù)據(jù)。

XREADGROUP GROUP mygroup consumer1 COUNT 2 STREAMS mystream >

當(dāng)然，關(guān)于消費(fèi)者組如何運(yùn)作還有更多的東西要了解。Redis Streams消費(fèi)者組旨在劃分?jǐn)?shù)據(jù)、實(shí)現(xiàn)災(zāi)難恢復(fù)并提供事務(wù)數(shù)據(jù)安全性。

如你所見，Redis Streams很容易上手。只需下載并安裝Redis 5.0，然后鉆研該項(xiàng)目網(wǎng)站上的Redis Streams教程。

原文標(biāo)題：How to use Redis Streams，作者：Roshan Kumar

作者簡介：Roshan Kumar是Redis Labs的資深產(chǎn)品經(jīng)理。他在軟件開發(fā)和技術(shù)營銷方面有著豐富的從業(yè)經(jīng)驗(yàn)。Roshan曾供職于惠普和多家成功的硅谷初創(chuàng)公司，包括ZillionTV、 Salorix、Alopa和ActiveVideo。

【51CTO譯稿，合作站點(diǎn)轉(zhuǎn)載請注明原文譯者和出處為51CTO.com】

責(zé)任編輯：龐桂玉來源： 51CTO