Redis Stream הוא סוג נתונים מופשט חדש שהוצג ב-Redis עם גרסה 5.0
מבחינה קונספטואלית, Redis Stream היא רשימה שאליה אתה יכול להוסיף ערכים. לכל ערך יש מזהה ייחודי. כברירת מחדל, המזהה נוצר אוטומטית וכולל חותמת זמן. לכן, אתה יכול לבצע שאילתות בטווחי רשומות לאורך זמן, או לקבל נתונים חדשים כשהם מגיעים לזרם, בדומה לפקודת Unix "tail -f" קוראת קובץ יומן וקופא בזמן ההמתנה לנתונים חדשים. שים לב שלקוחות מרובים יכולים להאזין לשרשור בו-זמנית, בדיוק כפי שתהליכי "tail -f" רבים יכולים לקרוא קובץ בו-זמנית מבלי להתנגש זה עם זה.

כדי להבין את כל היתרונות של סוג הנתונים החדש, הבה נסתכל במהירות על מבני Redis הקיימים ותיקים המשכפלים חלקית את הפונקציונליות של Redis Stream.

Redis PUB/SUB

Redis Pub/Sub היא מערכת הודעות פשוטה שכבר מובנית בחנות המפתח שלך. עם זאת, לפשטות יש מחיר:

• אם המוציא לאור מסיבה כלשהי נכשל, אז הוא מאבד את כל המנויים שלו
• המוציא לאור צריך לדעת את הכתובת המדויקת של כל המנויים שלו
• מפרסם עלול להעמיס על המנויים שלו בעבודה אם הנתונים מתפרסמים מהר יותר ממה שהם מעובדים
• ההודעה נמחקת מהמאגר של המפרסם מיד לאחר הפרסום, ללא קשר לכמה מנויים היא נמסרה וכמה מהר הם הצליחו לעבד את ההודעה הזו.
• כל המנויים יקבלו את ההודעה בו זמנית. המנויים עצמם חייבים איכשהו להסכים ביניהם על סדר העיבוד של אותה הודעה.
• אין מנגנון מובנה לאשר שמנוי עיבד הודעה בהצלחה. אם מנוי מקבל הודעה וקורס במהלך העיבוד, המפרסם לא יידע על כך.

רשימת Redis

Redis List הוא מבנה נתונים התומך בחסימת פקודות קריאה. אתה יכול להוסיף ולקרוא הודעות מתחילת או סוף הרשימה. בהתבסס על המבנה הזה, אתה יכול ליצור מחסנית או תור טוב למערכת המבוזרת שלך, וברוב המקרים זה יספיק. ההבדלים העיקריים מ-Redis Pub/Sub:

• ההודעה מועברת ללקוח אחד. הלקוח הראשון שנחסם לקריאה יקבל את הנתונים תחילה.
• קלינט חייב ליזום את פעולת הקריאה עבור כל הודעה בעצמו. רשימה לא יודעת דבר על לקוחות.
• הודעות מאוחסנות עד שמישהו קורא אותן או מוחק אותן במפורש. אם תגדיר את שרת Redis לשטוף נתונים לדיסק, אזי האמינות של המערכת עולה באופן דרמטי.

מבוא לסטרימינג

הוספת ערך לזרם

קבוצה XADD מוסיף ערך חדש לזרם. רשומה היא לא רק מחרוזת, היא מורכבת מזוג מפתח-ערך אחד או יותר. לפיכך, כל ערך כבר מובנה ודומה למבנה של קובץ CSV.

> XADD mystream * sensor-id 1234 temperature 19.8
1518951480106-0

בדוגמה שלמעלה, אנו מוסיפים שני שדות לזרם עם השם (מפתח) "mystream": "חיישן מזהה" ו-"טמפרטורה" עם הערכים "1234" ו-"19.8", בהתאמה. בתור הארגומנט השני, הפקודה לוקחת מזהה שיוקצה לערך - מזהה זה מזהה באופן ייחודי כל ערך בזרם. עם זאת, במקרה הזה עברנו * כי אנחנו רוצים ש-Redis יפיק לנו מזהה חדש. כל תעודת זהות חדשה תגדל. לכן, לכל ערך חדש יהיה מזהה גבוה יותר ביחס לערכים קודמים.

פורמט מזהה

מזהה הכניסה שהוחזר על ידי הפקודה XADD, מורכב משני חלקים:

{millisecondsTime}-{sequenceNumber}

מילי-שניות זמן - זמן יוניקס באלפיות שניות (זמן שרת Redis). עם זאת, אם השעה הנוכחית זהה או קטנה מהזמן של ההקלטה הקודמת, אזי נעשה שימוש בחותמת הזמן של ההקלטה הקודמת. לכן, אם זמן השרת חוזר אחורה בזמן, המזהה החדש עדיין ישמור על מאפיין התוספת.

רצף מספר משמש עבור רשומות שנוצרו באותה אלפיות השנייה. רצף מספר יוגדל ב-1 ביחס לערך הקודם. בגלל ה רצף מספר הוא בגודל 64 סיביות, אז בפועל לא צריך להיתקל במגבלה על מספר הרשומות שניתן להפיק תוך אלפית שנייה אחת.

הפורמט של מזהים כאלה עשוי להיראות מוזר במבט ראשון. קורא חסר אמון עשוי לתהות מדוע הזמן הוא חלק מהמזהה. הסיבה היא שזרמי Redis תומכים בשאילתות טווח לפי מזהה. מכיוון שהמזהה משויך לזמן שבו נוצרה הרשומה, הדבר מאפשר לבצע שאילתות בטווחי זמן. נסתכל על דוגמה ספציפית כאשר נסתכל על הפקודה XRANGE.

אם מסיבה כלשהי המשתמש צריך לציין מזהה משלו, אשר, למשל, משויך למערכת חיצונית כלשהי, אז נוכל להעביר אותו לפקודה XADD במקום * כפי שמוצג להלן:

> XADD somestream 0-1 field value
0-1
> XADD somestream 0-2 foo bar
0-2

שים לב שבמקרה זה עליך לעקוב בעצמך אחר תוספת המזהה. בדוגמה שלנו, המזהה המינימלי הוא "0-1", כך שהפקודה לא תקבל מזהה אחר השווה או קטן מ-"0-1".

> XADD somestream 0-1 foo bar
(error) ERR The ID specified in XADD is equal or smaller than the target stream top item

מספר רשומות לכל זרם

אפשר לקבל את מספר הרשומות בזרם פשוט על ידי שימוש בפקודה XLEN. לדוגמא שלנו, פקודה זו תחזיר את הערך הבא:

> XLEN somestream
(integer) 2

שאילתות טווח - XRANGE ו-XREVRANGE

כדי לבקש נתונים לפי טווח, עלינו לציין שני מזהים - ההתחלה והסוף של הטווח. הטווח המוחזר יכלול את כל האלמנטים, כולל הגבולות. ישנם גם שני מזהים מיוחדים "-" ו-"+", כלומר המזהה הקטן ביותר (רשומה ראשונה) והגדול ביותר (רשומה אחרונה) בזרם. הדוגמה למטה תפרט את כל ערכי הזרם.

> XRANGE mystream - +
1) 1) 1518951480106-0
   2) 1) "sensor-id"
      2) "1234"
      3) "temperature"
      4) "19.8"
2) 1) 1518951482479-0
   2) 1) "sensor-id"
      2) "9999"
      3) "temperature"
      4) "18.2"

כל רשומה שהוחזרה היא מערך של שני אלמנטים: מזהה ורשימת צמדי מפתח-ערך. כבר אמרנו שמזהי רשומות קשורים לזמן. לכן, אנו יכולים לבקש טווח של פרק זמן מסוים. עם זאת, אנו יכולים לציין בבקשה לא את המזהה המלא, אלא רק את זמן יוניקס, תוך השמטת החלק הקשור ל רצף מספר. החלק המושמט של המזהה יוגדר אוטומטית לאפס בתחילת הטווח ולערך המקסימלי האפשרי בסוף הטווח. להלן דוגמה כיצד ניתן לבקש טווח של שתי אלפיות שניות.

> XRANGE mystream 1518951480106 1518951480107
1) 1) 1518951480106-0
   2) 1) "sensor-id"
      2) "1234"
      3) "temperature"
      4) "19.8"

יש לנו רק ערך אחד בטווח הזה, אולם במערכות נתונים אמיתיות התוצאה המוחזרת יכולה להיות עצומה. מהסיבה הזו XRANGE תומך באפשרות COUNT. על ידי ציון הכמות, נוכל פשוט לקבל את N הרשומות הראשונות. אם אנחנו צריכים לקבל את ה-N הרשומות הבאות (דפיפון), נוכל להשתמש במזהה האחרון שהתקבל, להגדיל אותו רצף מספר באחד ושאל שוב. בואו נסתכל על זה בדוגמה הבאה. אנחנו מתחילים להוסיף 10 אלמנטים עם XADD (בהנחה שה-mystream כבר היה מלא ב-10 אלמנטים). כדי להתחיל באיטרציה לקבל 2 אלמנטים לכל פקודה, אנחנו מתחילים עם הטווח המלא אבל עם COUNT שווה ל-2.

> XRANGE mystream - + COUNT 2
1) 1) 1519073278252-0
   2) 1) "foo"
      2) "value_1"
2) 1) 1519073279157-0
   2) 1) "foo"
      2) "value_2"

כדי להמשיך לחזור על שני האלמנטים הבאים, עלינו לבחור את המזהה האחרון שהתקבל, כלומר 1519073279157-0, ולהוסיף 1 ל רצף מספר.
המזהה המתקבל, במקרה זה 1519073279157-1, יכול לשמש כעת כארגומנט תחילת הטווח החדש עבור הקריאה הבאה XRANGE:

> XRANGE mystream 1519073279157-1 + COUNT 2
1) 1) 1519073280281-0
   2) 1) "foo"
      2) "value_3"
2) 1) 1519073281432-0
   2) 1) "foo"
      2) "value_4"

וכולי. כי מורכבות XRANGE הוא O(log(N)) לחיפוש ולאחר מכן O(M) להחזרת M אלמנטים, ואז כל שלב איטרציה מהיר. לפיכך, באמצעות XRANGE ניתן לבצע איטרציה יעילה של זרמים.

קבוצה XREVRANGE הוא המקביל XRANGE, אבל מחזיר את הרכיבים בסדר הפוך:

> XREVRANGE mystream + - COUNT 1
1) 1) 1519073287312-0
   2) 1) "foo"
      2) "value_10"

שימו לב שהפקודה XREVRANGE לוקח ארגומנטים של טווח להתחיל ולעצור בסדר הפוך.

קריאת ערכים חדשים באמצעות XREAD

לעתים קרובות עולה המשימה של הרשמה לזרם וקבלת הודעות חדשות בלבד. הרעיון הזה אולי נראה דומה ל-Redis Pub/Sub או לחסימת Redis List, אבל יש הבדלים מהותיים בשימוש ב-Redis Stream:

1. כל הודעה חדשה נמסרת לכל מנוי כברירת מחדל. התנהגות זו שונה מרשימת Redis חסימת, שבה הודעה חדשה תיקרא רק על ידי מנוי אחד.
2. בעוד שב-Redis Pub/Sub כל ההודעות נשכחות ולעולם לא נמשכות, ב-Stream כל ההודעות נשמרות ללא הגבלת זמן (אלא אם הלקוח גורם במפורש למחיקה).
3. Redis Stream מאפשר לך להבדיל בין גישה להודעות בתוך זרם אחד. מנוי ספציפי יכול לראות רק את היסטוריית ההודעות האישית שלו.

אתה יכול להירשם לשרשור ולקבל הודעות חדשות באמצעות הפקודה XREAD. זה קצת יותר מסובך מאשר XRANGE, אז נתחיל תחילה עם הדוגמאות הפשוטות יותר.

> XREAD COUNT 2 STREAMS mystream 0
1) 1) "mystream"
   2) 1) 1) 1519073278252-0
         2) 1) "foo"
            2) "value_1"
      2) 1) 1519073279157-0
         2) 1) "foo"
            2) "value_2"

הדוגמה שלמעלה מציגה טופס לא חוסם XREAD. שים לב שהאפשרות COUNT היא אופציונלית. למעשה, אפשרות הפקודה היחידה הנדרשת היא אפשרות STREAMS, שמציינת רשימה של זרמים יחד עם המזהה המקסימלי המתאים. כתבנו "STREAMS mystream 0" - אנחנו רוצים לקבל את כל הרשומות של זרם mystream עם מזהה גדול מ-"0-0". כפי שניתן לראות מהדוגמה, הפקודה מחזירה את שם השרשור מכיוון שאנו יכולים להירשם למספר שרשורים בו זמנית. נוכל לכתוב, למשל, "STREAMS mystream otherstream 0 0". שימו לב שאחרי אפשרות STREAMS עלינו לספק תחילה את שמות כל הזרמים הנדרשים ורק לאחר מכן רשימת מזהים.

בצורה פשוטה זו הפקודה לא עושה שום דבר מיוחד בהשוואה XRANGE. עם זאת, הדבר המעניין הוא שאנחנו יכולים בקלות להפוך XREAD לפקודת חסימה, תוך ציון הארגומנט BLOCK:

> XREAD BLOCK 0 STREAMS mystream $

בדוגמה שלמעלה, אפשרות BLOCK חדשה מצוינת עם פסק זמן של 0 אלפיות שניות (משמעות הדבר היא המתנה ללא הגבלת זמן). יתרה מכך, במקום להעביר את המזהה הרגיל ל-stream mystream, הועבר מזהה מיוחד $. מזהה מיוחד זה אומר זאת XREAD חייב להשתמש במזהה המקסימלי ב-mystream כמזהה. כך שנקבל רק הודעות חדשות החל מהרגע שהתחלנו להאזין. במובנים מסוימים זה דומה לפקודת Unix "tail -f".

שימו לב שבשימוש באפשרות BLOCK אנחנו לא בהכרח צריכים להשתמש במזהה המיוחד $. אנחנו יכולים להשתמש בכל מזהה שקיים בזרם. אם הצוות יכול לטפל בבקשה שלנו באופן מיידי בלי לחסום, הוא יעשה זאת, אחרת הוא יחסום.

חסימה XREAD יכול גם להאזין למספר שרשורים בבת אחת, אתה רק צריך לציין את שמותיהם. במקרה זה, הפקודה תחזיר רשומה של הזרם הראשון שקיבל נתונים. המנוי הראשון שנחסם עבור שרשור נתון יקבל נתונים תחילה.

קבוצות צרכנות

במשימות מסוימות, אנו רוצים להגביל את גישת המנוי להודעות בתוך שרשור אחד. דוגמה שבה זה יכול להיות שימושי היא תור הודעות עם עובדים שיקבלו הודעות שונות משרשור, מה שיאפשר עיבוד הודעות להגדלה.

אם נדמיין שיש לנו שלושה מנויים C1, C2, C3 ושרשור המכיל הודעות 1, 2, 3, 4, 5, 6, 7, אז ההודעות יוגשו כמו בתרשים שלהלן:

1 -> C1
2 -> C2
3 -> C3
4 -> C1
5 -> C2
6 -> C3
7 -> C1

כדי להשיג את האפקט הזה, Redis Stream משתמש במושג שנקרא Consumer Group. מושג זה דומה למנוי פסאודו, שמקבל נתונים מזרם, אך למעשה מוגש על ידי מספר מנויים בתוך קבוצה, תוך מתן ערבויות מסוימות:

1. כל הודעה נמסרת למנוי אחר בתוך הקבוצה.
2. בתוך קבוצה, מנויים מזוהים בשמם, שהוא מחרוזת תלוית רישיות. אם מנוי נושר זמנית מהקבוצה, ניתן להחזיר אותו לקבוצה באמצעות השם הייחודי שלו.
3. כל קבוצת צרכנים פועלת לפי המושג "הודעה ראשונה שלא נקראה". כאשר מנוי מבקש הודעות חדשות, הוא יכול לקבל רק הודעות שמעולם לא נמסרו בעבר למנוי כלשהו בקבוצה.
4. ישנה פקודה לאשר במפורש שההודעה עובדה בהצלחה על ידי המנוי. עד שתקרא פקודה זו, ההודעה המבוקשת תישאר במצב "בהמתנה".
5. בתוך קבוצת הצרכנים, כל מנוי יכול לבקש היסטוריית הודעות שנמסרו לו, אך טרם טופלו (בסטטוס "בהמתנה")

במובן מסוים, מצב הקבוצה יכול לבוא לידי ביטוי באופן הבא:

+----------------------------------------+
| consumer_group_name: mygroup          
| consumer_group_stream: somekey        
| last_delivered_id: 1292309234234-92    
|                                                           
| consumers:                                          
|    "consumer-1" with pending messages  
|       1292309234234-4                          
|       1292309234232-8                          
|    "consumer-42" with pending messages 
|       ... (and so forth)                             
+----------------------------------------+

עכשיו הגיע הזמן להכיר את הפקודות העיקריות של קבוצת הצרכנים, כלומר:

• XGROUP משמש ליצירה, הרס וניהול קבוצות
• XREADGROUP משמש לקריאת זרם דרך הקבוצה
• XACK - פקודה זו מאפשרת למנוי לסמן את ההודעה כעובדה בהצלחה

הקמת קבוצת צרכנים

בואו נניח שה-mystream כבר קיים. ואז פקודת יצירת הקבוצה תיראה כך:

> XGROUP CREATE mystream mygroup $
OK

בעת יצירת קבוצה, עלינו להעביר מזהה, שמתחיל ממנו הקבוצה תקבל הודעות. אם אנחנו רק רוצים לקבל את כל ההודעות החדשות, אז נוכל להשתמש במזהה המיוחד $ (כמו בדוגמה שלנו למעלה). אם תציין 0 במקום מזהה מיוחד, אז כל ההודעות בשרשור יהיו זמינות לקבוצה.

כעת לאחר יצירת הקבוצה, אנו יכולים להתחיל מיד לקרוא הודעות באמצעות הפקודה XREADGROUP. פקודה זו דומה מאוד ל XREAD ותומך באפשרות ה-BLOCK האופציונלית. עם זאת, ישנה אפשרות GROUP נדרשת שתמיד יש לציין עם שני ארגומנטים: שם הקבוצה ושם המנוי. גם האפשרות COUNT נתמכת.

לפני קריאת השרשור, בוא נכניס שם כמה הודעות:

> XADD mystream * message apple
1526569495631-0
> XADD mystream * message orange
1526569498055-0
> XADD mystream * message strawberry
1526569506935-0
> XADD mystream * message apricot
1526569535168-0
> XADD mystream * message banana
1526569544280-0

עכשיו בואו ננסה לקרוא את הזרם הזה דרך הקבוצה:

> XREADGROUP GROUP mygroup Alice COUNT 1 STREAMS mystream >
1) 1) "mystream"
   2) 1) 1) 1526569495631-0
         2) 1) "message"
            2) "apple"

הפקודה שלעיל קוראת מילה במילה כדלקמן:

"אני, המנוי אליס, חברה ב-mygroup, רוצה לקרוא הודעה אחת מ-mystream שמעולם לא נמסרה לאף אחד לפני כן."

בכל פעם שמנוי מבצע פעולה בקבוצה, עליו לספק את שמה, ולזהות את עצמו באופן ייחודי בתוך הקבוצה. יש עוד פרט אחד חשוב מאוד בפקודה לעיל - המזהה המיוחד ">". המזהה המיוחד הזה מסנן הודעות ומשאיר רק את אלה שמעולם לא נמסרו קודם לכן.

כמו כן, במקרים מיוחדים, ניתן לציין מזהה אמיתי כגון 0 או כל מזהה חוקי אחר. במקרה זה הפקודה XREADGROUP יחזיר לך היסטוריה של הודעות בסטטוס "בהמתנה" שנמסרו למנוי שצוין (אליס) אך עדיין לא קיבלו אישור באמצעות הפקודה XACK.

אנו יכולים לבדוק התנהגות זו על ידי ציון מיידי של המזהה 0, ללא אפשרות COUNT. אנו פשוט נראה הודעה אחת ממתינה, כלומר, הודעת Apple:

> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
   2) 1) 1) 1526569495631-0
         2) 1) "message"
            2) "apple"

עם זאת, אם נאשר שההודעה עובדה בהצלחה, היא לא תוצג עוד:

> XACK mystream mygroup 1526569495631-0
(integer) 1
> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
   2) (empty list or set)

עכשיו תורו של בוב לקרוא משהו:

> XREADGROUP GROUP mygroup Bob COUNT 2 STREAMS mystream >
1) 1) "mystream"
   2) 1) 1) 1526569498055-0
         2) 1) "message"
            2) "orange"
      2) 1) 1526569506935-0
         2) 1) "message"
            2) "strawberry"

בוב, חבר ב-mygroup, ביקש לא יותר משתי הודעות. הפקודה מדווחת רק על הודעות שלא נמסרו עקב המזהה המיוחד ">". כפי שאתה יכול לראות, ההודעה "תפוח" לא תוצג מכיוון שהיא כבר נמסרה לאליס, אז בוב מקבל "תפוז" ו"תות".

בדרך זו, אליס, בוב וכל מנוי אחר לקבוצה יכולים לקרוא הודעות שונות מאותו זרם. הם יכולים גם לקרוא את ההיסטוריה של הודעות לא מעובדות או לסמן הודעות כעובדות.

יש לזכור כמה דברים:

• ברגע שהמנוי מחשיב את ההודעה כפקודה XREADGROUP, הודעה זו עוברת למצב "בהמתנה" והיא מוקצית לאותו מנוי ספציפי. מנויים אחרים בקבוצה לא יוכלו לקרוא הודעה זו.
• מנויים נוצרים אוטומטית עם אזכור ראשון, אין צורך ליצור אותם במפורש.
• עם XREADGROUP אתה יכול לקרוא הודעות ממספר שרשורים שונים בו-זמנית, אולם כדי שזה יעבוד עליך ליצור תחילה קבוצות עם אותו שם עבור כל שרשור באמצעות XGROUP

התאוששות לאחר כישלון

המנוי יכול להתאושש מהכשל ולקרוא מחדש את רשימת ההודעות שלו עם הסטטוס "בהמתנה". עם זאת, בעולם האמיתי, מנויים עלולים בסופו של דבר להיכשל. מה קורה להודעות התקועות של מנוי אם המנוי לא מצליח להתאושש מתקלה?
Consumer Group מציעה תכונה המשמשת בדיוק למקרים כאלה - כאשר צריך לשנות את הבעלים של ההודעות.

הדבר הראשון שאתה צריך לעשות הוא לקרוא לפקודה XPENDING, המציג את כל ההודעות בקבוצה עם הסטטוס "בהמתנה". בצורתה הפשוטה ביותר, הפקודה נקראת עם שני ארגומנטים בלבד: שם השרשור ושם הקבוצה:

> XPENDING mystream mygroup
1) (integer) 2
2) 1526569498055-0
3) 1526569506935-0
4) 1) 1) "Bob"
      2) "2"

הצוות הציג את מספר ההודעות הלא מעובדות עבור כל הקבוצה ועבור כל מנוי. יש לנו רק לבוב עם שתי הודעות יוצאות דופן כי ההודעה היחידה שאליס ביקשה אושרה XACK.

אנו יכולים לבקש מידע נוסף באמצעות טיעונים נוספים:

XPENDING {key} {groupname} [{start-id} {end-id} {count} [{consumer-name}]]
{start-id} {end-id} - טווח מזהים (תוכל להשתמש ב-"-" ו-"+")
{count} - מספר ניסיונות המסירה
{consumer-name} - שם הקבוצה

> XPENDING mystream mygroup - + 10
1) 1) 1526569498055-0
   2) "Bob"
   3) (integer) 74170458
   4) (integer) 1
2) 1) 1526569506935-0
   2) "Bob"
   3) (integer) 74170458
   4) (integer) 1

כעת יש לנו פרטים עבור כל הודעה: מזהה, שם מנוי, זמן סרק באלפיות שניות ולבסוף מספר ניסיונות המסירה. יש לנו שתי הודעות מבוב והן לא היו פעילות במשך 74170458 אלפיות השנייה, בערך 20 שעות.

שימו לב שאף אחד לא מונע מאיתנו לבדוק מה היה תוכן ההודעה פשוט על ידי שימוש XRANGE.

> XRANGE mystream 1526569498055-0 1526569498055-0
1) 1) 1526569498055-0
   2) 1) "message"
      2) "orange"

אנחנו רק צריכים לחזור על אותו מזהה פעמיים בטיעונים. עכשיו, כשיש לנו רעיון מסוים, אליס עשויה להחליט שאחרי 20 שעות של השבתה, בוב כנראה לא יתאושש, וזה הזמן לבצע שאילתות בהודעות האלה ולחדש את עיבודן עבור בוב. לשם כך אנו משתמשים בפקודה XCLAIM:

XCLAIM {key} {group} {consumer} {min-idle-time} {ID-1} {ID-2} ... {ID-N}

באמצעות פקודה זו, נוכל לקבל הודעה "זרה" שעדיין לא עובדה על ידי שינוי הבעלים ל-{consumer}. עם זאת, אנו יכולים גם לספק זמן סרק מינימלי {min-idle-time}. זה עוזר למנוע מצב שבו שני לקוחות מנסים לשנות בו זמנית את הבעלים של אותן הודעות:

Client 1: XCLAIM mystream mygroup Alice 3600000 1526569498055-0
Clinet 2: XCLAIM mystream mygroup Lora 3600000 1526569498055-0

הלקוח הראשון יאפס את זמן ההשבתה ויגדיל את מונה המשלוחים. אז הלקוח השני לא יוכל לבקש זאת.

> XCLAIM mystream mygroup Alice 3600000 1526569498055-0
1) 1) 1526569498055-0
   2) 1) "message"
      2) "orange"

ההודעה נתבעה בהצלחה על ידי אליס, שיכולה כעת לעבד את ההודעה ולאשר אותה.

מהדוגמה לעיל, ניתן לראות שבקשה מוצלחת מחזירה את תוכן ההודעה עצמה. עם זאת, אין צורך בכך. ניתן להשתמש באפשרות JUSTID להחזרת מזהי הודעות בלבד. זה שימושי אם אינך מעוניין בפרטי ההודעה ומעוניין להגביר את ביצועי המערכת.

דלפק משלוחים

המונה שאתה רואה בפלט XPENDING הוא מספר המסירות של כל הודעה. מונה כזה מוגדל בשתי דרכים: כאשר הודעה מתבקשת בהצלחה באמצעות XCLAIM או כאשר נעשה שימוש בשיחה XREADGROUP.

זה נורמלי שהודעות מסוימות מועברות מספר פעמים. העיקר שכל ההודעות יעובדו בסופו של דבר. לפעמים מתרחשות בעיות בעת עיבוד הודעה בגלל שההודעה עצמה פגומה, או שעיבוד ההודעות גורם לשגיאה בקוד המטפל. במקרה זה, עלול להתברר שאף אחד לא יוכל לעבד את ההודעה הזו. מכיוון שיש לנו מונה ניסיונות מסירה, אנו יכולים להשתמש במונה זה כדי לזהות מצבים כאלה. לכן, ברגע שספירת המסירה מגיעה למספר הגבוה שציינת, כנראה שיהיה חכם יותר לשים הודעה כזו בשרשור אחר ולשלוח הודעה למנהל המערכת.

מצב חוט

קבוצה XINFO משמש לבקשת מידע שונה על שרשור וקבוצותיו. לדוגמה, פקודה בסיסית נראית כך:

> XINFO STREAM mystream
 1) length
 2) (integer) 13
 3) radix-tree-keys
 4) (integer) 1
 5) radix-tree-nodes
 6) (integer) 2
 7) groups
 8) (integer) 2
 9) first-entry
10) 1) 1524494395530-0
    2) 1) "a"
       2) "1"
       3) "b"
       4) "2"
11) last-entry
12) 1) 1526569544280-0
    2) 1) "message"
       2) "banana"

הפקודה למעלה מציגה מידע כללי על הזרם שצוין. עכשיו דוגמה קצת יותר מורכבת:

> XINFO GROUPS mystream
1) 1) name
   2) "mygroup"
   3) consumers
   4) (integer) 2
   5) pending
   6) (integer) 2
2) 1) name
   2) "some-other-group"
   3) consumers
   4) (integer) 1
   5) pending
   6) (integer) 0

הפקודה למעלה מציגה מידע כללי עבור כל הקבוצות של השרשור שצוין

> XINFO CONSUMERS mystream mygroup
1) 1) name
   2) "Alice"
   3) pending
   4) (integer) 1
   5) idle
   6) (integer) 9104628
2) 1) name
   2) "Bob"
   3) pending
   4) (integer) 1
   5) idle
   6) (integer) 83841983

הפקודה למעלה מציגה מידע עבור כל המנויים של הזרם והקבוצה שצוינו.
אם שכחת את תחביר הפקודה, פשוט בקש עזרה מהפקודה עצמה:

> XINFO HELP
1) XINFO {subcommand} arg arg ... arg. Subcommands are:
2) CONSUMERS {key} {groupname}  -- Show consumer groups of group {groupname}.
3) GROUPS {key}                 -- Show the stream consumer groups.
4) STREAM {key}                 -- Show information about the stream.
5) HELP                         -- Print this help.

מגבלת גודל זרם

יישומים רבים לא רוצים לאסוף נתונים לתוך זרם לנצח. לעתים קרובות שימושי לאפשר מספר מקסימלי של הודעות בכל שרשור. במקרים אחרים, כדאי להעביר את כל ההודעות משרשור למאגר קבוע אחר כאשר מגיעים לגודל השרשור שצוין. אתה יכול להגביל את גודל הזרם באמצעות הפרמטר MAXLEN בפקודה XADD:

> XADD mystream MAXLEN 2 * value 1
1526654998691-0
> XADD mystream MAXLEN 2 * value 2
1526654999635-0
> XADD mystream MAXLEN 2 * value 3
1526655000369-0
> XLEN mystream
(integer) 2
> XRANGE mystream - +
1) 1) 1526654999635-0
   2) 1) "value"
      2) "2"
2) 1) 1526655000369-0
   2) 1) "value"
      2) "3"

בעת שימוש ב-MAXLEN, רשומות ישנות נמחקות אוטומטית כשהן מגיעות לאורך מוגדר, כך שלזרם יש גודל קבוע. עם זאת, גיזום במקרה זה אינו מתרחש בצורה היעילה ביותר בזיכרון Redis. אתה יכול לשפר את המצב באופן הבא:

XADD mystream MAXLEN ~ 1000 * ... entry fields here ...

הארגומנט ~ בדוגמה שלמעלה אומר שאנחנו לא בהכרח צריכים להגביל את אורך הזרם לערך מסוים. בדוגמה שלנו, זה יכול להיות כל מספר גדול או שווה ל-1000 (לדוגמה, 1000, 1010 או 1030). רק ציינו במפורש שאנחנו רוצים שהזרם שלנו יאחסן לפחות 1000 רשומות. זה הופך את ניהול הזיכרון להרבה יותר יעיל בתוך Redis.

יש גם צוות נפרד XTRIM, שעושה את אותו הדבר:

> XTRIM mystream MAXLEN 10

> XTRIM mystream MAXLEN ~ 10


אחסון ושכפול מתמשכים

Redis Stream משוכפל באופן אסינכרוני לצמתי עבדים ונשמר בקבצים כמו AOF (תצלום של כל הנתונים) ו-RDB (יומן של כל פעולות הכתיבה). גם שכפול של מצב קבוצות צרכנים נתמך. לכן, אם הודעה נמצאת במצב "בהמתנה" בצומת הראשי, אז בצמתי העבד תהיה להודעה זו אותו סטטוס.

הסרת אלמנטים בודדים מזרם

יש פקודה מיוחדת למחיקת הודעות XDEL. הפקודה מקבלת את שם השרשור ואחריו את מזהי ההודעה שיש למחוק:

> XRANGE mystream - + COUNT 2
1) 1) 1526654999635-0
   2) 1) "value"
      2) "2"
2) 1) 1526655000369-0
   2) 1) "value"
      2) "3"
> XDEL mystream 1526654999635-0
(integer) 1
> XRANGE mystream - + COUNT 2
1) 1) 1526655000369-0
   2) 1) "value"
      2) "3"

בעת שימוש בפקודה זו, עליך לקחת בחשבון שהזיכרון בפועל לא ישוחרר מיד.

זרמים באורך אפס

ההבדל בין זרמים למבני נתונים אחרים של Redis הוא שכאשר למבני נתונים אחרים אין יותר אלמנטים בתוכם, כתופעת לוואי, מבנה הנתונים עצמו יוסר מהזיכרון. כך, למשל, הסט הממוין יוסר לחלוטין כאשר הקריאה ZREM תסיר את האלמנט האחרון. במקום זאת, חוטים רשאים להישאר בזיכרון גם מבלי שיהיו בפנים אלמנטים כלשהם.

מסקנה

Redis Stream אידיאלי ליצירת מתווכים של הודעות, תורי הודעות, רישום מאוחד ומערכות צ'אט לשמירת היסטוריה.

כמו שאמרתי פעם ניקלאוס וירט, תוכניות הן אלגוריתמים בתוספת מבני נתונים, ו-Redis כבר נותן לך את שניהם.

מקור: www.habr.com