有一段時間沒有寫文章，techempower的測試規(guī)則評分竟然發(fā)生了變化，只能忘著補充一下占比權(quán)重最多的數(shù)據(jù)更新示例了和深入設計一下組件模塊化加載的設計。但在不久前有用戶問了一下組件是否支持redis的Stream功能，看了一樣相關(guān)資料后把功能實現(xiàn)之；接下來就介紹一下如何用Beetlex.Redis來調(diào)用redis的Stream功能。

什么是Stream

是Redis5.0的Stream是一個新的強大的支持多播的可持久化的消息隊列,它提供了消息添加，多組和多消費者一致性讀取和ack確認等功能；更詳細的介紹就不多說了可以通過網(wǎng)絡找到更多詳細描述。

創(chuàng)建Stream

組件通過RedisDB對象的GetStream訪求來創(chuàng)建一個Stream訪問對象，對象創(chuàng)建后就可以進行一系列的 XACK| XADD| XDEL| XGROUP| XLEN| XRANGE| XREAD| XREADGROUP| XREVRANGE|XTRIM等指令操作。創(chuàng)建代碼如下:

RedisStream<Employee> stream = DB.GetStream<Employee>("employees_stream");

XADD

在介紹這個操作前先說一下Stream里存儲的格式，默認Stream消息是K-V的格式，從基礎指令上可以了解到這種結(jié)構(gòu)

XADD mystream * sensor-id 1234 temperature 19.8

但這種格式操作起來并不友好，所以組件除了支持這種K-V的方式外，還支持以對象的方式進行Stream消息處理。接下來看一下插入對象的調(diào)用

RedisStream<Employee> stream = DB.GetStream<Employee>("employees_stream"); var id = await stream.Add(DataHelper.Defalut.Employees[0]); id = await stream.Add(DataHelper.Defalut.Employees[1]); id = await stream.Add(DataHelper.Defalut.Employees[2]); var len = await stream.Len();

組件支持直接入插對象，其基礎指令就是

XADD employees_stream * date employeejson

組件直接采用一個K-V的方式來存儲對象，對于原則多個K-V的方式組件同樣也支持，只是在構(gòu)建Stream指定類型用Dictionary<string,string>即可；接下其他就不多說了直接上指令用例了。

XLEN

RedisStream<Employee> stream = DB.GetStream<Employee>("employees_stream"); var len = await stream.Len();

XDEL

RedisStream<Employee> stream = DB.GetStream<Employee>("employees_stream"); var items = await stream.Read(null, null, "0-0"); await stream.Del((from item in items select item.ID).ToArray());

XRANGE

RedisStream<Employee> stream = DB.GetStream<Employee>("employees_stream"); var items = await stream.Range(); items = await stream.RangeAll();

XREVRANGE

RedisStream<Employee> stream = DB.GetStream<Employee>("employees_stream"); var items = await stream.RevRange(); items = await stream.RevRangeAll();

XREAD

RedisStream<Employee> stream = DB.GetStream<Employee>("employees_stream"); var items = await stream.Read(0, null, "0-0"); items = await stream.Read();

Stream的消費組

前面介紹的指令感覺列表結(jié)構(gòu)都能滿足，其實Stream重要的功能是在組消費這一塊，Redis可以針對Stream創(chuàng)建多個消費組和消費者，而消息會做一致性消費處理。

XGROUP

RedisStream<Employee> stream = DB.GetStream<Employee>("employees_stream"); var group = await stream.GetGroup("henry");

XREAD

RedisStream<Employee> stream = DB.GetStream<Employee>("employees_stream"); var group = await stream.GetGroup("g1"); var items = await group.Read("henry", "0");

實際XRead提供了是否等待和起始讀已取參數(shù)

public async ValueTask<List<StreamDataItem<T>>> ReadWait(string consumer,int timeout=0public ValueTask<List<StreamDataItem<T>>> Read(string consumer,string start = null) public async ValueTask<List<StreamDataItem<T>>> Read(string consumer, int? block, int? count, string start = null)

一般情況下可以通過readwait來不停地消息新的消息

while(true) {items = await group.ReadWait("henry");//處理消息foreach(var item in items){await item.Ack();} }

XACK

RedisStream<Employee> stream = DB.GetStream<Employee>("employees_stream"); var group = await stream.GetGroup("g1"); var items = await group.Read("henry", "0"); foreach (var item in items)await item.Ack();

以上是BeetleX.Redis組件提供操作Stream的基礎指令，實際上Stream還有一些和運維相關(guān)的指令，只是這些在實際業(yè)務上用不上所以就沒有去實現(xiàn)了。

總結(jié)

以上是生活随笔為你收集整理的Beetlex.Redis之Stream功能详解的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。