Idempotent Consumer: Tekrar Mesaj Engelleme

Idempotent Consumer: Dağıtık Sistemlerde Tekrar Mesajları Güvenle İşlemek

Mesajlaşma sistemlerinde (RabbitMQ, Azure Service Bus, Kafka), mesajların en az bir kez (at-least-once) teslim edilmesi yaygın bir garantidir. Bu, bir mesajın ağ hatası, broker yeniden başlatması veya tüketici (consumer) hatası nedeniyle birden fazla kez iletilebileceği anlamına gelir. Eğer tüketici bu mesajı her seferinde aynı şekilde işlerse, sistemde istenmeyen yan etkiler (örneğin, aynı siparişin iki kez oluşturulması, aynı e-postanın iki kez gönderilmesi) meydana gelebilir.

Idempotent Consumer (Yinelenebilir Tüketici) deseni, bu sorunu çözmek için tasarlanmıştır. Bir işlemin birden fazla kez yapılmasının, tek bir kez yapılmasıyla aynı sonucu doğurmasını (yan etkisiz olmasını) sağlar. Bu yazıda, idempotency'nin ne olduğunu, neden gerekli olduğunu, deduplication stratejilerini, idempotency key kullanımını ve .NET'te bu desenin nasıl uygulanacağını detaylıca ele alacağız.

1. Neden Idempotency? At-Least-Once Teslimatı Anlamak

Mesajlaşma sistemleri, mesaj kaybını önlemek için genellikle at-least-once (en az bir kez) teslimat garantisi sunar. Bu, bir mesajın en az bir kez tüketiciye ulaştırılmasının garanti edildiği anlamına gelir. Ancak bu, aşağıdaki durumlarda mesajın birden fazla kez iletilebileceği anlamına gelir:

Tüketici (Consumer) Başarısız: Tüketici mesajı aldı, işledi, ancak broker'a onay (ack) gönderemeden çöktü. Broker, mesajı tekrar gönderir.
Ağ Zaman Aşımı (Network Timeout): Tüketici mesajı işledi ve onay gönderdi, ancak onay broker'a ulaşamadı (ağ kopması). Broker, onay alamadığı için mesajı yeniden gönderir.
Broker Yeniden Başlatması: Broker yeniden başlatıldığında, bazı mesajlar yeniden teslim edilebilir.
Consumer Group Rebalancing: Kafka gibi sistemlerde, partition'lar tüketiciler arasında yeniden dağıtıldığında (rebalance), bazı mesajlar yeniden işlenebilir.

Eğer consumer idempotent değilse, bu tekrarlar veri bozukluğuna, çift işleme ve sistem tutarsızlığına yol açar.

2. Idempotency Nedir?

Bir işlemin idempotent olması, aynı girdiyle (mesaj) birden fazla kez çağrıldığında, yan etkilerinin (side effects) ilk çağrıdakiyle aynı olması ve sistem durumunda istenmeyen bir değişiklik yaratmamasıdır.

Örnekler:

Idempotent: "Kullanıcıyı oluştur" (aynı kullanıcı tekrar oluşturulmaz, var olan döndürülür).
Non-Idempotent: "Kullanıcının bakiyesini 10 TL artır" (her çağrıda bakiye 10 TL artar, bu istenmeyen bir durumdur).

3. İdempotent Consumer Oluşturmanın Yolları

İdempotent bir consumer oluşturmanın üç ana yöntemi vardır:

A. Doğal Olarak İdempotent İşlemler
Bazı işlemler zaten idempotenttir.

Veritabanı UPSERT (MERGE): Eğer aynı veriyi tekrar eklemeye çalışırsanız, mevcut kayıt güncellenir veya eklenmez.
Silme (DELETE) İşlemi: Aynı kaydı iki kez silmek, ilk silmeden sonra hiçbir şey değiştirmez.
Öncelikli Güncellemeler: SET last_login = NOW() gibi mutlak değer atamaları.

B. Deduplication (Veri Tekrarını Önleme)
Mesajın benzersiz bir kimliğini (ID) kullanarak, daha önce işlenen mesajları bir depoda (cache, veritabanı) izlemek.

C. Idempotency Key (Yinelenebilirlik Anahtarı)
Üretici (producer), her mesaja benzersiz bir Idempotency-Key ekler. Tüketici, bu anahtarı kullanarak mesajın daha önce işlenip işlenmediğini kontrol eder.

4. Deduplication Stratejileri

A. Veritabanı Tabanlı Deduplication (En Güvenilir)

Mesaj işlendikten sonra, mesajın benzersiz kimliğini (veya işlemin sonucunu) aynı transaction içinde bir "processed messages" tablosuna kaydedin. Aynı mesaj tekrar gelirse, bu tabloda sorgulama yapıp zaten işlenmişse atlayın.

csharp

// ProcessedMessages tablosu
public class ProcessedMessage
{
    public string MessageId { get; set; } // Benzersiz mesaj ID'si
    public DateTime ProcessedAt { get; set; }
    public string Result { get; set; } // İsteğe bağlı, sonucu sakla
}

// Consumer (Deduplication ile)
public class OrderConsumer : IConsumer<OrderCreated>
{
    private readonly AppDbContext _dbContext;

    public async Task Consume(ConsumeContext<OrderCreated> context)
    {
        var messageId = context.MessageId.ToString(); // MassTransit'de MessageId var

        // 1. Mesaj daha önce işlenmiş mi kontrol et
        bool alreadyProcessed = await _dbContext.ProcessedMessages
            .AnyAsync(pm => pm.MessageId == messageId);

        if (alreadyProcessed)
        {
            // Mesajı onayla (ack) ve işleme
            await context.ConsumeCompleted(); // veya context.CompleteTask()
            return;
        }

        // 2. İş mantığını çalıştır
        var order = new Order { Id = context.Message.OrderId, ... };
        _dbContext.Orders.Add(order);

        // 3. Mesajı işlenmiş olarak işaretle (Aynı Transaction!)
        var processed = new ProcessedMessage
        {
            MessageId = messageId,
            ProcessedAt = DateTime.UtcNow
        };
        _dbContext.ProcessedMessages.Add(processed);

        // 4. Tek bir transaction ile kaydet
        await _dbContext.SaveChangesAsync();
    }
}

Artıları: En güvenli yöntemdir. ACID transaction ile tam tutarlılık sağlar.
Eksileri: Her mesaj için bir veritabanı sorgusu ek yük getirir. Yüksek hacimli sistemlerde performans düşüşüne neden olabilir.

B. Dağıtık Cache (Redis) ile Deduplication

İşlenen mesaj ID'lerini bir dağıtık cache'de (Redis) geçici olarak saklayın.

csharp

public class OrderConsumer : IConsumer<OrderCreated>
{
    private readonly IDistributedCache _cache;
    private readonly AppDbContext _dbContext;

    public async Task Consume(ConsumeContext<OrderCreated> context)
    {
        var messageId = context.MessageId.ToString();
        var cacheKey = $"processed:{messageId}";

        // 1. Cache'te var mı kontrol et
        var cached = await _cache.GetStringAsync(cacheKey);
        if (cached != null)
        {
            await context.ConsumeCompleted();
            return;
        }

        // 2. İş mantığını çalıştır
        // ...

        // 3. Cache'e ekle (TTL ile)
        var options = new DistributedCacheEntryOptions
        {
            AbsoluteExpirationRelativeToNow = TimeSpan.FromHours(24)
        };
        await _cache.SetStringAsync(cacheKey, "processed", options);

        await _dbContext.SaveChangesAsync();
    }
}

Artıları: Veritabanından daha hızlıdır, yüksek performans sağlar.
Eksileri: Cache temizlenirse veya tutarsız olursa, aynı mesaj yeniden işlenebilir. Kalıcı (persistent) değildir.

C. Idempotency Key (Kafka / HTTP) ile

Bu yöntem özellikle Kafka veya REST API'lerinde yaygındır. Üretici, her mesaja benzersiz bir Idempotency-Key ekler. Tüketici, bu anahtarı kullanarak mesajın işlenip işlenmediğini kontrol eder.

csharp

// Üretici (Producer)
var headers = new Headers();
headers.Add("Idempotency-Key", Encoding.UTF8.GetBytes(Guid.NewGuid().ToString()));

await producer.ProduceAsync("order-topic", new Message<string, string>
{
    Key = "order-123",
    Value = JsonSerializer.Serialize(order),
    Headers = headers
});

// Tüketici (Consumer)
public Task Consume(ConsumeContext<OrderCreated> context)
{
    var idempotencyKey = context.Headers.Get<string>("Idempotency-Key");
    // ... idempotencyKey'i kullanarak deduplication yap
}

MassTransit'de MessageId zaten bu işlevi görür. ConsumeContext.MessageId property'si her mesaj için otomatik olarak benzersiz bir ID sağlar.

5. MassTransit ile Idempotent Consumer

MassTransit, idempotency'yi destekleyen bir Consumer Middleware sunar.

Adım 1: Idempotency Cache / Repository Oluşturma

csharp

// IIdempotentConsumerRepository implementasyonu
public class EfCoreIdempotentConsumerRepository : IIdempotentConsumerRepository
{
    private readonly AppDbContext _context;

    public async Task<bool> IsMessageProcessedAsync(Guid messageId, CancellationToken cancellationToken)
    {
        return await _context.ProcessedMessages
            .AnyAsync(pm => pm.MessageId == messageId.ToString(), cancellationToken);
    }

    public async Task MarkMessageAsProcessedAsync(Guid messageId, CancellationToken cancellationToken)
    {
        _context.ProcessedMessages.Add(new ProcessedMessage
        {
            MessageId = messageId.ToString(),
            ProcessedAt = DateTime.UtcNow
        });
        await _context.SaveChangesAsync(cancellationToken);
    }
}

Adım 2: MassTransit Konfigürasyonuna Ekleme

csharp

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<OrderConsumer>();
    
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("localhost", "/", h => { ... });

        cfg.ReceiveEndpoint("order-queue", e =>
        {
            // Idempotency middleware'ini ekle
            e.UseIdempotentConsumer<OrderConsumer, OrderCreated>(
                new EfCoreIdempotentConsumerRepository()
            );
            
            e.ConfigureConsumer<OrderConsumer>(context);
        });
    });
});

6. MassTransit'de Idempotent Consumer Middleware

MassTransit, UseIdempotentConsumer extension metodu ile bu işlemi otomatikleştirir. Bu middleware, mesajı işlemeden önce repository'de kontrol eder, işlenmişse atlar, işlenmemişse işler ve sonrasında işlendi olarak işaretler.

csharp

// MassTransit içinde Idempotent Consumer Middleware kullanımı
public class OrderConsumer : IConsumer<OrderCreated>
{
    public async Task Consume(ConsumeContext<OrderCreated> context)
    {
        // Normal iş mantığı (idempotency middleware zaten kontrol ediyor)
        var order = new Order { OrderId = context.Message.OrderId };
        // ...
    }
}

7. En İyi Pratikler

Idempotency Key'i Seçin: Benzersiz ve üretici tarafından üretilen bir anahtar (ör. MessageId, TransactionId). Aynı işlemi temsil eden tüm mesajlar aynı anahtara sahip olmalıdır.
Kalıcı (Persistent) Depo Kullanın: Veritabanı, deduplication için en güvenilir yöntemdir. Cache, geçici bir çözüm olarak kullanılabilir.
Transaction ile Birlikte Kullanın: İşlemi ve deduplication kaydını aynı transaction içinde yaparak veri tutarlılığını koruyun.
Mesaj TTL'sini Ayarlayın: İşlenen mesaj kayıtlarını sonsuza kadar saklamak yerine, belirli bir süre sonra temizleyin (ör. 7 gün).
Hata Yönetimi: Idempotency deposuna erişim hatası durumunda (ör. veritabanı bağlantısı koptu), mesajın işlenmesini engellemek veya yeniden denemek için uygun bir strateji belirleyin.

Sonuç:

Idempotent Consumer deseni, dağıtık sistemlerde mesaj tekrarlarının (duplicate messages) neden olduğu veri bozukluğunu ve istenmeyen yan etkileri önlemenin en temel ve en etkili yoludur. At-least-once teslimatı garanti eden sistemlerde, idempotency olmazsa olmazdır.

Deduplication (veritabanı veya cache tabanlı) ve Idempotency Key kullanımı, en yaygın stratejilerdir. MassTransit, UseIdempotentConsumer middleware'i ile bu deseni uygulamayı oldukça kolaylaştırır.

Unutmayın: Her mesajı sadece bir kez işlemek (exactly-once) garanti edilemez, ancak idempotent bir consumer ile mesaj tekrarları zararsız hale getirilebilir. Bu, dağıtık sistemlerin dayanıklılığını (resilience) sağlayan en önemli adımlardan biridir.

İlgili Yazılar