gRPC Streaming: Server, Client ve Bidirectional Streaming ile Backpressure

Sürüm notu: Bu yazıdaki kodlar .NET 10 ve Grpc.AspNetCore 2.85 ile derlenip denenmiştir.

Önce Tahmin Et

Depodaki el terminalleri Inventory'ye abone olup stok değişikliklerini canlı izliyor. Bir terminalin Wi-Fi'ı zayıf, mesajları saniyede ancak 200 tane okuyabiliyor. O sırada bir tedarikçi teslimatı sisteme giriliyor ve birkaç saniye içinde 3000 stok değişikliği oluşuyor.

Server tarafında ne olur? Okunmayan mesajlar bellekte sınırsızca birikir mi? Stoğu güncelleyen kod yavaş terminali bekleyip yavaşlar mı? Yoksa mesajlar sessizce mi düşer? Bir de şunu düşün: terminal okumayı tamamen bıraksa, server ona kaç mesaj gönderebilir?

Contract'a Üç Streaming RPC'si

Unary çağrıda bir istek gider, bir yanıt gelir; ikisi de tek bir HTTP isteğinin parçasıdır. Streaming'de taraflardan biri ya da ikisi, aynı HTTP/2 isteği üzerinde birden çok mesaj gönderir. inventory.proto'ya üç RPC (başka bir makinedeki fonksiyonu yerel gibi çağırma) ekliyoruz:

service InventoryService {
  rpc GetStock (GetStockRequest) returns (GetStockResponse);
  rpc ReserveStock (ReserveStockRequest) returns (ReserveStockResponse);
  rpc ReleaseStock (ReleaseStockRequest) returns (ReleaseStockResponse);

  rpc WatchStock (WatchStockRequest) returns (stream StockChanged);
  rpc BulkUpdateStock (stream StockAdjustment) returns (BulkUpdateSummary);
  rpc ScanBarcodes (stream BarcodeScan) returns (stream ScanResult);
}

message WatchStockRequest {
  repeated string book_ids = 1;
  int64 after_sequence = 2;
}

message StockChanged {
  int64 sequence = 1;
  string book_id = 2;
  int32 available = 3;
  google.protobuf.Timestamp changed_at = 4;
}

message StockAdjustment {
  string book_id = 1;
  int32 delta = 2;
  string reason = 3;
}

message BulkUpdateSummary {
  int32 applied = 1;
  int32 rejected = 2;
  repeated string errors = 3;
}

message BarcodeScan {
  string isbn = 1;
}

message ScanResult {
  string isbn = 1;
  bool found = 2;
  string title = 3;
  bookstore.common.v1.Money price = 4;
  bookstore.common.v1.Money basket_total = 5;
}

Yeni RPC'ler ve mesajlar eklemek, ilk yazıdaki kurallara göre geriye uyumlu: eski client'lar bu method'ları bilmez ama mevcut çağrıları etkilenmez. stream anahtar kelimesinin nerede durduğu türü belirliyor: yalnız dönüşte ise server streaming, yalnız parametrede ise client streaming, ikisinde de ise bidirectional.

Server Streaming: WatchStock

Server streaming method'u bir istek alır ve IServerStreamWriter<T>'a istediği kadar yazar. Method döndüğünde stream biter ve client status'u (normalde OK) alır. Inventory'nin service class'ı artık partial; streaming method'ları ayrı bir dosyada:

public override async Task WatchStock(
    WatchStockRequest request, IServerStreamWriter<StockChanged> responseStream, ServerCallContext context)
{
    HashSet<Guid> books = [.. request.BookIds.Select((id, i) => ParseBookId(id, $"book_ids[{i}]"))];

    StockSubscription subscription;
    try
    {
        subscription = feed.Subscribe(request.AfterSequence);
    }
    catch (HistoryGoneException ex)
    {
        throw new RpcException(new Grpc.Core.Status(StatusCode.OutOfRange,
            $"Sequence {request.AfterSequence} is gone, oldest is {ex.OldestSequence}. Resync with GetStock."));
    }

    using (subscription)
    {
        int written = 0;
        try
        {
            foreach (StockChange change in subscription.Replay.Where(c => books.Count == 0 || books.Contains(c.BookId)))
            {
                await responseStream.WriteAsync(ToMessage(change), context.CancellationToken);
                written++;
            }

            await foreach (StockChange change in subscription.Reader.ReadAllAsync(context.CancellationToken))
            {
                if (books.Count == 0 || books.Contains(change.BookId))
                {
                    // client okumazsa HTTP/2 flow control burada bekletir
                    await responseStream.WriteAsync(ToMessage(change), context.CancellationToken);
                    written++;
                }
            }
        }
        catch (SlowConsumerException ex)
        {
            logger.LogWarning("WatchStock closed after {Written} messages: {Reason}", written, ex.Message);
            throw new RpcException(new Grpc.Core.Status(StatusCode.ResourceExhausted, ex.Message));
        }
    }
}

feed, stok her değiştiğinde haber yayınlayan küçük bir class (StockChangeFeed); birazdan içine bakacağız. Method'un yapısı şu: abone ol, varsa kaçırılan değişiklikleri (Replay) gönder, sonra yenilerini geldikçe yaz. Client bağlantıyı kapatana ya da iptal edene kadar await foreach döner. İptal geldiğinde context.CancellationToken tetiklenir, döngü OperationCanceledException ile çıkar, using aboneliği temizler. Client tarafında bir stream'i durdurmanın tek yolu iptal etmektir; server streaming'de client'ın "artık yeter" diyebileceği bir istek kanalı yok.

Client'ta ResponseStream.ReadAllAsync() stream'i bir IAsyncEnumerable<T>'a çevirir. C# Asenkron serisinde anlattığımız await foreach burada doğrudan çalışıyor. Bir terminal beş saniyeliğine abone olurken başka bir süreç toplu güncelleme gönderdiğinde terminalin gördüğü:

#1 3f2c9a1e available=17
#2 c4a7e2d9 available=38
#3 c4a7e2d9 available=48
received=3 last=#3

Her değişikliğin bir sequence numarası var. Şimdilik sadece sıralı geldiklerini gösteriyor; kopan bir stream'i kaldığı yerden sürdürmek için asıl işi yazının ikinci yarısında yapacak.

Client Streaming: CSV'den Toplu Stok Güncellemesi

Depo sayımından sonra elinde bir CSV dosyası var: kitap, fark ve sebep. Her satır için ayrı bir unary çağrı yerine tek bir client streaming çağrısı açıyoruz; satırlar okundukça gönderiliyor, server sonunda tek bir özet dönüyor:

public override async Task<BulkUpdateSummary> BulkUpdateStock(
    IAsyncStreamReader<StockAdjustment> requestStream, ServerCallContext context)
{
    var summary = new BulkUpdateSummary();

    await foreach (StockAdjustment adjustment in requestStream.ReadAllAsync(context.CancellationToken))
    {
        string? error = Guid.TryParse(adjustment.BookId, out Guid bookId)
            ? store.Adjust(bookId, adjustment.Delta)
            : "invalid book id";

        if (error is null)
        {
            summary.Applied++;
            continue;
        }

        summary.Rejected++;
        if (summary.Errors.Count < 20)
        {
            summary.Errors.Add($"{adjustment.BookId}: {error}");
        }
    }

    return summary;
}

Hatalı bir satır bütün işlemi düşürmüyor; reddedilenleri sayıp ilk yirmisini özete koyuyoruz. Hata listesine sınır koymanın sebebi mesaj boyutu: yüz bin satırlık bir dosyada her satır hatalıysa özet de çok büyük olur. Client:

using (AsyncClientStreamingCall<StockAdjustment, BulkUpdateSummary> bulk = inventory.BulkUpdateStock(
    deadline: DateTime.UtcNow.AddSeconds(30)))
{
    foreach (string line in File.ReadLines(args[1]).Skip(1))
    {
        string[] parts = line.Split(',');
        await bulk.RequestStream.WriteAsync(new StockAdjustment
        {
            BookId = parts[0],
            Delta = int.Parse(parts[1]),
            Reason = parts[2]
        });
    }

    // "başka mesaj yok" sinyali; bunu unutursan server sonsuza kadar bekler
    await bulk.RequestStream.CompleteAsync();

    BulkUpdateSummary summary = await bulk;
    Console.WriteLine($"applied={summary.Applied} rejected={summary.Rejected}");
    foreach (string error in summary.Errors)
    {
        Console.WriteLine($"  {error}");
    }
}

applied=3 rejected=2
  8b1d4e6a-2c3f-4a5b-9e7d-1f0a2b3c4d5e: would go negative (1 -3)
  not-a-guid: invalid book id

CompleteAsync kritik satır. Server'daki await foreach ancak client stream'i kapattığında biter; bu satırı unutursan server döngüde bekler, client da await bulk'ta bekler. Deadline koyduğun için en azından sonsuza kadar sürmez. Bir de File.ReadLines seçimine dikkat: dosyayı satır satır okuyor, yani bir milyon satırlık bir CSV'yi belleğe almadan akıtabilirsin.

Bidirectional Streaming: Kasada Barkod Okutmak

Kasada her barkod okutulduğunda kitabın adı, fiyatı ve sepetin güncel toplamı ekrana gelmeli. Bu, iki yönün birbirinden bağımsız aktığı bir konuşma:

public override async Task ScanBarcodes(
    IAsyncStreamReader<BarcodeScan> requestStream, IServerStreamWriter<ScanResult> responseStream, ServerCallContext context)
{
    decimal basketTotal = 0m;

    await foreach (BarcodeScan scan in requestStream.ReadAllAsync(context.CancellationToken))
    {
        StockItem? item = store.FindByIsbn(scan.Isbn);
        var result = new ScanResult { Isbn = scan.Isbn, Found = item is not null };

        if (item is not null)
        {
            basketTotal += item.UnitPrice;
            result.Title = item.Title;
            result.Price = ToMoney(item.UnitPrice);
        }

        result.BasketTotal = ToMoney(basketTotal);
        await responseStream.WriteAsync(result, context.CancellationToken);
    }
}

Sepet toplamı bir yerel değişkende duruyor; çağrının ömrü boyunca yaşıyor ve başka hiçbir kasayla paylaşılmıyor. Stream, server tarafında doğal bir oturum gibi davranıyor. Client tarafında okuma ve yazmayı ayırmak gerekiyor:

using (AsyncDuplexStreamingCall<BarcodeScan, ScanResult> till = inventory.ScanBarcodes(
    deadline: DateTime.UtcNow.AddMinutes(5)))
{
    // okuma ve yazma ayrı görevlerde: yanıtlar kendi hızında gelir
    Task reader = Task.Run(async () =>
    {
        await foreach (ScanResult result in till.ResponseStream.ReadAllAsync())
        {
            Console.WriteLine(result.Found
                ? $"{result.Isbn} {result.Title,-28} {result.Price.Amount.ToDecimal(),7}  total {result.BasketTotal.Amount.ToDecimal()}"
                : $"{result.Isbn} not found                          total {result.BasketTotal.Amount.ToDecimal()}");
        }
    });

    foreach (string isbn in new[] { "9780000000001", "9780000000003", "9789999999999", "9780000000001" })
    {
        await till.RequestStream.WriteAsync(new BarcodeScan { Isbn = isbn });
        await Task.Delay(300);
    }

    await till.RequestStream.CompleteAsync();
    await reader;
}

9780000000001 Distributed Systems Primer     249.9  total 249.9
9780000000003 Streaming in Practice          320.5  total 570.4
9789999999999 not found                          total 570.4
9780000000001 Distributed Systems Primer     249.9  total 820.3

Okumayı ayrı bir görevde yapmazsan ve "her yazdığıma bir yanıt gelir" varsayımıyla yaz-oku-yaz-oku sırası kurarsan, server'ın bir mesaja sıfır ya da iki yanıt verdiği ilk durumda kilitlenirsin. Bidirectional'da iki yön gerçekten bağımsız; protokol "bir istek, bir yanıt" garanti etmiyor.

Backpressure: Yavaş Consumer Server'ı Nasıl Frenler?

Baştaki ikinci sorudan başlayalım: client hiç okumazsa ne olur? Terminali abone edip dört saniye boyunca tek mesaj okumadan beklettim; bu sırada saniyede 1000 değişiklik üreten bir yük gönderdim. Sonra okumaya başladım:

read after stall: 1385
ResourceExhausted: Subscriber buffer full at sequence 1389.

Server'ın logu da aynı sayıyı söylüyor: WatchStock closed after 1385 messages. Bu 1385 mesajın 256'sı birazdan göreceğimiz kendi buffer'ımızda bekliyordu. Geri kalan yaklaşık 1130 mesaj, her biri 60 byte civarında, toplam 70 KB kadarı, client okumadığı hâlde yola çıkmıştı. Sonra WriteAsync beklemeye başladı. Bunu yapan HTTP/2 flow control: her stream'in ve bağlantının bir alma penceresi (window) var; alıcı okumadıkça pencere dolar ve gönderen durur. gRPC'de WriteAsync'in döndürdüğü Task tam bu yüzden önemli: onu await etmek, karşı tarafın hızına uymak demek.

Bu mekanizma tek bir stream için harika. Ama bizim üreticimiz tek değil: stok değişikliği ReserveStock'tan, toplu güncellemeden, kasadan gelir ve onlarca terminal abonedir. Üretici, en yavaş terminalin WriteAsync'ini bekleyemez; beklerse bir terminalin zayıf Wi-Fi'ı bütün rezervasyonları yavaşlatır. Araya her abone için sınırlı bir Channel<T> koyuyoruz. C# Asenkron serisinin Channels yazısındaki producer/consumer köprüsünün aynısı:

public void Publish(Guid bookId, int available)
{
    lock (_gate)
    {
        var change = new StockChange(++_sequence, bookId, available, DateTimeOffset.UtcNow);
        _history.Enqueue(change);
        if (_history.Count > HistorySize)
        {
            _history.Dequeue();
        }

        foreach (Channel<StockChange> subscriber in _subscribers.ToArray())
        {
            // producer asla beklemez: yetişemeyen subscriber'ın stream'i kapatılır
            if (!subscriber.Writer.TryWrite(change))
            {
                subscriber.Writer.TryComplete(new SlowConsumerException(change.Sequence));
                _subscribers.Remove(subscriber);
            }
        }
    }
}

Abone başına channel Channel.CreateBounded<StockChange>(256) ile, FullMode = BoundedChannelFullMode.Wait olarak oluşturuluyor. Wait modunda TryWrite beklemez, channel doluysa false döner. Burada üç seçenek vardı. Beklemek (üreticiyi en yavaş aboneye bağlar). En eskiyi atmak (DropOldest; stok gibi "son değer önemli" verilerde cazip, ama client hangi mesajı kaçırdığını bilmez). Ya da bizim yaptığımız: aboneliği hata ile kapatmak. Channel'ı bir exception ile tamamlayınca, server'daki await foreach buffer'da kalan mesajları gönderir, sonra exception'ı fırlatır; biz de bunu ResourceExhausted status'una çeviririz. Client kopuşu açıkça görür ve son aldığı sequence'tan devam etmeyi seçebilir.

Baştaki ilk sorunun cevabı da bu: server belleği sınırsız büyümüyor (abone başına 256 mesaj artı ağdaki pencere), stok güncelleyen kod beklemiyor ve hiçbir mesaj sessizce düşmüyor.

Uzun Ömürlü Stream'ler: Keepalive ve Kaldığı Yerden Devam

Bir izleme stream'i saatlerce açık kalır. Bu sürede iki şey onu öldürebilir. Birincisi aradaki ağ cihazlarının boşta bekleme zaman aşımları: nginx'in gRPC proxy'si için grpc_read_timeout varsayılanı 60 saniye ve iki okuma arasındaki süreye uygulanıyor; stok bir dakika değişmezse bağlantı kesilir. İkincisi gerçek ağ kopmaları ve server'ın yeniden başlaması.

Bağlantı düzeyinde HTTP/2 PING frame'leri yardımcı oluyor. Client'ta SocketsHttpHandler ile açılır:

var handler = new SocketsHttpHandler
{
    PooledConnectionIdleTimeout = Timeout.InfiniteTimeSpan,
    KeepAlivePingDelay = TimeSpan.FromSeconds(5), // deneme için kısa
    KeepAlivePingTimeout = TimeSpan.FromSeconds(10),
    EnableMultipleHttp2Connections = true
};
using (GrpcChannel keepAliveChannel = GrpcChannel.ForAddress("http://localhost:5101", new GrpcChannelOptions { HttpHandler = handler }))
{
    await WatchAsync(new InventoryService.InventoryServiceClient(keepAliveChannel), TimeSpan.Zero, TimeSpan.FromSeconds(int.Parse(args[1])));
}

Kestrel'in HTTP/2 loglarını Trace seviyesine alıp boşta duran bir izleme stream'i açtım:

00:28:13 trce: Microsoft.AspNetCore.Server.Kestrel.Http2[37]
      Connection id "0HNQ1PK32GTJP" received HEADERS frame for stream ID 1 with length 217 and flags END_HEADERS.
00:28:19 trce: Microsoft.AspNetCore.Server.Kestrel.Http2[37]
      Connection id "0HNQ1PK32GTJP" received PING frame for stream ID 0 with length 8 and flags NONE.
00:28:26 trce: Microsoft.AspNetCore.Server.Kestrel.Http2[37]
      Connection id "0HNQ1PK32GTJP" received PING frame for stream ID 0 with length 8 and flags NONE.
00:28:32 trce: Microsoft.AspNetCore.Server.Kestrel.Http2[37]
      Connection id "0HNQ1PK32GTJP" received PING frame for stream ID 0 with length 8 and flags NONE.

Boşluk beş saniyeyi geçince PING'ler başlıyor. Production'da 30–60 saniye makul; 5 saniye yalnız deneme içindi. İki uyarı: server PING'leri desteklemiyorsa ilk birkaçını yok sayıp bağlantıyı GOAWAY ile kapatabilir, bu yüzden server'ın desteğini kontrol etmeden açma. Kestrel'de server tarafı için Limits.Http2.KeepAlivePingDelay ve KeepAlivePingTimeout var. Daha önemlisi, PING bağlantıyı canlı tutar ama nginx gibi bir proxy'nin "stream'de veri akmadı" sayacını sıfırlamaz. Seyrek değişen bir akışta uygulama düzeyinde periyodik bir heartbeat mesajı göndermek gerekebilir.

Kopma kaçınılmaz olduğu için client'ın kaldığı yerden devam edebilmesi gerekiyor. sequence numarası ve feed'in son 10.000 değişikliği tutan geçmişi bunun için var. Client son gördüğü sequence ile yeniden abone olur; server aradaki değişiklikleri önce Replay ile gönderir:

while (!stop.IsCancellationRequested)
{
    try
    {
        using AsyncServerStreamingCall<StockChanged> call = inventory.WatchStock(
            new WatchStockRequest { AfterSequence = lastSequence },
            cancellationToken: stop.Token);

        await foreach (StockChanged change in call.ResponseStream.ReadAllAsync(stop.Token))
        {
            if (lastSequence > 0 && change.Sequence != lastSequence + 1)
            {
                Console.WriteLine($"gap: {lastSequence} -> {change.Sequence}");
            }

            lastSequence = change.Sequence;
            backoff = TimeSpan.FromMilliseconds(250);
            if (++received <= 3 || readDelay == TimeSpan.Zero)
            {
                Console.WriteLine($"#{change.Sequence} {change.BookId[..8]} available={change.Available}");
            }

            await Task.Delay(readDelay);
        }
    }
    catch (RpcException ex) when (ex.StatusCode is StatusCode.ResourceExhausted or StatusCode.Unavailable)
    {
        // kaldığı yerden devam: son görülen sequence ile yeniden abone ol
        Console.WriteLine($"{ex.StatusCode}: {ex.Status.Detail} resume after #{lastSequence}");

        // server düştüyse sıkı döngüde yeniden bağlanma: bekle, her denemede iki katına çıkar
        await Task.Delay(backoff, stop.Token).ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
        backoff = TimeSpan.FromTicks(Math.Min(backoff.Ticks * 2, TimeSpan.FromSeconds(10).Ticks));
        continue;
    }
    catch (RpcException ex) when (ex.StatusCode == StatusCode.OutOfRange)
    {
        // geçmiş penceresi kaçtı: güncel durumu baştan oku, canlı akıştan devam et
        GetStockResponse snapshot = await inventory.GetStockAsync(
            new GetStockRequest { BookId = ThirdBook }, deadline: DateTime.UtcNow.AddSeconds(2));
        Console.WriteLine($"{ex.StatusCode}: resync, {ThirdBook[..8]} available={snapshot.Available}");
        lastSequence = 0;
    }
    catch (RpcException ex) when (ex.StatusCode == StatusCode.Cancelled)
    {
        break;
    }
    catch (OperationCanceledException)
    {
        break;
    }
}

Önce baştaki senaryo: mesaj başına 5 ms harcayan yavaş bir terminal ve hızla gelen 3000 değişiklik.

#2004 c4a7e2d9 available=47
#2005 c4a7e2d9 available=48
#2006 c4a7e2d9 available=47
ResourceExhausted: Subscriber buffer full at sequence 3384. resume after #3383
received=3000 last=#5003

Terminal geride kaldı, server stream'i ResourceExhausted ile kapattı, terminal 3383'ten yeniden abone oldu ve geri kalanı replay ile aldı. Sonuç 3000 değişikliğin 3000'i, hiçbir "gap" satırı yok. Aynı terminale bu kez 15.000 değişiklik gönderdim:

#5004 c4a7e2d9 available=47
#5005 c4a7e2d9 available=48
#5006 c4a7e2d9 available=47
ResourceExhausted: Subscriber buffer full at sequence 6402. resume after #6401
OutOfRange: resync, c4a7e2d9 available=48
received=1398 last=#0

Terminal yeniden bağlandığında 6401 artık geçmişte yoktu (en eski kayıt 10004'tü). Server OutOfRange döndü, terminal güncel stoğu GetStock ile baştan okuyup canlı akıştan devam etti. Geçmişin ne kadar tutulacağı bir tasarım kararı. Penceren ne olursa olsun bu "resync" yolu mutlaka olmalı.

Yeniden bağlanırken beklemeyi de unutma. Bu döngünün ilk hâlinde backoff yoktu. Server'ı başlatmadan testi çalıştırdığımda client saniyede binlerce kez Unavailable: Error connecting to subchannel. yazdırıp 40 MB'lık bir çıktı üretti. Server düştüğünde bütün terminallerin aynı anda sıkı döngüde bağlanmaya çalışması, server'ı ayağa kalktığı anda yeniden düşürmenin en kısa yolu.

Ne Zaman Streaming Kullanmamalı?

Tarayıcıya canlı veri. Tarayıcı gRPC'nin streaming'ini doğrudan konuşamıyor; gRPC-Web ile yalnız server streaming mümkün ve o da ek altyapı ister. Sipariş durumu ya da stok rozeti gibi tarayıcıya giden canlı veri için Real-time Uygulamalar serisinin ilk yazısında karşılaştırdığımız SSE (sunucudan istemciye tek yönlü event akışı) ya da SignalR daha doğal.

Büyük dosya. gRPC mesaj tabanlı; her mesaj bütün olarak belleğe alınır. Bir PDF'i ya da kapak görselini 64 KB'lık parçalara bölüp stream etmek mümkün, ama ASP.NET Core'da düz bir HTTP endpoint'i request ve response gövdesini doğrudan stream olarak verir. Dosya için çoğu zaman o daha basit.

"Performans için her şeyi stream'e çevirmek." Açık bir bidirectional stream üzerinde mesaj göndermek, her seferinde yeni bir unary çağrı açmaktan hızlı. Ama yük dengeleme stream'i bölemez, hata yönetimi ve yeniden bağlanma senin sorumluluğuna geçer. Bunu ancak ölçümle gRPC'nin darboğaz olduğunu gördüğünde düşün.

Üretimde Dikkat

Bir stream'e aynı anda tek WriteAsync. Stream writer'ları thread-safe değil. Birden çok kaynaktan aynı stream'e yazacaksan araya bir Channel<T> koy ve tek bir döngüden yaz; WatchStock'ta yaptığımız tam olarak bu.

Uzun stream'ler bağlantıdaki stream kotasını tüketir. Çoğu server bir HTTP/2 bağlantısında aynı anda 100 stream'e izin verir; 100 terminalin izleme stream'i tek bir channel'dan açılırsa 101. çağrı (bir GetStock bile) sırada bekler. EnableMultipleHttp2Connections bu durumda ek bağlantı açtırır. Load balancing tarafını bir sonraki yazıda ele alıyoruz.

Streaming çağrılarını mutlaka dispose et. Yarım bırakılan bir çağrı yalnız client'ta değil server'da da çalışmaya devam eder. Örneklerdeki using blokları bu yüzden var.

Deadline'ı stream'in doğasına göre seç. Toplu güncelleme ve kasa oturumu gibi sonu belli stream'lere deadline koy. Saatlerce açık kalacak bir izleme stream'inde kısa bir deadline anlamsız; orada iptal token'ı, keepalive ve yeniden bağlanma döngüsü işi görür.

Büyük mesajlarda pencere boyutuna bak. Kestrel'in varsayılan stream penceresi 768 KB; bundan büyük mesajları sıkça alan bir service'te veri kesik kesik akar. InitialStreamWindowSize ve InitialConnectionWindowSize ile büyütülebilir, ama her bağlantı için daha fazla bellek demek.

Sık Düşülen Tuzaklar

CompleteAsync'i unutmak. Client streaming'de server "son mesaj geldi mi" diye bekler, client yanıtı bekler; ikisi de birbirini bekler.

Bidirectional'da yaz-oku-yaz-oku. Okuma ayrı bir görevde değilse, bir mesajın yanıtsız kaldığı ya da iki yanıt ürettiği ilk anda kilitlenme.

Sınırsız buffer. Channel.CreateUnbounded ya da bir List<T> ile "nasılsa okur" demek, yavaş bir aboneyi server için bir bellek sızıntısına çevirir.

WriteAsync'i await etmemek. Fire-and-forget yazınca flow control'ün frenini devre dışı bırakırsın, üstüne eşzamanlı yazma exception'ı alırsın.

Kopmayı istisna saymak. Uzun ömürlü her stream bir gün kopar. Sequence numarası, replay ve resync yolu olmayan bir izleme akışı, ilk ağ dalgalanmasında sessizce eksik veri gösterir.

Küçük Bir Deneme

★☆☆ Client streaming örneğinde CompleteAsync satırını kaldır ve deadline'ı 5 saniyeye indir. İki tarafta neler olduğunu loglardan takip et.

★★☆ StockChangeFeed'i BoundedChannelFullMode.DropOldest ile değiştir ve yavaş terminal senaryosunu tekrar çalıştır. Client'taki "gap" satırlarını say ve hangi durumda bunun kabul edilebilir olduğunu tartış.

★★★ StockChanged mesajını bir oneof ile Heartbeat mesajı da taşıyacak şekilde genişlet. Server 20 saniye değişiklik olmadığında heartbeat göndersin, client 45 saniye hiç mesaj alamazsa bağlantıyı kendisi kapatıp yeniden abone olsun.

Inventory artık tek bir instance olarak çalışıyor. Bir sonraki yazıda production'a çıkıyoruz: interceptor'lar, JWT ile authentication, HTTP/2'nin neden klasik load balancer'larla anlaşamadığı, retry ve health check.