Dağıtık Sistemlerde Veri Bütünlüğü: Choreography-Based Saga

Dağıtık sistemlerde veri bütünlüğünü sağlamak, monolitik sistemlere göre daha karmaşık bir meseledir. Bu makalede, dağıtık işlemlerin koordinasyonu için popüler bir yaklaşım olan Choreography-Based Saga desenini ele alacağız. Örnek olarak, basit bir not uygulamasını kullanacağız.

Senaryo

Not uygulamamızda üç temel servis bulunmaktadır: NoteService, CommentService ve AttachmentService. Bir not oluşturulduğunda, NoteService otomatik bir yorum eklemek için CommentService'e ve bir ek (attachment) eklemek için AttachmentService'e başvurur. Eğer bu işlemlerden herhangi biri başarısız olursa, önceki işlemlerin geri alınması gerekmektedir. Örneğimizde her bir servis farklı bir veritabanı ile çalışabilir ve her biri farklı dilde yazılmış bir servis olabilir. Hataya ilişkin tanımladığımız ortak bir event ile tüm servislerin bu durumdan haberdar olmasını ve her birinin kendi içerisinde ilgili veriyi tutarlılığı sağlayacak şekilde event tipine göre aksiyonlarını almasını sağlayacağız. Gerçekleştireceğimiz örnekte tüm servisler .net core ortamında olacağından tek bir hata consumer i üreterek ilerleyeceğiz. Bu örnek, MassTransit ile RabbitMq üzerindeki bir örneği göstermektedir gerçekte, daha fazla konfigürasyon ve özellik (örn. dead letter queues, retry policies) gerekebilir.

Choreography-Based Saga Nedir?

Choreography-Based Saga, her servisin kendi işlemlerini ve tazminat işlemlerini (compensating transactions) bilmesine dayanır. Bir servis, bir işlemi tamamladığında bir event yayınlar. Diğer servisler bu event’i dinler ve kendi işlemlerini gerçekleştirirler. Eğer bir hata oluşursa, bir tazminat işlemi gerçekleştirilir. .Net core üzerinde birkaç kod örneği ile konuyu ele almaya çalışalım.

Örnek Uygulama

1. Event Bus: Bu basit event bus, servisler arasında event yayınlama ve dinleme yeteneği sağlar.
2. Events: NoteCreatedEvent ve NoteCreationFailedEvent olmak üzere iki temel event tanımlıyoruz.
3. NoteService: Bir not oluşturulduğunda, NoteCreatedEvent event'ini yayınlar.
4. CommentService: NoteCreatedEvent event'ini dinler ve otomatik bir yorum ekler. Eğer bir hata oluşursa, NoteCreationFailedEvent event'ini yayınlar.
5. AttachmentService: NoteCreatedEvent event'ini dinler ve bir ek ekler. Eğer bir hata oluşursa, NoteCreationFailedEvent event'ini yayınlar.

Event Tanımlamaları:

public interface INoteCreated
{
    Guid NoteId { get; }
}

public interface INoteCreationFailed
{
    Guid NoteId { get; }
}

Mass Transit Konfigürasyonu

public void ConfigureServices(IServiceCollection services)
{
    services.AddMassTransit(x =>
    {
        x.UsingRabbitMq((context, cfg) =>
        {
            cfg.Host("rabbitmq://localhost");
        });
    });

    services.AddMassTransitHostedService();
}

NoteService

public class NoteController : ControllerBase
{
    private readonly IPublishEndpoint _publishEndpoint;

    public NoteController(IPublishEndpoint publishEndpoint)
    {
        _publishEndpoint = publishEndpoint;
    }

    [HttpPost]
    public async Task<IActionResult> CreateNote()
    {
        var noteId = Guid.NewGuid();
        // Notu kaydet
        // ...

        await _publishEndpoint.Publish<INoteCreated>(new { NoteId = noteId });
        return Ok(new { NoteId = noteId });
    }
}

CommentService

public class NoteCreatedConsumer : IConsumer<INoteCreated>
{
    public async Task Consume(ConsumeContext<INoteCreated> context)
    {
        // Default yorumu ekleyin
        // ...

        // Eğer bir hata oluşursa:
        await context.Publish<INoteCreationFailed>(new { NoteId = context.Message.NoteId });
    }
}

AttachmentService

public class NoteCreatedConsumerForAttachment : IConsumer<INoteCreated>
{
    public async Task Consume(ConsumeContext<INoteCreated> context)
    {
        // Attachment ekleyin
        // ...

        // Eğer bir hata oluşursa:
        await context.Publish<INoteCreationFailed>(new { NoteId = context.Message.NoteId });
    }
}

Bu yaklaşım, servisler arasında sıkı bir koordinasyon olmadan dağıtık işlemlerin koordinasyonunu sağlar. Eğer bir hata oluşursa, tazminat işlemleri gerçekleştirilir, böylece sistemde veri bütünlüğü korunur.

Hata Consumer

public class NoteCreationFailedConsumer : IConsumer<INoteCreationFailed>
{
    private readonly NoteContext _noteContext;
    private readonly CommentContext _commentContext;
    private readonly AttachmentContext _attachmentContext;

    public NoteCreationFailedConsumer(NoteContext noteContext, CommentContext commentContext, AttachmentContext attachmentContext)
    {
        _noteContext = noteContext;
        _commentContext = commentContext;
        _attachmentContext = attachmentContext;
    }

    public async Task Consume(ConsumeContext<INoteCreationFailed> context)
    {
        var noteId = context.Message.NoteId;

        // SQL'den notu sil
        var note = _noteContext.Notes.FirstOrDefault(n => n.Id == noteId);
        if (note != null)
        {
            _noteContext.Notes.Remove(note);
            await _noteContext.SaveChangesAsync();
        }

        // MongoDB'den yorumu sil
        var commentFilter = Builders<Comment>.Filter.Eq(c => c.NoteId, noteId);
        await _commentContext.Comments.DeleteOneAsync(commentFilter);

        // Diğer veritabanından attachment'ı sil
        // Örnek olarak basit bir yapı kullanıldığı için bu kısmı doldurmadım.
        // _attachmentContext ile ilgili silme işlemlerini gerçekleştirin.

        Console.WriteLine($"Note, comment, and attachment with Note ID {noteId} have been deleted due to a failure in the creation process.");
    }
}

Sonuç

Choreography-Based Saga deseni, dağıtık sistemlerde veri bütünlüğünü sağlamak için etkili bir yöntemdir. Bu makalede ele aldığımız basit not uygulaması, bu desenin nasıl uygulanabileceğine dair bir örnektir. Bir sonraki yazımızda Saga patterninde genel olarak uygulanabilecek yöntemlerden bir diğeri olan Orchestration-Based Saga yöntemini incelemeye çalışacağız.