Redis Stream — нов абстрактен тип данни, представен в Redis с излизането на версия 5.0
Концептуално Redis Stream е списък, в който можете да добавяте записи. Всеки запис има уникален идентификатор. По подразбиране идентификаторът се генерира автоматично и включва времеви маркер. Следователно можете да запрашвате диапазони от записи по време или да получавате нови данни, докато те пристигат в потока, подобно на командата „tail -f“ в Unix, която прочита лог файл и остава в изчакване на нови данни. Обърнете внимание, че потокът може да бъде слушан едновременно от няколко клиента, както много процеси „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
(целочислено) 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"

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

> XRANGE mystream 1518951480106 1518951480107
1) 1) 1518951480106-0
   2) 1) "sensor-id"
      2) "1234"
      3) "температура"
      4) "19.8"

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

Командата изведе броя на необработените съобщения за цялата група и за всеки абонат. Имаме само Боб с две необработени съобщения, защото единственото съобщение, поискано от Алисa, беше потвърдено с помощта на 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) "съобщение"
      2) "портокал"

Просто трябва да повторим същия идентификатор два пъти в аргументите. Сега, когато имаме някаква представа, Алисa може да реши, че след 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) (целочислено) 2
   5) в очакване
   6) (целочислено) 2
2) 1) име
   2) "some-other-group"
   3) потребители
   4) (целочислено) 1
   5) в очакване
   6) (целочислено) 0

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

> XINFO CONSUMERS mystream mygroup
1) 1) име
   2) "Alice"
   3) в очакване
   4) (целочислено) 1
   5) неактивен
   6) (целочислено) 9104628
2) 1) име
   2) "Bob"
   3) в очакване
   4) (целочислено) 1
   5) неактивен
   6) (целочислено) 83841983

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

> XINFO HELP
1) XINFO {subcommand} arg arg ... arg. Подкомандите са:
2) CONSUMERS {key} {groupname}  -- Показва потребителските групи на група {groupname}.
3) GROUPS {key}                 -- Показва потребителските групи на потока.
4) STREAM {key}                 -- Показва информация за потока.
5) 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
(целочислено) 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 * ... полета на входа тук ...
Аргументът ~ в горния пример означава, че не е задължително да ограничаваме дължината на потока с конкретна стойност. В нашия пример това може да бъде всяко число, по-голямо или равно на 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 изтрие последния елемент. Вместо това, потоките могат да останат в паметта, дори и без нито един елемент вътре.