Redis Streams: Consumer Group, Pending Entry List, XAUTOCLAIM ve Idempotent XADD

Sürüm notu: Bu yazıdaki örnekler Redis 8.10.3 (redis:8), Valkey 9.1.3, .NET 10, StackExchange.Redis 3.4.35 ve Microsoft.Extensions.Hosting 10.0.13 ile çalıştırılmıştır.

Önce Tahmin Et

05'te Bookstore'un (sitedeki serilerin ortak örnek uygulaması, bir online kitapçı) sipariş onay e-postalarını bir list'le gönderiyorduk ve LMOVE ile çöken worker'ın işini processing list'inde kurtarmıştık; ama kimin aldığı, ne zamandır beklediği ve kaç kez denendiği yazmıyordu. Aynı senaryoyu bir Redis stream'iyle kuralım. emails consumer group'undaki emails-worker-1 üç sipariş olayını okuyor, yalnız birincisini onaylıyor (XACK) ve çöküyor:

StreamEntry[] first = await db.StreamReadGroupAsync(Stream, "emails", "emails-worker-1", StreamPosition.NewMessages, count: 10);
await db.StreamAcknowledgeAsync(Stream, "emails", first[0].Id);
Console.WriteLine($"emails-worker-1 read {first.Length}, acknowledged {first[0]["orderId"]}, then crashed");

// 1,1 saniye sonra ikinci worker okuyor: XREADGROUP ... CLAIM 1000
StreamEntry[] second = await db.StreamReadGroupAsync(Stream, "emails", "emails-worker-2", StreamPosition.NewMessages,
    count: 10, noAck: false, claimMinIdleTime: TimeSpan.FromSeconds(1));

Bu arada stream'e 1004 numaralı yeni bir sipariş geliyor. İkinci worker hangi siparişleri alır, hangi sırayla? Onaylanmamış 1002 ve 1003'e ne oldu?

dotnet run --project src/Bookstore.RedisLab -- stream-crash

emails-worker-1 read 3, acknowledged 1001, then crashed
XPENDING: 2 pending, emails-worker-1=2
emails-worker-2 got order 1002: idle=1110 ms, delivered before=1
emails-worker-2 got order 1003: idle=1110 ms, delivered before=1
emails-worker-2 got order 1004: idle=0 ms, delivered before=0
XPENDING after: 0

Redis onaylanmamış iki olayı unutmadı: XPENDING ikisinin de emails-worker-1'de beklediğini gösterdi. İkinci worker tek bir okumayla önce 1,1 saniyedir bekleyen bu iki olayı devraldı (her birinin daha önce bir kez teslim edildiği bilgisiyle), sonra yeni gelen 1004'ü aldı. 05'teki list'te bunun için ayrı bir hash, bir reaper ve epey kod gerekiyordu. Bu yazıda Redis Streams'i anlatıyoruz: stream'in yapısı, consumer group ve pending entry list (PEL; okunmuş ama henüz onaylanmamış girdilerin listesi), XAUTOCLAIM ve Redis 8.4'ün XREADGROUP … CLAIM'i, 8.6'nın idempotent XADD'i, 8.8'in XNACK'i, .NET'te bir BackgroundService consumer'ı, poison message ve trimming.

Stream'in Yapısı: XADD, ID'ler ve XRANGE

Stream, sonuna ekleme yapılan, her girdisinin bir ID'si ve field-value çiftleri olan bir log. Girdiler okununca silinmez; list'ten en büyük farkı bu. Okunabilir olsun diye ilk iki ID'yi elle veriyorum, üçüncüsünü Redis üretiyor:

127.0.0.1:6379> XADD demo:orders:events 1794650000001-0 type order-created orderId 1001 isbn 9780441013593 quantity 2
"1794650000001-0"
127.0.0.1:6379> XADD demo:orders:events 1794650000002-0 type order-created orderId 1002 isbn 9780141439518 quantity 1
"1794650000002-0"
127.0.0.1:6379> XADD demo:orders:events * type order-created orderId 1003 isbn 9780451524935 quantity 1
"1794655623244-0"
127.0.0.1:6379> XLEN demo:orders:events
(integer) 3
127.0.0.1:6379> XRANGE demo:orders:events - + COUNT 1
1) 1) "1794650000001-0"
   2) 1) "type"
      2) "order-created"
      3) "orderId"
      4) "1001"
      5) "isbn"
      6) "9780441013593"
      7) "quantity"
      8) "2"

ID iki parçalı: girdinin eklendiği Unix milisaniyesi ve aynı milisaniyedeki sıra numarası. * ile Redis ID'yi kendi saatinden üretir ve her zaman bir öncekinden büyük olmasını garanti eder; elle verilen ID de sondakinden büyük olmak zorunda. Bu yüzden stream zamana göre sıralı ve XRANGE ile zaman aralığına göre okunabiliyor (- en baştan, + en sona). XREAD ise bir ID'den sonrasını okur; gruplara gerek duymadan, her okuyucunun her şeyi görmesi istenen durumlar (örneğin bir canlı izleme ekranı) için uygun. İçeride girdiler bir radix tree'nin düğümlerinde listpack olarak, aynı field adları bir kez yazılıp ID'ler bir önceki girdiye göre fark olarak saklanıyor; ölçümde bir milyon sipariş olayı girdi başına 42,9 byte tuttu (aşağıda).

Consumer Group ve Pending Entry List

Consumer group, bir stream'i okuyan bir grup worker. Grup, stream'in hangi noktasına kadar dağıttığını (last-delivered-id) ve dağıttığı ama henüz onaylanmamış girdileri (PEL) tutar. Aynı stream'e birden çok grup bağlanabilir ve her grup her girdiyi bir kez alır; bir grubun içindeki worker'lar ise işi paylaşır:

127.0.0.1:6379> XGROUP CREATE demo:orders:events emails 0
OK
127.0.0.1:6379> XGROUP CREATE demo:orders:events stock 0
OK
127.0.0.1:6379> XREADGROUP GROUP emails emails-worker-1 COUNT 2 STREAMS demo:orders:events >
1) 1) "demo:orders:events"
   2) 1) 1) "1794650000001-0"
         2) 1) "type"
            2) "order-created"
            3) "orderId"
            4) "1001"
            5) "isbn"
            6) "9780441013593"
            7) "quantity"
            8) "2"
      2) 1) "1794650000002-0"
         2) 1) "type"
            2) "order-created"
            3) "orderId"
            4) "1002"
            5) "isbn"
            6) "9780141439518"
            7) "quantity"
            8) "1"
127.0.0.1:6379> XREADGROUP GROUP stock stock-worker-1 COUNT 1 STREAMS demo:orders:events >
1) 1) "demo:orders:events"
   2) 1) 1) "1794650000001-0"
         2) 1) "type"
            2) "order-created"
            3) "orderId"
            4) "1001"
            5) "isbn"
            6) "9780441013593"
            7) "quantity"
            8) "2"
127.0.0.1:6379> XACK demo:orders:events emails 1794650000001-0
(integer) 1
127.0.0.1:6379> XPENDING demo:orders:events emails - + 10
1) 1) "1794650000002-0"
   2) "emails-worker-1"
   3) (integer) 150
   4) (integer) 1

XGROUP CREATE … 0 grubu stream'in başından başlatır; $ verseydik yalnız bundan sonra gelenleri görürdü. > "bu gruba henüz hiç dağıtılmamış girdiler" demek. stock grubu, emails'in okuduğu 1001'i yine aldı: gruplar birbirinden bağımsız. XACK'ten sonra emails'in PEL'inde yalnız 1002 kaldı; XPENDING'in ayrıntılı biçimi onun hangi consumer'da olduğunu, 150 ms'dir beklediğini ve bir kez teslim edildiğini gösteriyor. Bir girdi PEL'den yalnız XACK ile çıkar; worker çökerse orada kalır ve devralınmayı bekler.

Ölen Worker'ın İşini Devralmak: XAUTOCLAIM, CLAIM ve XNACK

Redis 6.2'den beri XAUTOCLAIM key group consumer min-idle-time start, belirtilen süreden uzun bekleyen girdileri başka bir consumer'a devrediyor. Redis 8.4 bunu okuma komutuna ekledi: XREADGROUP … CLAIM min-idle-time … >, dokümana göre önce yeterince uzun bekleyen girdileri (en uzun bekleyen önce), sonra yeni girdileri döndürüyor ve devralınan her girdinin yanına ne kadar beklediğini ve daha önce kaç kez teslim edildiğini ekliyor. Önce Tahmin Et'teki idle=1110 ms, delivered before=1 bu ek alanlar; StackExchange.Redis 3.4.35'te StreamEntry.IdleTime ve StreamEntry.DeliveryCount olarak geliyor. Böylece worker'ın ayrı bir "sahipsiz iş toplayıcı" döngüsüne ihtiyacı kalmıyor.

Redis 8.8 ters yönü ekledi: XNACK. Bir worker bir girdiyi işleyemeyeceğini anladığında onu beklemeden gruba geri bırakıyor:

127.0.0.1:6379> XNACK demo:orders:events emails FAIL IDS 1 1794650000002-0
(integer) 1
127.0.0.1:6379> XPENDING demo:orders:events emails - + 10
1) 1) "1794650000002-0"
   2) ""
   3) (integer) -1
   4) (integer) 1
127.0.0.1:6379> XAUTOCLAIM demo:orders:events emails emails-worker-2 0 0-0 COUNT 10
1) "0-0"
2) 1) 1) "1794650000002-0"
      2) 1) "type"
         2) "order-created"
         3) "orderId"
         4) "1002"
         5) "isbn"
         6) "9780141439518"
         7) "quantity"
         8) "1"
3) (empty array)

Bırakılan girdinin sahibi boş, bekleme süresi -1 oldu; son teslim zamanı sıfırlandığı için herhangi bir minimum bekleme süresiyle hemen devralınabiliyor. Üç modu var: SILENT teslim sayacını bir azaltır (kapanırken bırakılan iş "sayılmasın"), FAIL sayacı olduğu gibi bırakır (bu worker başaramadı, başkası denesin), FATAL sayacı en büyük değere çeker (mesaj kalıcı olarak bozuk). Valkey 9.1.3'te XNACK yok (ERR unknown command), XREADGROUP … CLAIM syntax hatası veriyor; orada XAUTOCLAIM ile ayrı bir toparlama döngüsü kurulur.

Bookstore: Sipariş Olayları ve Idempotent XADD

Bookstore.Api her yeni siparişi bookstore:orders:events stream'ine yazıyor. Ağ koparsa üretici XADD'in sunucuya ulaşıp ulaşmadığını bilemez ve yeniden dener; klasik XADD bu durumda aynı olayı iki kez ekler. Redis 8.6'nın IDMP seçeneği bir üretici kimliği ve bir idempotency kimliği alıyor; aynı ikiliyle gelen ikinci XADD yeni girdi eklemek yerine ilk girdinin ID'sini döndürüyor. Kimlik olarak 03'teki Idempotency-Key header'ını kullanıyoruz:

// XADD bookstore:orders:events IDMP bookstore-api <Idempotency-Key> MAXLEN ~ 100000 * ...
// Aynı idempotency key ile tekrarlanan XADD yeni girdi eklemez, ilk girdinin ID'sini döndürür (Redis 8.6+)
private static Task<RedisValue> PublishOrderCreatedAsync(IDatabase db, CreateOrderRequest request, long orderId, string idempotencyKey) =>
    db.StreamAddAsync(RedisKeys.OrderEvents,
        [
            new("type", "order-created"),
            new("orderId", orderId),
            new("customerId", request.CustomerId),
            new("isbn", request.Isbn),
            new("quantity", request.Quantity)
        ],
        new StreamIdempotentId("bookstore-api", idempotencyKey), maxLength: 100_000, useApproximateMaxLength: true);

dotnet run --project src/Bookstore.RedisLab -- idempotent-xadd

plain attempt 1: 1794655555019-0
plain attempt 2: 1794655555021-0
IDMP  attempt 1: 1794655555022-0
IDMP  attempt 2: 1794655555022-0
XLEN: 3
idmp-duration=100, idmp-maxsize=100, pids-tracked=1, iids-tracked=1, iids-added=1, iids-duplicates=1

Klasik XADD iki girdi oluşturdu, IDMP'li ikinci deneme aynı ID'yi döndürdü. XINFO STREAM'deki idmp-duration=100 ve idmp-maxsize=100, dokümandaki varsayılanlarla aynı: Redis üretici başına son 100 idempotency kimliğini en fazla 100 saniye hatırlıyor; XCFGSET ile 86.400 saniyeye ve 10.000 kimliğe kadar artırılabiliyor. Doküman bu özelliği "at-most-once production" olarak adlandırıyor: üreticinin retry'ı çift girdi üretmez. Ama bu pencere sınırlı; 100 saniyeden sonra gelen bir tekrar yeni girdi olur. Asıl idempotency hâlâ API'de, 03'teki SET … NX ile; IDMP onun arkasındaki ikinci güvence. Valkey 9.1.3'te IDMP yok: komut IDMP'yi ID sanıp Invalid stream ID hatası verdi.

.NET Worker: BackgroundService ile Consumer

Snapshot'a üçüncü bir proje eklendi: src/Bookstore.Worker. İçinde aynı stream'i okuyan iki BackgroundService var: e-posta gönderen emails grubu ve depodan kitap ayıran stock grubu. İkisi de aynı taban class'ı kullanıyor:

protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
    IDatabase db = redis.GetDatabase();
    // Sabit bir ad: restart'tan sonra bu instance yine aynı consumer olur
    string consumer = $"{Group}-{config["Worker:ConsumerName"] ?? "worker-1"}";
    await EnsureGroupAsync(db);

    while (!stoppingToken.IsCancellationRequested)
    {
        // XREADGROUP GROUP g c COUNT 10 CLAIM 30000 STREAMS key >: önce başkalarının beklemedeki girdileri, sonra yeniler (Redis 8.4+)
        StreamEntry[] entries = await db.StreamReadGroupAsync(Stream, Group, consumer, StreamPosition.NewMessages,
            count: 10, noAck: false, claimMinIdleTime: ClaimAfter);
        if (entries.Length == 0)
        {
            await Task.Delay(TimeSpan.FromMilliseconds(500), stoppingToken);   // paylaşılan multiplexer'da BLOCK yok
            continue;
        }

        foreach (StreamEntry entry in entries)
        {
            if (entry.DeliveryCount >= MaxDeliveries)
            {
                await DeadLetterAsync(db, entry);
                continue;
            }
            try
            {
                await HandleAsync(entry, stoppingToken);
                await db.StreamAcknowledgeAsync(Stream, Group, entry.Id);          // XACK
            }
            catch (Exception ex) when (ex is not OperationCanceledException)
            {
                logger.LogWarning("{Group}: {Id} failed ({Error}), released for retry", Group, entry.Id, ex.Message);
                await db.StreamNegativeAcknowledgeAsync(Stream, Group, StreamNackMode.Fail, entry.Id); // XNACK ... FAIL (Redis 8.8+)
            }
        }
    }
}

Üç tasarım kararı var. Consumer adı sabit ve configuration'dan geliyor (Worker:ConsumerName); restart'ta yeni bir rastgele ad üretilseydi eski ad PEL'de sahipsiz girdilerle kalır, XINFO CONSUMERS her restart'ta biraz daha uzardı. XREADGROUP'un BLOCK seçeneği kullanılmıyor: 05'te gördüğümüz gibi StackExchange.Redis'te blocking komutlar paylaşılan bağlantıyı dondurur; boş okumada 500 ms beklemek yeterli. İstersen 09'daki gibi bir pub/sub sinyaliyle bu beklemeyi kısaltabilirsin. Son olarak Redis 8.10'un MAXCOUNT ve MAXSIZE seçenekleri birden çok stream'i tek komutla okurken toplam girdi sayısını ve cevap boyutunu sınırlıyor; tek stream okuyan bu worker'da COUNT 10 yeterli. BackgroundService'in yaşam döngüsü ve graceful shutdown Background Job'lar serisinin ".NET'te IHostedService ve BackgroundService" yazısında.

API ve worker'ı birlikte çalıştırıp iki sipariş verdim, sonra redis-cli ile bozuk bir olay ekledim (quantity two):

dotnet run --project src/Bookstore.Worker
curl -X POST http://localhost:5080/orders -H "Content-Type: application/json" -H "Idempotency-Key: k-10-1" -d '{"customerId":7,"isbn":"9780441013593","quantity":2}'
{"orderId":1,"replayed":false}
curl -X POST http://localhost:5080/orders -H "Content-Type: application/json" -H "Idempotency-Key: k-10-2" -d '{"customerId":7,"isbn":"9780141439518","quantity":1}'
{"orderId":2,"replayed":false}
127.0.0.1:6379> XADD bookstore:orders:events * type order-created orderId 9999 customerId 7 isbn 9780451524935 quantity two
"1794655604137-0"

info: Bookstore.Worker.StockConsumer[0]
      stock: 1794655604072-0 2 x 9780441013593 picked from the warehouse
info: Bookstore.Worker.EmailConsumer[0]
      emails: 1794655604072-0 confirmation e-mail for order 1 to customer 7
info: Bookstore.Worker.EmailConsumer[0]
      emails: 1794655604086-0 confirmation e-mail for order 2 to customer 7
info: Bookstore.Worker.StockConsumer[0]
      stock: 1794655604086-0 1 x 9780141439518 picked from the warehouse
info: Bookstore.Worker.EmailConsumer[0]
      emails: 1794655604137-0 confirmation e-mail for order 9999 to customer 7
warn: Bookstore.Worker.StockConsumer[0]
      stock: 1794655604137-0 failed (The input string 'two' was not in a correct format.), released for retry
warn: Bookstore.Worker.StockConsumer[0]
      stock: 1794655604137-0 failed (The input string 'two' was not in a correct format.), released for retry
warn: Bookstore.Worker.StockConsumer[0]
      stock: 1794655604137-0 failed (The input string 'two' was not in a correct format.), released for retry
fail: Bookstore.Worker.StockConsumer[0]
      stock: 1794655604137-0 moved to bookstore:orders:events:dead as 1794655604237-0 after 3 deliveries

İki grup aynı olayları birbirinden habersiz işledi. Bozuk olay emails grubunda sorun çıkarmadı (o grup miktara bakmıyor), stock grubunda ise üç kez denendi. Her hatada XNACK FAIL olayı hemen geri bıraktı, bir sonraki XREADGROUP … CLAIM onu öncelikli olarak tekrar getirdi; 30 saniye beklemeye gerek kalmadı.

Poison Message ve Dead-Letter Stream

Hiç işlenemeyecek bir olay sonsuza kadar denenirse worker'ı meşgul eder ve PEL'i şişirir. Worker'daki kural basit: teslim sayısı 3'e ulaşınca olay bookstore:orders:events:dead stream'ine kopyalanıyor ve onaylanıyor. 08'deki gibi ikisi tek bir MULTI/EXEC'te; kopyalama yapılıp onay kaybolursa olay iki kez dead-letter'a düşerdi:

// Poison message: dead-letter stream'e kopyala ve onayla, tek bir MULTI/EXEC'te
private async Task DeadLetterAsync(IDatabase db, StreamEntry entry)
{
    NameValueEntry[] fields = [new("sourceId", entry.Id), new("group", Group), new("deliveries", entry.DeliveryCount), .. entry.Values];
    ITransaction tx = db.CreateTransaction();
    Task<RedisValue> added = tx.StreamAddAsync(DeadLetters, fields);
    Task<long> acked = tx.StreamAcknowledgeAsync(Stream, Group, entry.Id);
    await tx.ExecuteAsync();
    logger.LogError("{Group}: {Id} moved to {DeadLetters} as {DeadId} after {Count} deliveries",
        Group, entry.Id, DeadLetters, await added, entry.DeliveryCount);
    await acked;
}

127.0.0.1:6379> XRANGE bookstore:orders:events:dead - +
1) 1) "1794655604237-0"
   2)  1) "sourceId"
       2) "1794655604137-0"
       3) "group"
       4) "stock"
       5) "deliveries"
       6) "3"
       7) "type"
       8) "order-created"
       9) "orderId"
      10) "9999"
      11) "customerId"
      12) "7"
      13) "isbn"
      14) "9780451524935"
      15) "quantity"
      16) "two"

Dead-letter stream'inde olay, nereden geldiği ve kaç kez denendiğiyle duruyor; düzeltilip ana stream'e yeniden eklenebilir. Dead-letter kuyruğunu izleme, inceleme ve yeniden işleme süreci Message Queue'larla Çalışmak serisinin "Dead-Letter Queue Yönetimi: Poison Message'ı Bulmak, İncelemek ve Yeniden İşlemek" yazısında. Bir ayrıntı daha: stream teslimatı at-least-once. Worker bir olayı işleyip XACK'ten hemen önce çökerse olay başka bir worker'da yeniden işlenir; e-posta iki kez gidebilir. Consumer tarafındaki idempotency (inbox tablosu) aynı serinin "Idempotent Consumer ve Inbox" yazısında.

Trimming ve Bellek

Stream okunan girdileri silmediği için sınır koymazsan sonsuza kadar büyür. Bir milyon sipariş olayıyla ölçtüm:

dotnet run --project src/Bookstore.RedisLab -- stream-memory

1,000,000 events: 42,896,854 bytes (42.9 bytes per event)
XTRIM MAXLEN ~ 900000:        removed 9,981, length 990,019 (1 ms)
XTRIM MAXLEN 800000:          removed 190,019, length 800,000 (0 ms)
XTRIM MAXLEN ~ 700000 LIMIT 0: removed 99,956, length 700,044 (0 ms)

Bir milyon olay 41 MB; field adları her düğümde bir kez yazıldığı için girdi başına 43 byte'ın altında. Gerçek olaylar daha büyük değerler taşıdıkça bu sayı büyür. Asıl ilginç olan ilk satır: XADD doküman sayfası ~ ile yaklaşık trimming'in eşiğin "biraz" üstünde girdi bırakabileceğini söylüyor; ölçümde 900.000 hedefine karşı 990.019 girdi kaldı. Nedeni aynı sayfadaki başka bir cümlede: LIMIT verilmezse Redis tek komutta en fazla 100 × stream-node-max-entries (varsayılan 100), yani 10.000 girdiye bakıyor. Her XADD'de MAXLEN ~ kullanırsan bu sınır sorun değil, çünkü stream her eklemede biraz kırpılıyor; ama büyümüş bir stream'i tek XTRIM ~ ile küçültmeye çalışırsan yalnız 10.000'ini siliyorsun. LIMIT 0 sınırı kaldırıyor ve 99.956 girdi silindi. StackExchange.Redis 3.4.35'in StreamTrimAsync'i ise limit: 0'ı istemci tarafında ArgumentOutOfRangeException ile reddetti; bu yüzden o komutu ExecuteAsync ile gönderdim. Kesin trimming (=) de bu boyutta hızlıydı; ama düğümlerin ortasından girdi silmesi gerektiği için her eklemede çağrıldığında yaklaşık olandan pahalıdır.

Uzunluk yerine zamanla sınırlamak için MINID var: ID'ler zaman damgası olduğu için XTRIM key MINID ~ <unix-ms-7-days-ago> son yedi günü tutar. Redis 8.2 ile trimming'e consumer group'ları dikkate alan seçenekler geldi: ACKED yalnız bütün grupların onayladığı girdileri siler; varsayılan KEEPREF ise PEL'de bekleyen bir girdiyi de silebilir, bu durumda worker o girdiyi okuduğunda içerik yerine nil görür.

Üretimde Dikkat

İzlenecek üç sayı var: her grubun XINFO GROUPS'taki lag'i (grubun henüz okumadığı girdi sayısı; worker'ların yetişip yetişmediği), pending'i (okunmuş ama onaylanmamış girdiler; büyüyorsa worker'lar ölüyor ya da hata alıyor) ve dead-letter stream'inin uzunluğu. Yukarıdaki çalıştırmanın sonunda iki grupta da lag ve pending 0 idi. Stream'i tutan Redis'te eviction policy'nin bu key'i silmeyecek şekilde ayarlandığından emin ol; bellek dolunca bir stream'in sessizce silinmesi kayıp mesaj demek (12'de noeviction). Dayanıklılık Redis'in persistence ayarına bağlı: AOF (yazma komutlarının sırayla eklendiği log dosyası) kapalıysa ya da everysec ile çalışıyorsa bir çökme son yazılanları götürebilir; ayrıntısı 12'de.

Ne zaman Streams, ne zaman RabbitMQ ya da Kafka? Zaten Redis işletiyorsan, mesaj hacmi belleğe sığıyorsa ve retention günlerle ölçülüyorsa Streams pratik bir seçim: consumer group, PEL, claim ve replay hazır. Disk tabanlı, aylarca saklanan büyük event log'ları, routing ve dead-letter'ın broker tarafından yönetilmesi, ya da hazır entegrasyon ekosistemi gerekiyorsa Kafka ya da RabbitMQ daha uygun. Karşılaştırmanın tamamı Message Queue'larla Çalışmak serisinin ".NET'te Message Queue ile Çalışmaya Giriş" yazısında.

Sık Düşülen Tuzaklar

XACK'i unutmak ya da her şeyi NOACK ile okumak: birincisinde PEL sonsuza kadar büyür, ikincisinde çöken worker'ın işi kaybolur.

Sınırsız büyüyen stream: okunan girdiler silinmez; MAXLEN ~ ya da MINID koy.

Consumer adını her restart'ta rastgele üretmek: eski adlar PEL'de sahipsiz girdilerle kalır.

Tek bir XTRIM ~ çağrısıyla büyük bir stream'i küçültebileceğini sanmak: varsayılan sınır yüzünden 100.000 yerine 9.981 girdi silindi.

Poison message için sınır koymamak: bozuk bir olay sonsuza kadar denenir.

Redis 8.4–8.8'in CLAIM, IDMP ve XNACK'ini Valkey'de ya da eski bir managed Redis'te varsaymak.

Küçük Bir Deneme

★☆☆ Worker'ı iki farklı Worker:ConsumerName değeriyle iki kez başlat ve 20 sipariş ver. XINFO CONSUMERS bookstore:orders:events emails işin nasıl paylaşıldığını gösteriyor mu?

★★☆ StockConsumer'a geçici bir hata ekle: ilk denemede exception fırlatsın, ikincide başarsın. Log'da olayın kaç kez işlendiğine ve XNACK FAIL yerine SILENT kullanınca teslim sayacının nasıl değiştiğine bak.

★★★ Worker'ı bir siparişi işledikten sonra ama XACK'ten önce öldür (örneğin Environment.FailFast), sonra yeniden başlat. Olay kaç kez işlendi? E-postanın iki kez gitmesini önlemek için bookstore:emails:sent:{orderId} gibi bir key'le SET … NX kullanan bir kontrol ekle.

Bir sonraki yazıda Redis'in JSON tipine ve Query Engine'e geçiyoruz: JSON document'lar, JSONPath ile kısmi güncelleme, FT.CREATE ile index, FT.SEARCH ve FT.AGGREGATE ile katalog araması ve NRedisStack.