Redis Stream — нов абстрактен тип данни, представен в Redis с излизането на версия 5.0
Концептуално, Redis Stream е списък, в който можете да добавяте записи. Всяка запис има уникален идентификатор. По подразбиране идентификаторът се генерира автоматично и включва времева печат. Затова можете да заявявате диапазони от записи по време или да получавате нови данни, докато те постъпват в потока, подобно на Unix командата „tail -f“, която прочита лог файл и чака нови данни. Обърнете внимание, че потокът може да бъде слушан едновременно от няколко клиента, както много процеси „tail -f“ могат едновременно да четат файл, без да влизат в конфликт помежду си.

За да разберем всички предимства на новия тип данни, нека бегло си припомним съществуващите структури Redis, които частично повтарят функционалността на Redis Stream.

Redis PUB/SUB

Redis Pub/Sub е проста система за съобщения, вече вградена във вашето key-value хранилище. Въпреки това, за простотата се плаща:

Ако издателят по някаква причина се провали, той губи всички свои абонати.
На издателя му е необходимо да знае точния адрес на всичките си абонати.
Издателят може да натовари абонатите си, ако данните се публикуват по-бързо, отколкото могат да се обработват.
Сообщението се премахва от буфера на издателя веднага след публикуването, независимо от това колко абонати са го получили и колко бързо са успели да обработят това съобщение.
Всички абонати ще получат съобщението едновременно. Абонатите сами трябва да се съгласят помежду си относно реда на обработка на едно и също съобщение.
Няма вграден механизъм за потвърждение за успешна обработка на съобщение от абоната. Ако абонатът получи съобщение и се срине по време на обработка, издателят не ще узнае за това.

Redis List

Redis List е структура от данни, която поддържа команди за четене с блокировка. Можете да добавяте и прочитате съобщения от началото или края на списъка. На базата на тази структура можете да направите добър стек или опашка за вашата разпределена система и това в повечето случаи ще бъде достатъчно. Основните разлики от Redis Pub/Sub:

Съобщението се доставя на един клиент. Първият блокиран за четене клиент получава данните пръв.
Клинт трябва сам да иницира операцията по четене на всяко съобщение. List не знае нищо за клиентите.
Съобщенията се съхраняват, докато не бъдат прочетени или явно изтрити. Ако сте настроили Redis сървър да записва данни на диск, надеждността на системата значително нараства.

Въведение в Stream

Добавяне на записа в потока

Екип XADD добавя нов запис в потока. Записът не е просто низ, той се състои от едно или повече ключ-стойност двойки. По този начин всеки запис вече е структуриран и напомня структурата на CSV файл.

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

В примера по-горе добавяме в потока с името (ключа) «mystream» две полета: «sensor-id» и «temperature» с стойности «1234» и «19.8» съответно. Като втори аргумент командата приема идентификатора, който ще бъде присвоен на записа — този идентификатор уникално идентифицира всеки запис в потока. Въпреки това, в този случай предадохме *, защото искаме Redis да генерира нов идентификатор за нас. Всеки нов идентификатор ще бъде увеличен. Следователно всеки нов запис ще има по-голям идентификатор в сравнение с предишните записи.

Формат на идентификатора

Идентификаторът на записа, върнат от командата XADD, се състои от две части:

{millisecondsTime}-{sequenceNumber}
millisecondsTime — Unix време в милисекунди (време сървър Redis). Въпреки това, ако текущото време е равно или по-малко от времето на предишния запис, ще се използва времевата метка на предишния запис. Следователно, ако времето на сървъра потъне назад, новият идентификатор все още ще запази свойството на увеличаване.

sequenceNumber се използва за записи, създадени в една и съща милисекунда. sequenceNumber ще бъде увеличен с 1 спрямо предишния запис. Тъй като sequenceNumber има размер 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 Идентификаторът, посочен в XADD, е равен или по-малък от целевия елемент на потока

Брой записи в потока

Можете да получите броя на записите в потока, просто използвайки командата 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"

Всяка връчена запис представлява масив от два елемента: идентификатор и списък от ключ-стойност. Вече казахме, че идентификаторите на записите са свързани с времето. Следователно можем да запитваме диапазон от конкретен времеви интервал. Можем обаче да посочим в запитването непълен идентификатор, а само Unix време, пропускайки частта, свързана с sequenceNumberПропуснатата част от идентификатора автоматично ще бъде приравнена на нула в началото на диапазона и на максималната възможна стойност в края на диапазона. По-долу е даден пример за това как да поискате диапазон, равен на две милисекунди.

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

Имаме само една запис в този диапазон, но в реални набори от данни върнатият резултат може да бъде огромен. Поради тази причина XRANGE поддържа опцията COUNT. Посочвайки количество, можем просто да получим първите N записа. Ако трябва да получим следващите N записа (пагинация), можем да използваме последния получен идентификатор, да увеличим него sequenceNumber с едно и да повторим заявката. Нека да разгледаме това в следващия пример. Започваме да добавяме 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 към sequenceNumber.
Резултатният идентификатор, в този случай 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 приема аргументите на диапазона start и stop в обратен ред.

Четене на нови записи с помощта на XREAD

Често възниква задача да се абонираме за поток и да получаваме само нови съобщения. Тази концепция може да изглежда подобна на Redis Pub/Sub или блокиращ Redis List, но има принципни разлики в начина, по който се използва Redis Stream:

Всяко ново съобщение по подразбиране се доставя до всеки подписчик. Това поведение се различава от блокиращия Redis List, където новото съобщение ще бъде прочетено само от един подписчик.
Докато в Redis Pub/Sub всички съобщения се забравят и никога не се запазват, в Stream всички съобщения се съхраняват за неопределен срок (освен ако клиентът не извика експлицитно премахването им).
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 милисекунди (т.е. означава безкрайно изчакване). Освен това, вместо да предадем обикновен идентификатор за потока 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_name: mygroup          
| consumer_group_stream: somekey        
| last_delivered_id: 1292309234234-92    
|                                                           
| consumers:                                          
|    "consumer-1" с изостанали съобщения  
|       1292309234234-4                          
|       1292309234232-8                          
|    "consumer-42" с изостанали съобщения 
|       ... (и така нататък)                             
+----------------------------------------+

Сега е време да се запознаем с основните команди за Consumer Group, а именно:

XGROUP използва се за създаване, унищожаване и управление на групи
XREADGROUP използва се за четене на потока през групата
XACK — тази команда позволява на абоната да маркира съобщение като успешно обработено

Създаване на Consumer Group

Да предположим, че потокът 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 ще ви върне историята на съобщенията с статус «pending», които са били доставени на посочения абонат (Алиса), но все още не са били потвърдени чрез командата XACK.

Можем да проверим това поведение, като веднага посочим идентификатор 0, без опцията COUNT. Просто ще видим единственото изчакващо съобщение, а именно съобщението с ябълката:

> 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) (празен списък или набор)

Сега дойде редът на Боб да прочете нещо:

> 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, поиска не повече от две съобщения. Командата съобщава само за недоставените съобщения поради специалния идентификатор «>». Както виждате, съобщението «apple» не се показва, тъй като вече е било доставено на Алиса, и затова Боб получава «orange» и «strawberry».

По този начин Алиса, Боб и всеки друг абонат на групата могат да четат различни съобщения от един и същ поток. Те също могат да четат историята на необработените съобщения или да маркират съобщенията като обработени.

Има няколко неща, които трябва да имате предвид:

Веднъж щом абонатът прочете съобщението с командата XREADGROUP, това съобщение преминава в състояние «pending» и се закрепва за този конкретен абонат. Други абонати на групата няма да могат да прочетат това съобщение.
Абонатите се създават автоматично при първото споменаване, не е необходимо да се създават изрично.
С помощта на XREADGROUP можете да четете съобщения от няколко различни потока едновременно, но за да проработи, трябва предварително да създадете групи с еднакво име за всеки поток с помощта на XGROUP

Восстановление после сбоя

Абонатът може да се възстанови след срив и да прегледа списъка си със съобщения със статус „в обработка“. В действителността обаче абонатите могат да не успеят окончателно. Какво ще стане с висящите съобщения на абоната, ако той не успее да се възстанови след срив?
Consumer Group предлага функция, която се използва точно в такива случаи — когато е необходимо да се смени собственикът на съобщения.

Първо трябва да извикате командата XPENDING, която показва всички съобщения в групата със статус „в обработка“. В най-простата си форма командата се извиква само с два аргумента: името на потока и името на групата:

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

Командата изведе броя на необработените съобщения за цялата група и за всеки абонат. Имаме само Боб с две необработени съобщения, тъй като единственото съобщение, поискано от Алиса, беше потвърдено с помощта на XACK.

Можем да поискаме допълнителна информация, използвайки повече аргументи:

ОЧАКВАНЕ {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}. Това помага да се избегне ситуацията, при която двама клиенти се опитват едновременно да сменят собственика на едни и същи съобщения:

Клиент 1: XCLAIM mystream mygroup Алиса 3600000 1526569498055-0 Клиент 2: XCLAIM mystream mygroup Лора 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) име
   2) "mygroup"
   3) потребители
   4) (integer) 2
   5) чакащи
   6) (integer) 2
2) 1) име
   2) "some-other-group"
   3) потребители
   4) (integer) 1
   5) чакащи
   6) (integer) 0

Командата по-горе показва обща информация за всички групи на указаното течение

> XINFO CONSUMERS mystream mygroup
1) 1) име
   2) "Alice"
   3) чакащи
   4) (integer) 1
   5) бездействие
   6) (integer) 9104628
2) 1) име
   2) "Bob"
   3) чакащи
   4) (integer) 1
   5) бездействие
   6) (integer) 83841983

Командата по-горе показва информация за всички абонати на указаното течение и група.
Ако забравите синтаксиса на командата, просто се обърнете за помощ към самата команда:

> XINFO HELP
1) XINFO {подкоманда} аргумент аргумент ... аргумент. Подкоманди са:
2) CONSUMERS {ключ} {група}  -- Показва потребителските групи на група {група}.
3) GROUPS {ключ}                 -- Показва групите на потребителите на течението.
4) STREAM {ключ}                 -- Показва информация за течението.
5) HELP                         -- Отпечатва тази помощ.

Ограничение на размера на течението

Много приложения не искат да събират данни в течение завинаги. Често е полезно да имате максимално допустим брой съобщения в течението. В другите случаи е полезно да прехвърлите всички съобщения от течението в друго постоянно хранилище при достигане на зададения размер на течението. Можете да ограничите размера на течението с помощта на параметъра MAXLEN в командата XADD:

> XADD mystream MAXLEN 2 * стойност 1
1526654998691-0
> XADD mystream MAXLEN 2 * стойност 2
1526654999635-0
> XADD mystream MAXLEN 2 * стойност 3
1526655000369-0
> XLEN mystream
(integer) 2
> XRANGE mystream - +
1) 1) 1526654999635-0
   2) 1) "стойност"
      2) "2"
2) 1) 1526655000369-0
   2) 1) "стойност"
      2) "3"

При използване на MAXLEN старите записи автоматично се изтриват при достигане на указаната дължина, така че течението има постоянен размер. Въпреки това, отрязването в този случай се извършва не по най-производителния начин в паметта на Redis. Можете да подобрите ситуацията, както следва:

XADD mystream MAXLEN ~ 1000 * ... полета за записи тук ...
Аргументът ~ в примера по-горе означава, че не е необходимо да ограничаваме дължината на течението на конкретна стойност. В нашия пример това може да бъде всяко число, което е равно на или по-голямо от 1000 (например, 1000, 1010 или 1030). Просто ние явно посочихме, че искаме нашето течение да съхрани не по-малко от 1000 записа. Това прави работата с паметта много по-ефективна в Redis.

Съществува и отделна команда XTRIM, която извършва същото:

> XTRIM mystream MAXLEN 10

> XTRIM mystream MAXLEN ~ 10

Постоянно съхранение и репликация

Redis Stream асинхронно реплицира на slave ноди и се съхранява в файлове от тип AOF (снимка на всички данни) и RDB (лог на всички операции за запис). Също така се поддържа репликация на състоянието на Consumer Groups. Следователно, ако съобщението е в статус «pending» на master нодата, то на slave нодите това съобщение ще има същия статус.

Изтриване на отделни елементи от потока

За изтриване на съобщения съществува специална команда 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 премахне последния елемент. Вместо това на потоките им е разрешено да останат в паметта, дори да нямат нито един елемент вътре.