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

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

Redis PUB/SUB

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

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

Redis списък

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 да генерира нов идентификатор за нас. Всеки нов ID ще се увеличава. Следователно всеки нов запис ще има по-висок идентификатор в сравнение с предишните записи.

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

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

{millisecondsTime}-{sequenceNumber}

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

последователен номер използва се за записи, създадени за същата милисекунда. последователен номер ще се увеличи с 1 спрямо предишния запис. Тъй като последователен номер е с размер 64 бита, тогава на практика не трябва да се натъквате на ограничение за броя записи, които могат да бъдат генерирани в рамките на една милисекунда.

Форматът на такива идентификатори може да изглежда странен на пръв поглед. Един недоверчив читател може да се чуди защо времето е част от идентификатора. Причината е, че потоците на Redis поддържат заявки за обхват по ID. Тъй като идентификаторът е свързан с времето на създаване на записа, това прави възможно запитването за времеви диапазони. Ще разгледаме конкретен пример, когато разглеждаме командата XRANGE.

Ако по някаква причина потребителят трябва да посочи свой собствен идентификатор, който например е свързан с някаква външна система, тогава можем да го предадем на командата XADD вместо *, както е показано по-долу:

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

Моля, обърнете внимание, че в този случай трябва сами да наблюдавате нарастването на ID. В нашия пример минималният идентификатор е "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"

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

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

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

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

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

1. Всяко ново съобщение се доставя до всеки абонат по подразбиране. Това поведение е различно от блокиращ Redis List, където ново съобщение ще бъде прочетено само от един абонат.
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 милисекунди (това означава изчакване за неопределено време). Освен това, вместо да се предаде обичайният идентификатор за потока 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)                             
+----------------------------------------+

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

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

Създаване на потребителска група

Да приемем, че mystream вече съществува. Тогава командата за създаване на група ще изглежда така:

> XGROUP CREATE mystream mygroup $
OK

Когато създаваме група, трябва да подадем идентификатор, започвайки от който групата ще получава съобщения. Ако просто искаме да получаваме всички нови съобщения, тогава можем да използваме специалния идентификатор $ (както в нашия пример по-горе). Ако зададете 0 вместо специален идентификатор, всички съобщения в нишката ще бъдат достъпни за групата.

Сега, когато групата е създадена, можем веднага да започнем да четем съобщения с помощта на командата XREADGROUP. Тази команда е много подобна на XREAD и поддържа незадължителната опция БЛОКИРАНЕ. Въпреки това има задължителна опция 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"

Горната команда гласи дословно, както следва:

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

Всеки път, когато абонат извършва операция върху група, той трябва да предостави своето име, което уникално се идентифицира в групата. В горната команда има още един много важен детайл - специалният идентификатор ">". Този специален идентификатор филтрира съобщенията, оставяйки само тези, които никога преди не са били доставяни.

Освен това в специални случаи можете да зададете реален идентификатор като 0 или друг валиден идентификатор. В този случай командата XREADGROUP ще ви върне хронология на съобщенията със статус "в очакване", които са били доставени на посочения абонат (Алиса), но все още не са били потвърдени с помощта на командата XACK.

Можем да тестваме това поведение, като незабавно посочим ID 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) (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"

Боб, член на моята група, поиска не повече от две съобщения. Командата отчита само недоставени съобщения поради специалния идентификатор ">". Както можете да видите, съобщението "ябълка" няма да се покаже, тъй като вече е доставено на Алис, така че Боб получава "портокал" и "ягода".

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

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

• Веднага след като абонатът счита съобщението за команда 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

Сега имаме подробности за всяко съобщение: ID, име на абонат, време на неактивност в милисекунди и накрая броят опити за доставка. Имаме две съобщения от Боб и те са били неактивни за 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