Redis Stream е нов апстрактен тип на податоци воведен во Redis со верзија 5.0
Концептуално, Redis Stream е листа на која можете да додавате записи. Секој запис има единствен идентификатор. Стандардно, ID се генерира автоматски и вклучува временски печат. Затоа, можете да барате опсег на записи со текот на времето или да примате нови податоци како што пристигнуваат во протокот, слично како командата „tail -f“ на Unix што чита датотека за евиденција и замрзнува додека чека нови податоци. Забележете дека повеќе клиенти можат да слушаат нишка во исто време, исто толку процеси "tail -f" можат да читаат датотека истовремено без да се во конфликт еден со друг.

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

Редис PUB/SUB

Redis Pub/Sub е едноставен систем за пораки веќе вграден во вашата продавница за клучеви. Сепак, едноставноста има цена:

• Ако издавачот поради некоја причина не успее, тогаш тој ги губи сите свои претплатници
• Издавачот треба да ја знае точната адреса на сите негови претплатници
• Издавачот може да ги преоптоварува своите претплатници со работа ако податоците се објавуваат побрзо отколку што се обработуваат
• Пораката се брише од тампонот на издавачот веднаш по објавувањето, без оглед на тоа на колку претплатници им била испорачана и колку брзо можеле да ја обработат оваа порака.
• Сите претплатници ќе ја добијат пораката во исто време. Самите претплатници мора некако да се договорат меѓу себе за редоследот на обработка на истата порака.
• Не постои вграден механизам да потврди дека претплатникот успешно обработил порака. Ако претплатникот прими порака и падне за време на обработката, издавачот нема да знае за тоа.

Редис листа

Redis List е податочна структура која поддржува блокирање на команди за читање. Можете да додавате и читате пораки од почетокот или крајот на листата. Врз основа на оваа структура, можете да направите добар оџак или редица за вашиот дистрибуиран систем, и во повеќето случаи тоа ќе биде доволно. Главните разлики од Redis Pub/Sub:

• Пораката се доставува до еден клиент. Првиот клиент блокиран за читање прв ќе ги добие податоците.
• Клинт мора самиот да ја иницира операцијата за читање за секоја порака. Листа не знае ништо за клиентите.
• Пораките се чуваат додека некој не ги прочита или експлицитно не ги избрише. Ако го конфигурирате серверот Redis да ги префрла податоците на дискот, тогаш веродостојноста на системот драстично се зголемува.

Вовед во Stream

Додавање запис во поток

Тим XADD додава нов запис во потокот. Записот не е само низа, тој се состои од еден или повеќе парови клуч-вредности. Така, секој запис е веќе структуриран и наликува на структурата на датотеката CSV.

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

Во горниот пример, додаваме две полиња на потокот со името (клучот) „mystream“: „sensor-id“ и „temperature“ со вредностите „1234“ и „19.8“, соодветно. Како втор аргумент, командата зема идентификатор што ќе биде доделен на записот - овој идентификатор уникатно го идентификува секој запис во потокот. Меѓутоа, во овој случај поминавме * затоа што сакаме Redis да генерира нов ID за нас. Секоја нова лична карта ќе се зголемува. Затоа, секој нов запис ќе има повисок идентификатор во однос на претходните записи.

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

ИД за влез вратен од командата XADD, се состои од два дела:

{millisecondsTime}-{sequenceNumber}

милисекундиВреме — Време на Unix во милисекунди (време на серверот Redis). Меѓутоа, ако моменталното време е исто или помало од времето на претходното снимање, тогаш се користи временскиот печат на претходната снимка. Затоа, ако времето на серверот се врати назад во времето, новиот идентификатор сè уште ќе го задржи својството за зголемување.

секвенцаБрој се користи за записи создадени во иста милисекунда. секвенцаБрој ќе се зголеми за 1 во однос на претходниот запис. Затоа што секвенцаБрој има големина од 64 бита, тогаш во пракса не треба да наидете на ограничување на бројот на записи што може да се генерираат во рок од една милисекунда.

Форматот на таквите идентификатори може да изгледа чуден на прв поглед. Недоверлив читател може да се запраша зошто времето е дел од идентификаторот. Причината е што стримовите на Редис поддржуваат прашања за опсег според ID. Бидејќи идентификаторот е поврзан со времето кога е креиран записот, ова овозможува да се побараат временски опсези. Ќе погледнеме конкретен пример кога ќе ја погледнеме командата 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"

Секој вратен запис е низа од два елементи: идентификатор и листа на парови клуч-вредност. Веќе рековме дека идентификаторите на записите се поврзани со времето. Затоа, можеме да побараме опсег од одреден временски период. Меѓутоа, во барањето можеме да го наведеме не целосниот идентификатор, туку само времето на 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"

За да продолжиме со повторување со следните два елементи, треба да го избереме последниот примен ID, т.е. 1519073279157-0 и да додадеме 1 на секвенцаБрој.
Добиениот ID, во овој случај 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 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)                             
+----------------------------------------+

Сега е време да се запознаете со главните команди за Групата на потрошувачи, имено:

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

Создавање на група на потрошувачи

Да претпоставиме дека mystream веќе постои. Тогаш командата за создавање група ќе изгледа вака:

> XGROUP CREATE mystream mygroup $
OK

Кога креираме група, мора да поминеме идентификатор, од кој групата ќе прима пораки. Ако сакаме само да ги примаме сите нови пораки, тогаш можеме да го користиме специјалниот идентификатор $ (како во нашиот пример погоре). Ако наведете 0 наместо посебен идентификатор, тогаш сите пораки во низата ќе бидат достапни за групата.

Сега кога групата е создадена, можеме веднаш да почнеме да читаме пораки користејќи ја командата XREADGROUP. Оваа команда е многу слична на XREAD и ја поддржува опционалната опција BLOCK. Сепак, постои потребна опција ГРУПА која секогаш мора да биде наведена со два аргументи: името на групата и името на претплатникот. Поддржана е и опцијата 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.

Можеме да го тестираме ова однесување со веднаш наведување на ID 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 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 {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 може да се користи само за враќање на ИД на пораки. Ова е корисно ако не сте заинтересирани за деталите на пораката и сакате да ги зголемите перформансите на системот.

Бројач за испорака

Бројачот што го гледате на излезот ТЕКОТ е бројот на испораки на секоја порака. Таквиот бројач се зголемува на два начина: кога пораката е успешно побарана преку ПРЕТСТАВУВАЊЕ или кога се користи повик 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 асинхроно се реплицира на slave јазли и се зачувува во датотеки како AOF (слика од сите податоци) и RDB (дневник на сите операции за запишување). Поддржана е и репликација на состојбата на Групите на потрошувачи. Затоа, ако пораката е во статус „во чекање“ на главниот јазол, тогаш на 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 ќе го отстрани последниот елемент. Наместо тоа, на нишките им е дозволено да останат во меморијата дури и без да имаат елементи внатре.

Заклучок

Redis Stream е идеален за создавање посредници за пораки, редици за пораки, унифицирано евидентирање и системи за разговор за чување историја.

Како што реков еднаш Никлаус Вирт, програмите се алгоритми плус структури на податоци, а Редис веќе ви ги дава и двете.

Извор: www.habr.com