Redis Streams: XADD, XRANGE, XREAD ve Akış Kırpma Rehberi

Stream ne işe yarar?

Belge akışı, tipik bir ekleme günlüğünün sınırlarını aşan işlemlerle donatılmış bir yapı olarak tanımlar. Kayıtlara kimlikleriyle doğrudan erişebilir, birden fazla tüketim stratejisi kullanabilirsiniz. Belgedeki örnek kullanım alanları şunlardır: kullanıcı tıklamaları gibi olayların kaydı (event sourcing), sahadaki cihazlardan gelen sensör okumaları ve her kullanıcının bildirimlerini ayrı bir akışta tutmak.

Akışlar içeride radix tree ile uygulanır. Kayıt eklemek O(1) karmaşıklıktadır; tek bir kayda erişim, kimliğin uzunluğuna bağlı O(n) olsa da kimlikler kısa ve sabit uzunlukta olduğu için pratikte sabit zamanlı bir aramaya dönüşür. Her kayıt bir ya da daha fazla alan-değer çiftinden oluşur; bu yönüyle bir hash'e benzer. Fark şudur: hash tek bir nesnenin güncel hâlini tutar, akış ise birbirini izleyen olayların sıralı geçmişini saklar.

XADD ile kayıt eklemek

Temel sözdizimi XADD key [NOMKSTREAM] [MAXLEN|MINID [=|~] threshold [LIMIT count]] id|* field value [field value ...] şeklindedir. Kimlik yerine * yazdığınızda sunucu yeni bir kimlik üretir ve bunu yanıt olarak döndürür. Belgedeki yarış örneğinde her bisikletçi bir kontrol noktasından geçtiğinde bir kayıt eklenir:

> XADD race:france * rider Castilla speed 30.2 position 1 location_id 1
"1692632086370-0"
> XADD race:france * rider Norem speed 28.8 position 3 location_id 1
"1692632094485-0"
> XADD race:france * rider Prickett speed 29.7 position 2 location_id 1
"1692632102976-0"
> XADD race:france * rider Castilla speed 29.9 position 1 location_id 2
"1692632147973-0"
> XLEN race:france
(integer) 4

Kimlikler zamana bağlı olduğu için kendi sunucunuzda farklı değerler görürsünüz; belge de bunu özellikle hatırlatır. Alanlar ve değerler, XADD ile verdiğiniz sırayla saklanır ve okuma komutları onları aynı sırayla döndürür. Akış anahtarı yoksa XADD onu oluşturur. Bunu istemiyorsanız NOMKSTREAM ekleyin; anahtar yoksa komut kayıt eklemez ve boş yanıt döner. XLEN akıştaki kayıt sayısını tamsayı olarak verir.

Kimliklerin yapısı

Bir kayıt kimliği milisaniye-sıra biçimindedir. İlk kısım, kimliği üreten Redis düğümünün milisaniye cinsinden Unix zamanıdır; ikinci kısım aynı milisaniyede üretilen kayıtları ayırmak için kullanılan sıra numarasıdır. İki kısım da 64 bitlik tamsayıdır, bu yüzden aynı milisaniyede üretilebilecek kayıt sayısı pratikte sınırsızdır.

Redis kimliklerin her zaman artmasını garanti eder. Saat geri giderse ya da bir failover sonrası yeni ana düğümün saati farklıysa, sunucu akıştaki en büyük kimliğin zamanını kullanıp sıra kısmını artırır. Böylece kayıtlar akış içinde tam sıralı kalır. Replikalara da tam belirtilmiş kimliklerle XADD komutları gönderildiği için replikalarda kimlikler ana düğümle aynıdır.

Kimliği kendiniz de verebilirsiniz; belge bunun yalnızca nadir durumlarda, örneğin kimlikleri başka bir sistem (bir SQL tablosu gibi) üretiyorsa ve akışın onlarla eşleşmesi gerekiyorsa yararlı olduğunu söyler. Geçerli en küçük kimlik 0-1'dir ve verdiğiniz kimlik akıştaki en büyük kimlikten büyük olmalıdır:

> XADD race:usa 0-1 racer Castilla
0-1
> XADD race:usa 0-2 racer Norem
0-2
> XADD race:usa 0-1 racer Prickett
(error) ERR The ID specified in XADD is equal or smaller than the target stream top item
> XADD race:usa 0-* racer Prickett
0-3

Son satırdaki 0-* biçimi milisaniye kısmını sizin vermenize, sıra kısmını Redis'in üretmesine izin verir; belge bu biçim için Redis 7 ya da üstünü şart koşar. Yalnızca milisaniye kısmını yazdığınız eksik bir kimlik ise sıra kısmı sıfır kabul edilerek yorumlanır.

XRANGE ve XREVRANGE ile aralık sorgusu

XRANGE key start end [COUNT n] iki kimlik arasındaki kayıtları döndürür ve aralık iki uçta da kapsayıcıdır. - mümkün olan en küçük, + en büyük kimlik anlamına gelir. Her kayıt iki elemanlı bir dizi olarak döner: kimlik ve alan-değer listesi. Belgedeki örnekte ilk iki kayıt istenir:

> XRANGE race:france 1692632086370-0 + COUNT 2
1) 1) "1692632086370-0"
   2) 1) "rider"
      2) "Castilla"
      3) "speed"
      4) "30.2"
      5) "position"
      6) "1"
      7) "location_id"
      8) "1"
2) 1) "1692632094485-0"
   2) 1) "rider"
      2) "Norem"
      3) "speed"
      4) "28.8"
      5) "position"
      6) "3"
      7) "location_id"
      8) "1"

Zaman aralığına göre sorgu

Kimliğin sol kısmı zaman olduğu için aralık sorguları kendiliğinden zaman sorgusudur. Sıra kısmını yazmazsanız başlangıçta 0, bitişte en büyük sıra numarası varsayılır. Böylece yalnızca iki milisaniye değeriyle o zaman diliminde üretilen tüm kayıtları alırsınız. Belgedeki örnekte iki milisaniyelik bir pencere sorgulanır ve içinde tek kayıt bulunur:

> XRANGE race:france 1692632086369 1692632086371
1) 1) "1692632086370-0"
   2) 1) "rider"
      2) "Castilla"
      3) "speed"
      4) "30.2"
      5) "position"
      6) "1"
      7) "location_id"
      8) "1"

Sayfalama

Gerçek verilerde birkaç saatlik bir aralık çok büyük bir yanıt üretebilir, bu yüzden COUNT ile parça parça okursunuz. Bir sonraki sayfayı istemek için son dönen kimliğin önüne ( koyarsınız; bu, o kimliği hariç tutan açık bir başlangıç anlamına gelir. Akışın sonuna geldiğinizde yanıt boş dizidir:

> XRANGE race:france (1692632094485-0 + COUNT 2
1) 1) "1692632102976-0"
   2) 1) "rider"
      2) "Prickett"
      3) "speed"
      4) "29.7"
      5) "position"
      6) "2"
      7) "location_id"
      8) "1"
2) 1) "1692632147973-0"
   2) 1) "rider"
      2) "Castilla"
      3) "speed"
      4) "29.9"
      5) "position"
      6) "1"
      7) "location_id"
      8) "2"
> XRANGE race:france (1692632147973-0 + COUNT 2
(empty array)

XRANGE aramayı O(log N), dönen M kaydı O(M) sürede yapar. Küçük bir COUNT ile her adım hızlıdır. Belge bu nedenle XRANGE komutunu akışların fiilî yineleyicisi olarak görür; ayrı bir XSCAN komutuna gerek yoktur. Anahtar uzayını gezen SCAN ile karıştırmayın; o konu SCAN yazısında anlatılıyor.

XREVRANGE aynı işi ters sırada yapar ve başlangıç ile bitişi de ters sırada alır. En pratik kullanımı, akıştaki son kaydı görmektir:

> XREVRANGE race:france + - COUNT 1
1) 1) "1692632147973-0"
   2) 1) "rider"
      2) "Castilla"
      3) "speed"
      4) "29.9"
      5) "position"
      6) "1"
      7) "location_id"
      8) "2"

XREAD ile okumak ve beklemek

XREAD [COUNT n] [BLOCK ms] STREAMS key [key ...] id [id ...], her akış için verdiğiniz kimlikten büyük kimlikli kayıtları döndürür. Tek zorunlu kısım STREAMS bölümüdür ve değişken sayıda argüman aldığı için her zaman en sonda yer almalıdır: önce tüm anahtarlar, ardından aynı sırayla her anahtar için son gördüğünüz kimlik gelir. Yanıtta anahtar adı da bulunur, çünkü tek çağrıda birden fazla akış okunabilir:

> XREAD COUNT 2 STREAMS race:france 0
1) 1) "race:france"
   2) 1) 1) "1692632086370-0"
         2) 1) "rider"
            2) "Castilla"
            3) "speed"
            4) "30.2"
            5) "position"
            6) "1"
            7) "location_id"
            8) "1"
      2) 1) "1692632094485-0"
         2) 1) "rider"
            2) "Norem"
            3) "speed"
            4) "28.8"
            5) "position"
            6) "3"
            7) "location_id"
            8) "1"

Burada 0, eksik kimlik kuralı gereği 0-0 demektir; yani akışın başından itibaren okunur. COUNT her akış için ayrı ayrı uygulanır.

BLOCK ve özel $ kimliği

BLOCK eklenince komut, döndürecek veri yoksa belirtilen milisaniye kadar bekler; 0 süresiz bekleme anlamına gelir. Redis'in çoğu bloklayan komutu saniye alırken bu komut milisaniye alır, ancak belge sunucunun zaman aşımı çözünürlüğünün 0,1 saniye civarında olduğunu not eder. Döndürecek veri varsa BLOCK hiç yokmuş gibi hemen yanıt verilir. Süre dolarsa yanıt boştur:

> XREAD COUNT 100 BLOCK 300 STREAMS race:france $
(nil)

$ kimliği "akıştaki en büyük kimlik" demektir; yani yalnızca beklemeye başladıktan sonra eklenen kayıtları istersiniz. Belge bunun çok önemli bir kuralını vurgular: $ yalnızca ilk çağrıda kullanılmalıdır. Sonraki çağrılarda en son aldığınız kaydın kimliğini vermelisiniz; aksi halde iki çağrı arasında eklenen kayıtları kaçırırsınız. Tipik döngü şöyledir:

> XREAD BLOCK 5000 COUNT 100 STREAMS mystream $
> XREAD BLOCK 5000 COUNT 100 STREAMS mystream 1526999644174-3

Birden fazla akışı beklerken, en az birinde veri varsa komut hemen döner; yoksa ilk veri alan akışın kayıtlarıyla döner. Aynı akışı bekleyen istemciler adil biçimde, ilk bekleyen ilk uyanacak şekilde sıraya alınır.

Özel + kimliği ve 8.10 sınırları

Belgeye göre Redis 7.4 sürümünden itibaren XREAD içinde + kullanarak her akışın yalnızca son kaydını isteyebilirsiniz; örneğin XREAD STREAMS streamA streamB + +. Bu durumda o akış için COUNT yok sayılır. Çok sayıda akışın son kaydını tek komutta almak, her biri için ayrı XREVRANGE göndermekten daha pratiktir.

Redis 8.10 ile gelen MAXCOUNT ve MAXSIZE seçenekleri, tüm akışlar boyunca toplam kayıt sayısını ve toplam yanıt boyutunu (bayt) sınırlar. COUNT akış başına sınırken N akışı COUNT C ile okumak N x C kayda kadar çıkabilir; bu seçenekler o toplamı denetim altına alır. İkisi birlikte verildiğinde MAXCOUNT değeri COUNT değerinden küçük olamaz. Yanıt en az bir kayıt içerir, yani MAXSIZE sınırından büyük tek bir kayıt yine döndürülür. Daha eski sunucularda bu seçenekler yoktur.

XREAD, listeler ve Pub/Sub arasındaki fark

Akışı dinlemek, Pub/Sub kanalına abone olmaya ya da bloklayan liste komutlarıyla beklemeye benzer görünür; ancak belge üç temel farkın altını çizer. Birincisi, aynı akışı bekleyen her tüketici yeni kaydın bir kopyasını alır (fan-out); listelerde ise her öğe tek bir tüketiciye gider. İkincisi, Pub/Sub mesajları saklamaz ve listeden alınan öğe silinir; akışta kayıtlar siz silene kadar kalır ve her tüketici neyin yeni olduğunu son gördüğü kimlikle bilir. Üçüncüsü, onay, bekleyen mesajları inceleme ve devralma gibi denetimler yalnızca tüketici gruplarında vardır.

Kuyruk tarafı için list komutları yazısına, yayın tarafı için Pub/Sub yazısına bakabilirsiniz.

Akışı sınırlamak: MAXLEN, MINID ve XTRIM

Çoğu uygulama veriyi akışta sonsuza kadar tutmak istemez. XADD içindeki MAXLEN seçeneği, sınır aşıldığında en eski kayıtları otomatik olarak atar. Belgedeki örnekte sınır 2 iken üç kayıt eklenir ve yalnızca son ikisi kalır:

> XADD race:italy MAXLEN 2 * rider Jones
"1692633189161-0"
> XADD race:italy MAXLEN 2 * rider Wood
"1692633198206-0"
> XADD race:italy MAXLEN 2 * rider Henshaw
"1692633208557-0"
> XLEN race:italy
(integer) 2
> XRANGE race:italy - +
1) 1) "1692633198206-0"
   2) 1) "rider"
      2) "Wood"
2) 1) "1692633208557-0"
   2) 1) "rider"
      2) "Henshaw"

Tam kırpma pahalı olabilir; akışlar bellek verimliliği için birkaç on kayıtlık makro düğümler hâlinde tutulur ve tek bir düğümü değiştirmek ideal değildir. MAXLEN ~ 1000 yazdığınızda "en az 1000 kayıt kalsın, birkaç on fazlası olabilir" demiş olursunuz; kırpma yalnızca bütün bir düğüm silinebildiğinde yapılır. Belge bunun genellikle istediğiniz şey olduğunu söyler. İstemci kütüphanelerinin varsayılanları farklı olabilir; belgeye göre Python istemcisi varsayılan olarak yaklaşık kırpma kullanır.

XTRIM aynı kırpmayı ekleme yapmadan, tek başına uygular ve silinen kayıt sayısını döndürür. İki strateji vardır: MAXLEN uzunluğa göre, MINID ise verilen kimlikten küçük kimlikli kayıtları atar. Kimlik zamana bağlı olduğu için MINID pratikte "belirli bir andan önceki kayıtları sil" anlamına gelir:

XTRIM mystream MAXLEN 1000
XTRIM mystream MAXLEN ~ 1000
XTRIM mystream MINID 649085820

Belge, akışta "şu süreden eski olmayanları tut" diye otomatik bir seçenek bulunmamasını bilinçli bir tercih olarak açıklar: ekleme hızı değiştikçe zamana göre kırpmanın maliyetini öngörmek zordur ve bir duraklamadan sonra sunucu uzun süre silme yapmak zorunda kalabilir. Bu yüzden kapasiteyi önceden planlayıp uzunluğa göre sınır koymanız ya da MINID ile kırpmayı bilinçli olarak kendiniz çağırmanız beklenir. ~ ile birlikte kullanılan LIMIT count tek seferde silinecek kayıt sayısına üst sınır koyar; verilmezse makro düğümdeki kayıt sayısının 100 katı kullanılır, 0 ise sınırı kaldırır.

Redis 8.2 ile kırpma seçeneklerine tüketici gruplarıyla etkileşimi belirleyen KEEPREF (varsayılan), DELREF ve ACKED eklendi. Grup yoksa bu seçeneklerin etkisi yoktur; gruplarla birlikte nasıl çalıştıkları consumer group yazısında anlatılıyor.

XDEL ve boş akışlar

Yalnızca sona eklenen bir yapıda ortadan kayıt silmek garip görünebilir, ancak belge gizlilik düzenlemeleri gibi durumlarda bunun yararlı olduğunu söyler. XDEL key id [id ...] verilen kimlikleri siler ve silinen sayıyı döndürür:

> XDEL race:italy 1692633208557-0
(integer) 1
> XRANGE race:italy - + COUNT 2
1) 1) "1692633198206-0"
   2) 1) "rider"
      2) "Wood"

Akışlar bir konuda diğer veri tiplerinden ayrılır. Bir sorted set son üyesi silindiğinde anahtarıyla birlikte yok olur; akış ise XDEL ya da sıfır sınırlı MAXLEN sonrasında sıfır kayıtla var olmaya devam eder. Bunun nedeni, akışa bağlı tüketici gruplarının durumunu kaybetmemektir. Belgeye göre şu anda grup olmasa bile akış silinmez. Anahtarı tamamen kaldırmak istiyorsanız DEL kullanmanız gerekir; anahtar komutları için keyspace yazısına bakın.

Kalıcılık ve mesaj güvenliği

Akışlar da diğer veri yapıları gibi replikalara asenkron olarak kopyalanır ve AOF ile RDB dosyalarına yazılır; tüketici gruplarının tam durumu da bunlara dahildir. Belge üç uyarı yapar: mesajların kalıcılığı önemliyse AOF güçlü bir fsync politikasıyla kullanılmalıdır; varsayılan asenkron replikasyon, failover sonrasında bazı XADD işlemlerinin eksik kalmayacağını garanti etmez; WAIT komutu değişikliklerin replikalara yayılmasını zorlayarak kayıp ihtimalini çok azaltır ama sıfırlamaz. Kalıcılık kavramlarının genel anlatımı Redis nedir yazısında bulunuyor.

Uygulamalı mini örnek: sipariş olay günlüğü

Bir e-ticaret uygulamasında her siparişin durum değişikliklerini tek bir akışa yazdığınızı düşünün. Komutlar belgelerdeki sözdizimine göre yazıldı; kimlikler sunucuya göre değişeceği için yanıtları göstermiyoruz.

> XADD siparis:olaylar MAXLEN ~ 100000 * siparis 5012 durum olusturuldu
> XADD siparis:olaylar MAXLEN ~ 100000 * siparis 5012 durum odendi
> XREVRANGE siparis:olaylar + - COUNT 10
> XREAD BLOCK 5000 COUNT 100 STREAMS siparis:olaylar $

İlk iki komut her olayı otomatik kimlikle ekler ve yanıt olarak yeni kaydın kimliğini döndürür; MAXLEN ~ akışı yaklaşık 100 bin kayıtla sınırlar. XREVRANGE panelde gösterilecek son on olayı yeniden eskiye getirir. Son komut bir bildirim servisinin döngüsünün ilk adımıdır: yalnızca bundan sonra gelen olayları bekler, beş saniye içinde olay gelmezse boş yanıt alır. Servis bir olay aldığında bir sonraki çağrıda $ yerine aldığı son kimliği kullanmalıdır. Aynı olayları birden fazla işçi arasında paylaştırmak ve her birinin işlendiğini onaylamak istediğinizde tüketici gruplarına geçmeniz gerekir.

Sık yapılan hatalar

• Döngüde sürekli $ kullanmak. İki çağrı arasında gelen kayıtlar kaçırılır; ilk çağrıdan sonra son alınan kimliği verin.
• STREAMS sonrasına seçenek yazmak. COUNT ve BLOCK gibi tüm seçenekler STREAMS bölümünden önce gelmelidir.
• Sayfalamada son kimliği yeniden istemek. Kapsayıcı aralık yüzünden son kayıt iki kez gelir; önüne ( koyun.
• Kendi kimliğinizi küçük vermek. Akıştaki en büyük kimlikten küçük ya da ona eşit bir kimlik hata ile reddedilir.
• Sınırsız büyüyen akış. MAXLEN ~ ya da düzenli XTRIM olmadan akış bellekte büyümeye devam eder; bellek sınırları için maxmemory yazısına bakın.
• $ ile + kimliklerini karıştırmak. + her akış için mümkün olan en büyük kimliktir, $ ise belirli bir akıştaki en büyük mevcut kimliktir.

Özel kimliklerin özeti

Sık Sorulan Sorular

Stream mi, List mi kullanmalıyım?

Her mesajın tek bir işçiye gitmesi ve alınınca silinmesi yetiyorsa liste basittir. Geçmişi saklamak, aynı olayı birden fazla tüketiciye ulaştırmak ya da işlemeyi onaylarla takip etmek istiyorsanız akış uygundur.

Belirli bir tarihten sonraki kayıtları nasıl okurum?

Tarihi milisaniye cinsinden Unix zamanına çevirip XRANGE key <milisaniye> + çağırın. Sıra kısmını yazmazsanız başlangıçta 0 kabul edilir.

Akış boşalınca anahtar silinir mi?

Hayır. Akışlar sıfır kayıtla da var olmaya devam eder. Anahtarı kaldırmak için DEL kullanın.

XREAD BLOCK 0 sonsuza kadar mı bekler?

Evet, 0 zaman aşımı olmadan beklemek demektir. Bağlantı kopmalarını yönetmek için sınırlı bir süre verip döngüde yeniden çağırmak daha kolay izlenen bir yapı sağlar. Bağlantı ve zaman aşımı ayarları için bağlantı yazısına bakabilirsiniz.

Marka adları sahiplerine aittir. Bu yazı bağımsız bir teknik rehberdir; adı geçen kuruluşlarla bayilik veya ortaklık ilişkisi bulunduğu anlamına gelmez.

İlgili içerikler: Redis öğrenme rehberi · Streams consumer group · List komutları · Redis Pub/Sub · Redis ile iş kuyruğu