Redis Streams Consumer Group: XREADGROUP, XACK, XPENDING ve XNACK

Tüketici grubu hangi sorunu çözer?

XREAD ile bir akışı dinleyen her istemci her mesajın kopyasını alır. Bu, aynı olayı birçok servise ulaştırmak için iyidir; ancak yavaş işlenen mesajlarda istediğiniz şey genellikle farklıdır: N işçinin her biri mesajların farklı bir kısmını almalıdır. Belgedeki şemada yedi mesaj üç tüketiciye sırayla dağıtılır; 1 numara C1'e, 2 numara C2'ye, 3 numara C3'e, 4 numara yeniden C1'e gider.

Belge bir grubu, akıştan veri alıp birden çok tüketiciye hizmet eden bir "sahte tüketici" gibi tarif eder ve şu garantileri sıralar:

• Bir mesaj gruptaki yalnızca bir tüketiciye teslim edilir.
• Tüketiciler grup içinde büyük-küçük harfe duyarlı bir adla tanınır. Bağlantı koparsa grup durumu korunur ve istemci aynı adla geri döner; benzersiz adı seçmek istemcinin sorumluluğudur.
• Her grup "hiç tüketilmemiş ilk kimliği" bilir; yeni mesaj isteyen tüketiciye yalnızca daha önce teslim edilmemiş mesajlar verilir.
• Mesajın tüketilmiş sayılması için açık bir onay (XACK) gerekir.
• Grup, teslim edilmiş ama onaylanmamış tüm mesajları izler. Her tüketici, geçmişe baktığında yalnızca kendisine teslim edilen mesajları görür.

Aynı akışa birden fazla grup bağlanabilir ve her grup akıştaki her mesajı ayrı ayrı alır. Hatta aynı akışı bir yanda XREAD ile, öte yanda farklı gruplar üzerinden XREADGROUP ile okuyan istemciler birlikte bulunabilir. Böylece örneğin fatura servisi ve bildirim servisi aynı sipariş akışını kendi gruplarıyla bağımsız işler.

XGROUP CREATE: grubu oluşturmak

Sözdizimi XGROUP CREATE key group id|$ [MKSTREAM] [ENTRIESREAD n] biçimindedir. Buradaki kimlik, grubun bakış açısından "son teslim edilen" mesajdır; grup bundan büyük kimlikli mesajları dağıtmaya başlar. $ akıştaki en son mesaj demektir ve yalnızca bundan sonra gelenler dağıtılır; 0 ise akışın tüm geçmişini baştan dağıtır:

> XGROUP CREATE race:france france_riders $
OK
> XGROUP CREATE race:italy italy_riders $ MKSTREAM
OK

Akış yoksa komut varsayılan olarak hata verir. MKSTREAM en sona eklendiğinde boş bir akış da oluşturulur. Belge, MKSTREAM olmadan var olmayan bir akışa grup kurulmaya çalışıldığında şu hatayı gösterir:

ERR The XGROUP subcommand requires the key to exist. Note that for CREATE you may want to use the MKSTREAM option to create an empty stream automatically.

BUSYGROUP: grup zaten var

Aynı adla ikinci kez grup oluşturmak BUSYGROUP hatası döndürür:

> XGROUP CREATE mystream mygroup $ MKSTREAM
OK
> XGROUP CREATE mystream mygroup $ MKSTREAM
(error) BUSYGROUP Consumer Group name already exists

Hata sayfasına göre bu çoğunlukla zararsızdır. Birçok uygulama her açılışta grubu oluşturmayı dener; ilk açılıştan sonra grup zaten vardır. Kodunuzda bu hatayı yakalayıp devam etmeniz yeterlidir. Grubun okumaya başlayacağı noktayı değiştirmek için grubu yeniden oluşturmak yerine XGROUP SETID kullanın. Baştan başlamak istiyorsanız grubu XGROUP DESTROY ile silip yeniden oluşturabilirsiniz; ancak bu, grubun bekleyen mesaj listesini de siler.

XREADGROUP: grup üzerinden okumak

XREADGROUP, XREAD ile neredeyse aynıdır; COUNT ve BLOCK seçenekleri aynı şekilde çalışır. Fark, zorunlu GROUP grup tüketici seçeneğidir. Tüketici ilk kez anıldığında grup içinde otomatik olarak oluşturulur. Belgedeki örnekte beş bisikletçi eklenir ve Alice bir mesaj okur:

> XADD race:italy * rider Castilla
"1692632639151-0"
> XADD race:italy * rider Royce
"1692632647899-0"
> XADD race:italy * rider Sam-Bodden
"1692632662819-0"
> XADD race:italy * rider Prickett
"1692632670501-0"
> XADD race:italy * rider Norem
"1692632678249-0"
> XREADGROUP GROUP italy_riders Alice COUNT 1 STREAMS race:italy >
1) 1) "race:italy"
   2) 1) 1) "1692632639151-0"
         2) 1) "rider"
            2) "Castilla"

Yeni mesajlar mı, bekleyen geçmiş mi?

STREAMS sonrasındaki kimlik, komutun davranışını tamamen değiştirir:

• >: yalnızca şimdiye kadar hiçbir tüketiciye teslim edilmemiş yeni mesajlar döner ve grubun son teslim kimliği ilerler.
• 0 ya da başka bir kimlik: yeni mesaj gelmez; yalnızca bu tüketiciye teslim edilmiş ama henüz onaylanmamış mesajların geçmişi, verilen kimlikten büyük olanlar döner. Bu modda BLOCK, NOACK ve CLAIM yok sayılır.

Alice, 0 ile bakınca bekleyen tek mesajını görür; onu onayladıktan sonra aynı sorgu boş döner:

> XREADGROUP GROUP italy_riders Alice STREAMS race:italy 0
1) 1) "race:italy"
   2) 1) 1) "1692632639151-0"
         2) 1) "rider"
            2) "Castilla"
> XACK race:italy italy_riders 1692632639151-0
(integer) 1
> XREADGROUP GROUP italy_riders Alice STREAMS race:italy 0
1) 1) "race:italy"
   2) (empty array)

Bob aynı grupla iki mesaj istediğinde Castilla gelmez, çünkü o Alice'e teslim edilmişti; Bob sıradaki iki mesajı alır:

> XREADGROUP GROUP italy_riders Bob COUNT 2 STREAMS race:italy >
1) 1) "race:italy"
   2) 1) 1) "1692632647899-0"
         2) 1) "rider"
            2) "Royce"
      2) 1) "1692632662819-0"
         2) 1) "rider"
            2) "Sam-Bodden"

Belgenin dikkat çektiği üç ayrıntı daha var. XREADGROUP okuma yapsa da grubu değiştirdiği için bir yazma komutudur ve yalnızca ana düğümde çalıştırılabilir. Birden fazla akıştan aynı anda okumak mümkündür, ama her akışta aynı adlı bir grup bulunmalıdır. NOACK verilirse okunan mesajlar PEL'e eklenmez, okunduğu anda onaylanmış sayılır; ara sıra mesaj kaybının kabul edilebildiği durumlar içindir.

NOGROUP: grup ya da akış yok

Var olmayan bir grupla okumaya çalışırsanız şu hatayı alırsınız:

> XREADGROUP GROUP mygroup consumer1 STREAMS mystream >
(error) NOGROUP No such key 'mystream' or consumer group 'mygroup' in XREADGROUP with GROUP option

Mesaj hem anahtarı hem grubu adlandırır, çünkü ikisinden biri eksik olabilir. Hata sayfası üç neden sayar: grup hiç oluşturulmamıştır; akış silinmiş ya da süresi dolmuştur (grup akışın parçası olduğu için onunla birlikte silinir); ya da ad farklıdır, çünkü adlar büyük-küçük harfe duyarlıdır. Bir akışın gruplarını görmek için XINFO GROUPS kullanılır. Genel hata mesajları için Redis hata mesajları yazısına bakabilirsiniz.

XACK ve çökme sonrası kurtarma

XACK key group id [id ...] mesajları grubun PEL listesinden çıkarır ve başarıyla onaylanan mesaj sayısını döndürür. Zaten onaylanmış ya da PEL'de olmayan kimlikler sayılmaz. Onay, hem mesajın yeniden işlenmesini önler hem de PEL kaydının tuttuğu belleği serbest bırakır.

Güvenilir bir tüketicinin iskeleti belgede şöyle anlatılır: tüketici açılırken önce kendi bekleyen mesajlarını 0 kimliğiyle okur; çünkü önceki çalışmasında mesaj almış ama onaylayamadan çökmüş olabilir. Bu geçmiş boş dönünce > kimliğine geçer ve yeni mesajları almaya başlar. Her mesajı işledikten sonra XACK gönderir. Akış şöyle özetlenebilir:

XREADGROUP GROUP siparis isci-1 COUNT 10 STREAMS siparis:olaylar 0
XACK siparis:olaylar siparis <islenen-id>
XREADGROUP GROUP siparis isci-1 BLOCK 2000 COUNT 10 STREAMS siparis:olaylar >

Belge bu yaklaşımın bir mesajı birden fazla kez işleyebileceğini açıkça söyler: tüketici mesajı işleyip onaylamadan önce çökerse, açıldığında aynı mesajı yeniden alır. Bu "en az bir kez" teslimdir. Bu nedenle işleme kodunuzun aynı mesajı iki kez gördüğünde zarar vermemesi, yani idempotent olması gerekir.

XPENDING ile bekleyenleri incelemek

XPENDING salt okunur bir komuttur, mesajların sahipliğini değiştirmez ve her zaman güvenle çağrılabilir. En basit hâlinde akış ve grup adını alır ve bir özet döndürür: toplam bekleyen sayısı, en küçük ve en büyük bekleyen kimlik ve tüketici başına bekleyen sayısı:

> XPENDING race:italy italy_riders
1) (integer) 2
2) "1692632647899-0"
3) "1692632662819-0"
4) 1) 1) "Bob"
      2) "2"

Ayrıntılı biçim XPENDING key group [[IDLE min-idle-time] start end count [consumer]] şeklindedir ve her mesaj için kimlik, sahibi, milisaniye cinsinden boşta kalma süresi ve teslim sayısını verir:

> XPENDING race:italy italy_riders - + 10
1) 1) "1692632647899-0"
   2) "Bob"
   3) (integer) 74642
   4) (integer) 1
2) 1) "1692632662819-0"
   2) "Bob"
   3) (integer) 74642
   4) (integer) 1

Boşta kalma süresi, mesajın bir tüketiciye son teslim edilmesinden bu yana geçen süredir. Örnekte Bob'un iki mesajı bir dakikadan uzun süredir bekliyor. Mesajın içeriğini görmek için aynı kimliği iki kez vererek XRANGE kullanabilirsiniz: XRANGE race:italy 1692632647899-0 1692632647899-0.

XCLAIM ve XAUTOCLAIM ile mesaj devralmak

Bir tüketici kalıcı olarak çökerse ve geri dönmezse, onun bekleyen mesajları siz bir şey yapmadıkça sonsuza kadar ona ait kalır. Belge bu özelliğin bilinçli olarak açık tutulduğunu vurgular: bekleyenleri inceleyip belirli mesajları talep etmek uygulamanın işidir. XCLAIM key group consumer min-idle-time id [id ...] verilen mesajların sahipliğini yeni tüketiciye geçirir; ancak yalnızca boşta kalma süresi verilen eşikten büyük olanlar için:

> XCLAIM race:italy italy_riders Alice 60000 1692632647899-0
1) 1) "1692632647899-0"
   2) 1) "rider"
      2) "Royce"

Eşik iki istemcinin aynı mesajı aynı anda talep ettiği durumu çözer. Başarılı bir talep, mesajın boşta kalma süresini sıfırlar ve teslim sayacını artırır; bu yüzden ikinci istemcinin talebi eşiği karşılayamaz ve başarısız olur. JUSTID verildiğinde yalnızca kimlikler döner; bant genişliğinden tasarruf sağlar.

Redis 6.2 ile gelen XAUTOCLAIM key group consumer min-idle-time start [COUNT n] [JUSTID], bu inceleme ve devralma döngüsünü sunucu tarafında yapar. Yanıtın ilk elemanı bir imleçtir; bir sonraki çağrıda bu imleçle devam edersiniz:

> XAUTOCLAIM race:italy italy_riders Alice 60000 0-0 COUNT 1
1) "0-0"
2) 1) 1) "1692632662819-0"
      2) 1) "rider"
         2) "Sam-Bodden"
> XAUTOCLAIM race:italy italy_riders Lora 60000 (1692632662819-0 COUNT 1
1) "1692632662819-0"
2) 1) 1) "1692632647899-0"
      2) 1) "rider"
         2) "Royce"

İmleç olarak 0-0 dönmesi, PEL'in sonuna ulaşıldığını gösterir; bu, yeni boşta mesaj olmadığı anlamına gelmez, döngü baştan yeniden çağrılır. Rehberdeki örnekte yanıtın iki elemanı görünür; komut sayfasına göre yanıt üç elemanlıdır ve üçüncü eleman, akışta artık bulunmadığı için PEL'den silinen kimlikleri listeler. XAUTOCLAIM, kırpılmış ya da XDEL ile silinmiş bir mesaja rastlarsa onu talep etmez, PEL'den çıkarır.

Redis 8.4 ile XREADGROUP komutuna CLAIM min-idle-time seçeneği eklendi. > ile birlikte kullanıldığında komut önce en az bu kadar süredir bekleyen mesajları (en uzun bekleyen önce) talep eder, sonra yeni mesajlara geçer. Talep edilen her mesajın yanıtında boşta kalma süresi ve teslim sayısı da bulunur. Böylece ayrı bir kurtarma süreci yazmadan, okuma döngüsünün kendisi takılan mesajları toplar.

Teslim sayacı ve ölü mektup

XPENDING çıktısındaki sayaç, mesajın kaç kez teslim edildiğini gösterir. Sayaç iki durumda artar: mesaj XCLAIM ile başarıyla talep edildiğinde ve XREADGROUP ile bekleyen geçmiş okunduğunda. Hatalar yaşandığında birkaç kez teslim normaldir. Ancak bozuk ya da işleme kodunda bir hatayı tetikleyen bir mesaj sürekli başarısız olur. Belgenin önerisi, sayaç sizin seçtiğiniz büyük bir değere ulaştığında mesajı başka bir akışa taşıyıp yöneticiye bildirim göndermektir. Redis Streams'te "dead letter" kavramı bu şekilde uygulanır.

XNACK: mesajı gruba geri bırakmak (Redis 8.8)

Redis 8.8 ile gelen XNACK key group SILENT|FAIL|FATAL IDS numids id [id ...], talep etmenin tersidir: tüketici kendi bekleyen mesajlarını onaylamadan gruba geri bırakır. Bırakılan mesajlar boşta kalma süresini beklemeden hemen yeniden teslim edilebilir. Belge şu kullanım alanlarını sayar: kontrollü kapanış, tüketici tarafında ağ ya da kaynak sorunu, büyük bir mesajı daha güçlü bir tüketiciye bırakmak ve zararlı ya da geçersiz mesajı işaretlemek.

> XNACK mystream mygroup FAIL IDS 2 1526569498055-0 1526569498056-0
(integer) 2
> XPENDING mystream mygroup - + 10
1) 1) "1526569498055-0"
   2) ""
   3) (integer) -1
   4) (integer) 1
2) 1) "1526569498056-0"
   2) ""
   3) (integer) -1
   4) (integer) 1

Bırakılan mesajın sahibi boş metin olur ve son teslim zamanı 0 yapılır; bu yüzden talep komutlarındaki boşta kalma eşiği ne olursa olsun hemen talep edilebilir. Bırakılan mesajlar PEL'in başındaki ayrı bir bölgede ilk giren ilk çıkar düzeniyle tutulur ve diğer bekleyenlerden önce teslim edilir. XACK'ten farklı olarak mesaj PEL'den çıkmaz. Komut yalnızca PEL'de olan kimlikleri işler; yanıttaki tamsayı geri bırakılan mesaj sayısıdır.

İdempotent üretim: tekrar gönderilen mesajları ayıklamak (Redis 8.6)

Tüketici tarafındaki "en az bir kez" sorunu kadar üretici tarafında da bir sorun vardır. Üretici XADD gönderdikten sonra, yanıtı almadan bağlantı koparsa ya da üretici çökerse, mesajın eklenip eklenmediğini bilemez ve yeniden gönderir; bu da çift kayıt demektir. Redis 8.6 ile XADD iki yeni kip kazandı:

• IDMP producer-id idempotent-id: üretici kimliği (pid) ve mesaj kimliği (iid) sizin tarafınızdan verilir. iid bir işlem numarası, sayaç ya da UUID olabilir. Karma hesaplanmadığı için daha hızlıdır.
• IDMPAUTO producer-id: yalnızca pid verilir; Redis iid değerini mesajın alan-değer çiftlerinden hesaplar. Aynı içerik aynı iid değerini üretir.

İki kip de yalnızca kimlik * olduğunda kullanılabilir. Aynı pid ve iid ile gelen ikinci XADD yeni kayıt oluşturmaz, ilk kaydın kimliğini döndürür. Komut sayfasındaki örnekte bu açıkça görülür: ikinci komutun değeri farklı olsa da iid aynı olduğu için aynı kimlik döner:

> XADD mystream IDMP producer1 msg1 * field value
"1774378417976-0"
> XADD mystream IDMP producer1 msg1 * field different_value
"1774378417976-0"
> XADD mystream IDMPAUTO producer2 * field value
"1774378417977-0"
> XADD mystream IDMPAUTO producer2 * field value
"1774378417977-0"
> XCFGSET mystream IDMP-DURATION 300 IDMP-MAXSIZE 1000
OK

Redis iid değerlerini sonsuza kadar tutmaz. XCFGSET ile akış başına iki ayar yapılır: IDMP-DURATION saklama süresi (1 ile 86400 saniye, varsayılan 100) ve IDMP-MAXSIZE üretici başına izlenen en fazla iid sayısı (1 ile 10.000, varsayılan 100). İkisinden biri dolunca eski iid silinir ve boyut sınırı süreden daha güçlüdür. Belgenin ölçü önerisi şudur: süreyi, üreticinizin çöktükten sonra yeniden mesaj göndermeye başlaması için gereken en uzun süreye göre seçin; boyutu ise "işaretleme gecikmesi (ms) x milisaniyedeki mesaj sayısı + pay" formülüyle hesaplayın. Örneğin saniyede 1000 mesaj gönderen ve her mesajı en fazla 80 ms içinde gönderildi olarak işaretleyen bir üretici için yaklaşık 100 yeterlidir.

Her üretici yeniden başladıktan sonra da aynı pid değerini kullanmalıdır; farklı pid değerleri aynı iid ile çakışmaz. İzleme için XINFO STREAM idempotentlik kullanıldığında pids-tracked, iids-tracked, iids-added ve iids-duplicates gibi ek alanlar döndürür. Bu kipler eski sunucularda yoktur.

İzleme: XINFO STREAM, GROUPS ve CONSUMERS

XINFO alt komutları alan adı-değer çiftleri döndürür; bu, insanların çıktıyı okumasını kolaylaştırır ve yeni alanların eski istemcileri bozmadan eklenmesine izin verir. XINFO GROUPS grupları, tüketici sayısını, bekleyen sayısını ve son teslim kimliğini gösterir:

> XINFO GROUPS race:italy
1) 1) "name"
   2) "italy_riders"
   3) "consumers"
   4) (integer) 3
   5) "pending"
   6) (integer) 2
   7) "last-delivered-id"
   8) "1692632662819-0"

XINFO CONSUMERS race:italy italy_riders her tüketicinin bekleyen mesaj sayısını ve boşta kalma süresini listeler. Belge, devralma işini ayrı bir süreçle yapmak isteyenlerin aktif tüketicileri bu komutla bulabileceğini belirtir. XINFO STREAM ise akışın uzunluğunu, grup sayısını, ilk ve son kaydını verir.

Birden fazla grup ve silme davranışı

Bir akışa bağlı her grup, her mesajı ayrı ayrı alır. Bir grubun XACK göndermesi yalnızca o grubun PEL kaydını siler; mesaj akışta ve diğer grupların PEL listelerinde kalır. Redis 8.2 öncesinde mesajı ancak tüm gruplar onayladıktan sonra silmek için karmaşık uygulama mantığı gerekiyordu. 8.2 ile kırpma ve silme komutlarına üç seçenek geldi:

• KEEPREF (varsayılan): kayıt silinir ama grupların PEL listelerindeki referanslar korunur.
• DELREF: kayıt silinir ve tüm grupların PEL listelerindeki referanslar da temizlenir.
• ACKED: yalnızca tüm grupların onayladığı kayıtlar silinir.

Bu seçenekler XADD ve XTRIM kırpmasında, XDELEX komutunda ve onaylamayla silmeyi tek atomik işlemde birleştiren XACKDEL komutunda kullanılır. Varsayılan davranışta PEL'de bulunan bir kayıt silinebilir; bu durumda PEL kimliği tutar ama içerik kaybolur ve tüketici geçmişi okuduğunda o kimliğin verisi yerine boş değer görür:

> XADD mystream 1 myfield mydata
"1-0"
> XGROUP CREATE mystream mygroup 0
OK
> XREADGROUP GROUP mygroup myconsumer STREAMS mystream >
1) 1) "mystream"
   2) 1) 1) "1-0"
         2) 1) "myfield"
            2) "mydata"
> XDEL mystream 1-0
(integer) 1
> XREADGROUP GROUP mygroup myconsumer STREAMS mystream 0
1) 1) "mystream"
   2) 1) 1) "1-0"
         2) (nil)

Tüketici kodunuz bu boş değeri bir hata gibi değil, "içerik kırpılmış" bilgisi olarak ele almalı ve kaydı onaylayıp geçmelidir; aksi halde aynı kimlik her açılışta yeniden karşınıza çıkar.

Kafka bölümlerinden farkı

Belge, Redis tüketici gruplarının işlev olarak Kafka gruplarına benzediğini ama uygulamada çok farklı olduğunu vurgular. Bölümler yalnızca mantıksaldır; mesajlar tek bir anahtarda durur ve kim hazırsa ona verilir. Bir tüketici çökerse yeni mesajlar kalanlara gider; hızlı bir tüketici orantılı olarak daha fazla mesaj alır. Tek bir akış birden fazla Redis sunucusuna otomatik bölünmez; bunun için birden çok anahtar ve cluster gibi bir parçalama düzeni gerekir. Bir akış ve N tüketiciyle yük dengelersiniz, ancak aynı mantıksal nesneye ait mesajlar sıra dışı işlenebilir. Sıranın kritik olduğu durumlarda her tüketiciye kendi akış alt kümesini veren N akış modeli daha uygundur.

Sık yapılan hatalar

• Her açılışta XGROUP CREATE sonucunda BUSYGROUP'u gerçek hata saymak. Bu hatayı yakalayıp devam edin.
• Çökmeden sonra doğrudan > ile başlamak. Önce 0 ile kendi bekleyenlerinizi işleyin.
• İki işçiye aynı tüketici adını vermek. Grup ikisini tek tüketici sayar ve bekleyen geçmiş karışır.
• Onayı unutmak. XACK gelmeyen her mesaj PEL'de birikir ve bellek tüketir.
• Teslim sayacına bakmamak. Zehirli bir mesaj sürekli yeniden teslim edilir; sınır koyup ölü mektup akışına taşıyın.
• Grubu replikada okumaya çalışmak. XREADGROUP yazma komutudur, ana düğümde çalışır.

Sık Sorulan Sorular

Tüketiciyi önceden oluşturmam gerekir mi?

Hayır. Tüketici, XREADGROUP içinde ilk kez anıldığında otomatik olarak oluşturulur.

Bir mesajın tam olarak bir kez işlenmesini garanti edebilir miyim?

Belge, genel durumda tam bir kez işlemenin elde edilemeyeceğini söyler. Tüketici tarafında idempotent işleme, üretici tarafında ise Redis 8.6 ile gelen IDMP kipleri çift kayıt riskini azaltır.

XCLAIM ile XAUTOCLAIM arasındaki fark nedir?

XCLAIM belirli kimlikleri devralır; kimlikleri önce XPENDING ile bulmanız gerekir. XAUTOCLAIM boşta kalan mesajları bir imleçle kendisi bulup devralır.

Grubu silersem bekleyen mesajlar ne olur?

XGROUP DESTROY grubun bekleyen mesaj listesini de siler. Mesajların kendisi akışta kalır.

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 · Redis Streams temelleri · Redis ile iş kuyruğu · Redis Pub/Sub · Redis hata mesajları