Stream في Redis

Redis Stream هي بنية بيانات جديدة تم إضافة إلى إصدار Redis 5.0.

يستخدم Redis Stream بشكل رئيسي كمسار لرسائل (MQ،Message Queue)،Redis نفسه يحتوي على نشر/اشتراك Redis (pub/sub) لتحقيق وظيفة مسار الرسائل، لكن له عيب هو أن الرسائل لا يمكن أن تكون مستدامة، إذا حدث انقطاع في الشبكة أو فشل Redis، سيتم فقد الرسائل.

بشكل بسيط، يمكن استخدام نشر/اشتراك (pub/sub) لنشر الرسائل، لكنه لا يمكنه تسجيل الرسائل التاريخية.

ويدعم Redis Stream ميزات التخزين الدائم للرسائل وتكرار النسخ الرئيسي والتابع، مما يسمح لأي عميل الوصول إلى البيانات في أي لحظة، ويتذكر موقف كل عميل في الوصول، ويضمن عدم فقدان الرسائل.

يكون بنية Redis Stream كما يلي، وهي تحتوي على قائمة رسائل، تربط جميع الرسائل المضافة، وكل رسالة تحتوي على ID فريد ومحتوى متعلق به:

كل Stream لديه اسم فريد، وهو مفتاح Redis، يتم إنشاؤه تلقائيًا عند استخدامنا لأول مرة الأمر xadd لإضافة رسائل.

التوضيح في الصورة أعلاه:

• مجموعة المستهلكين ال : مجموعة الاستهلاك، يتم إنشاؤها باستخدام الأمر XGROUP CREATE، ويكون لديها عدة مستهلكين (Consumer).

• last_delivered_id ال : علامة التتبع، لكل مجموعة استهلاك سيكون لديها علامة تتبع last_delivered_id، وسيتم تحريك علامة التتبع last_delivered_id للأمام عند قراءة أي مستهلك لرسالة.

• pending_ids ال متغير حالة المستهلك (Consumer) يساعد في الحفاظ على id غير المؤكد للمستهلك، حيث يُسجل pending_ids الرسائل التي تم قراءتها من قبل العميل ولكن لم يتم اعتمادها (Acknowledge character: رمز الاعتماد).

أوامر متعلقة بمجموعة الرسائل:

• XADD - إضافة الرسائل إلى نهاية الخط

• XTRIM - تقصير الخط، بحدود الطول

• XDEL - حذف الرسائل

• XLEN - الحصول على عدد العناصر الموجودة في الخط، أي طول الرسائل

• XRANGE - الحصول على قائمة الرسائل، سيتم تصفية الرسائل المحذوفة تلقائيًا

• XREVRANGE - الحصول على قائمة الرسائل بشكل معكوس، بترتيب الحجم من الأكبر إلى الأصغر

• XREAD - الحصول على قائمة الرسائل بطريقة مُحتجزة أو غير مُحتجزة

أوامر متعلقة بمجموعة المستهلكين:

• XGROUP CREATE - إنشاء مجموعة المستهلكين

• XREADGROUP GROUP - قراءة الرسائل من مجموعة المستهلكين

• XACK - وضع الرسائل في حالة "مُعالجة"

• XGROUP SETID - تعيين ID الرسالة الأخيرة المقدمة إلى مجموعة المستهلكين

• XGROUP DELCONSUMER - حذف المستهلك

• XGROUP DESTROY - حذف مجموعة المستهلكين

• XPENDING - عرض معلومات الرسائل القابلة للمعالجة

• XCLAIM - تحويل ملكية الرسائل

• XINFO - عرض معلومات الخط و مجموعة المستهلكين

• XINFO GROUPS - طباعة معلومات مجموعة المستهلكين

• XINFO STREAM - طباعة معلومات التدفق

XADD

استخدام XADD لإضافة رسائل إلى الخط، إذا لم يكن الخط المحدد موجودًا، فإنه يتم إنشاؤه، صيغة لغة XADD هي:

XADD　key　ID　field　value　[field　value　...]

• key ： اسم الصف، إذا لم يكن موجودًا، سيتم إنشاؤه.

• ID الرسالة id، نستخدم * لتعيين من قبل redis، يمكن تخصيصها، ولكن يجب أن تكون مضمونة التدرج.

• القيمة الحقل السجلات.

redis> XADD mystream * name Sara surname OConnor
"1601372323627-0"
redis> XADD mystream * field1 value1 field2 value2 field3 value3
"1601372323627-1"
redis>　XLEN　mystream
(integer) 2
redis> XRANGE mystream - +
1) 1) "1601372323627-0"
　　　2) 1) "name"
　　　　　　2) "Sara"
　　　　　　3) "surname"
　　　　　　4) "OConnor"
2) 1) "1601372323627-1"
　　　2) 1) "field1"
　　　　　　2) "value1"
　　　　　　3) "field2"
　　　　　　4) "value2"
　　　　　　5) "field3"
　　　　　　6) "value3"
redis>

XTRIM

استخدام XTRIM لتقليم السلسلة، محددة الطول، نموذج الجملة:

XTRIM key MAXLEN [~] count

• key ：اسم الورقة

• MAXLEN الطول

• count ：数量

127.0.0.1:6379> XADD mystream * field1 A field2 B field3 C field4 D
"1601372434568-0"
127.0.0.1:6379> XTRIM mystream MAXLEN 2
(integer) 0
127.0.0.1:6379> XRANGE mystream - +
1) 1) "1601372434568-0"
　　　2) 1) "field1"
　　　　　　2) "A"
　　　　　　3) "field2"
　　　　　　4) "B"
　　　　　　5) "field3"
　　　　　　6) "C"
　　　　　　7) "field4"
　　　　　　8) "D"
127.0.0.1:6379>　
redis>

XDEL

استخدام XDEL لحذف الرسائل، نموذج الجملة：

XDEL key ID [ID ...]

• key：اسم الورقة

• ID ：消息 ID

> XADD mystream * a 1
1538561698944-0
> XADD mystream * b 2
1538561700640-0
> XADD mystream * c 3
1538561701744-0
> XDEL mystream 1538561700640-0
(integer) 1
127.0.0.1:6379> XRANGE mystream - +
1)　1)　1538561698944-0
　　　2)　1)　"a"
　　　　　　2)　"1"
2)　1)　1538561701744-0
　　　2)　1)　"c"
　　　　　　2)　"3"

XLEN

استخدام XLEN للحصول على عدد العناصر في السلسلة، أي طول الرسائل، نموذج الجملة:

XLEN　key

• key：اسم الورقة

redis>　XADD　mystream　*　item　1
"1601372563177-0"
redis>　XADD　mystream　*　item　2
"1601372563178-0"
redis>　XADD　mystream　*　item　3
"1601372563178-1"
redis>　XLEN　mystream
(integer)　3
redis>

XRANGE

استخدام XRANGE للحصول على قائمة الرسائل، سيتم تصفية الرسائل المحذوفة تلقائيًا، نموذج الجملة:

XRANGE　key　start　end　[COUNT　count]

• key ：队列名

• البداية ：القيمة البداية، - تشير إلى الأقل

• النهاية ：القيمة النهائية، + تشير إلى الأكبر

• count ：数量

redis>　XADD　writers　*　name　Virginia　surname　Woolf
"1601372577811-0"
redis>　XADD　writers　*　name　Jane　surname　Austen
"1601372577811-1"
redis>　XADD　writers　*　name　Toni　surname　Morrison
"1601372577811-2"
redis>　XADD　writers　*　name　Agatha　surname　Christie
"1601372577812-0"
redis> XADD writers * name Ngozi surname Adichie
"1601372577812-1"
redis> XLEN writers
(integer) 5
redis>　XRANGE　writers　-　+　COUNT　2
1)　1)　"1601372577811-0"
　　　2) 1) "name"
　　　　　　2) "Virginia"
　　　　　　3) "surname"
　　　　　　4) "Woolf"
2)　1)　"1601372577811-1"
　　　2) 1) "name"
　　　　　　2) "Jane"
　　　　　　3) "surname"
　　　　　　4) "Austen"
redis>

XREVRANGE

استخدام XREVRANGE للحصول على قائمة الرسائل، سيتم تصفية الرسائل المحذوفة تلقائيًا، نموذج الجملة:

XREVRANGE　key　end　start　[COUNT　count]

• key ：队列名

• النهاية ：القيمة النهائية، + تشير إلى الأكبر

• البداية ：القيمة البداية، - تشير إلى الأقل

• count ：数量

redis>　XADD　writers　*　name　Virginia　surname　Woolf
"1601372731458-0"
redis>　XADD　writers　*　name　Jane　surname　Austen
"1601372731459-0"
redis>　XADD　writers　*　name　Toni　surname　Morrison
"1601372731459-1"
redis>　XADD　writers　*　name　Agatha　surname　Christie
"1601372731459-2"
redis> XADD writers * name Ngozi surname Adichie
"1601372731459-3"
redis> XLEN writers
(integer) 5
redis> XREVRANGE writers + - COUNT 1
1) 1) "1601372731459-3"
　　　2) 1) "name"
　　　　　　2) "Ngozi"
　　　　　　3) "surname"
　　　　　　4) "Adichie"
redis>

XREAD

使用 XREAD 以阻塞或非阻塞方式获取消息列表 ，语法格式：

XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key [...]] id [id ...]

• count ：数量

• milliseconds ：可选，阻塞毫秒数，没有设置就是非阻塞模式

• key ：队列名

• id ：消息 ID

# 从 Stream 头部读取两条消息
> XREAD COUNT 2 STREAMS mystream writers 0-0 0-0
1) 1) "mystream"
　　　2) 1) 1) 1526984818136-0
　　　　　　　　　2) 1) "duration"
　　　　　　　　　　　　2) "1532"
　　　　　　　　　　　　3) "event-id"
　　　　　　　　　　　　4) "5"
　　　　　　　　　　　　5) "user-id"
　　　　　　　　　　　　6) "7782813"
　　　　　　2) 1) 1526999352406-0
　　　　　　　　　2) 1) "duration"
　　　　　　　　　　　　2) "812"
　　　　　　　　　　　　3) "event-id"
　　　　　　　　　　　　4) "9"
　　　　　　　　　　　　5) "user-id"
　　　　　　　　　　　　6) "388234"
2) 1) "writers"
　　　2) 1) 1) 1526985676425-0
　　　　　　　　　2) 1) "name"
　　　　　　　　　　　　2) "Virginia"
　　　　　　　　　　　　3) "surname"
　　　　　　　　　　　　4) "Woolf"
　　　　　　2) 1) 1526985685298-0
　　　　　　　　　2) 1) "name"
　　　　　　　　　　　　2) "Jane"
　　　　　　　　　　　　3) "surname"
　　　　　　　　　　　　4) "Austen"

XGROUP CREATE

使用 XGROUP CREATE 创建消费者组，语法格式：

XGROUP [CREATE key groupname id-or-$] [SETID key groupname id-or-$] [DESTROY key groupname] [DELCONSUMER key groupname consumername]

• key ： اسم الصف، إذا لم يكن موجودًا، سيتم إنشاؤه.

• groupname ： اسم المجموعة.

• $ ： يعني البدء في الاستهلاك من النهاية، فقط تقبل الرسائل الجديدة، وسيتم تجاهل جميع رسائل Stream الحالية.

البدء في الاستهلاك من البداية:

XGROUP CREATE mystream consumer-group-name 0-0

البدء في الاستهلاك من النهاية:

XGROUP CREATE mystream consumer-group-name $

XREADGROUP GROUP

استخدام XREADGROUP GROUP لقراءة الرسائل من المجموعة الاستهلاكية، صيغة النص:

XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] ID [ID ...]

• group ： اسم المجموعة.

• consumer ： اسم المستهلك.

• count ： عدد القراءة.

• milliseconds ： عدد الميليسي ثانية.

• key ： اسم الصف.

• ID ： معرف الرسالة.

XREADGROUP GROUP consumer-group-name consumer-name COUNT 1 STREAMS mystream >