CQRS Örneği | Stok Yönetimi
Daha önceki yazımda CQRS (Command Query Responsibility Segregation), nihai tutarlılık (eventual consistency) gibi terimler üzerinde durmuştum. Bu yazımda ise .Net Core 3 üzerinde bir örnek oluşturacağım.
Uygulamamız olay güdümlü programlama tekniğine (event driven architecture) uygun geliştirilmiştir. Eğer bu tekniğe aşina değilseniz, kısa olan şu yazıma bir göz atabilirsiniz.
Kodlara GitHub üzerinden erişebilirsiniz.
Başlamadan Önce
DDD ve Microservices konsepti kullanılarak Ürün ve Stok ayrı domainler olarak bağımsız şekillerde geliştirilecektir. Biz Stok Yönetimi kısmına odaklanacağız.
Ürün domaini var olmuş olsaydı, ürün domaininde oluşan varlıkların belli başlı verileri otomatik olarak Stok domainine yansıtılacaktı. Ürün domaini olmadığı için ürünleri canımız istedikçe HTTP isteği ile Stok domaini altında manuel yaratacağız. Ürünler bizim asıl ilgi alanımız olmadığı ve üzerinde pek aksiyon almadığımız için CQRS konusuna dahil olmayacaktır.
Stok Yönetimi
Bu uygulamanın karşılayacağı ihtiyaçlar şunlardır:
- Sıfır stok ile sistemde yeni bir ürün yaratmak
- Ürün listesini sorgulamak
- Bir ürün için stok sayısını arttırmak
- Bir ürün için stok sayısını azaltmak
- Bir ürün için stok sayısını sıfırlamak
- Bir ürünün stok değerini sorgulamak
Tüm bu durumlar üzerinde tek tek durmayıp genel resmin anlaşılabilmesi için belli başlı isterlerin üzerinden geçeceğiz.
Geliştirmemizde yer alacak ana modeller şunlardır: Ürün (ProductModel), Stok Aksiyonu (StockActionModel) ve Stok (StockModel).
- Ürün: Üzerinde ürün kodu gibi bilgiler saklanacaktır. Okuma ve yazma işlemi aynı model üzerinde yapılmaktadır.
- Stok Aksiyonu: Yazma modelimizi göstermektedir ve stoğa ne kadar ürün giriş-çıkışı olduğuna dair bilgileri sağlamaktadır.
- Stok: Okuma modelimiz olacaktır. Üzerinde ürüne ait bilgiler ve stokta kalan miktara dair veriler bir arada yer almaktadır.
Ayrıca kodumuzun içinde Stok Durumu (StockSnapshotModel) adında bir sınıf yer almaktadır. Stok üzerinde alınan aksiyonların birleştirilmesi ile elde edilen sonucu saklamaktadır. Böyle bir sınıfa sahip olmasaydık stoktan ürün çıkarırken stok üzerinde yapılan tüm işlemleri birleştirmemiz gerekecekti. İşlemlerin birleşiminin sonucunda istenen ürün sayısını karşılayabilecek kadar stoğumuzun olup olmadığını anlayabilecektik. Bu model dışarıdan gözlenemeyecektir. Domain içerisinde işlerimizi kolaylaştırmak için oluşturduğumuz bir sınıftır.
Şekilde görüleceği üzere ürün ile ilgili olan iş emirleri, ürün ile ilgili olan sorgulama istekleri ve stok ile ilgili olan iş emirleri aynı veri tabanı üzerinde olacak şekilde kurgulanmıştır. Stok üzerine yapılacak sorgulamalar ise farklı veri tabanı üzerinde gerçekleştirilebilir.
- Ürün Oluşturulması
Ürün bilgisinin uygulamamız içerisinde kaydedilmesi ile başlayan bir sürece sahibiz. Ürün kodu ile kayıt işlemini başlattığımız zaman domain içerisinde Ürün Yaratıldı adında bir domain olayı ortaya çıkmaktadır. Domain olayını izleyen gözlemcilerden biri Stoğu İlkle etiketiyle Stok Aksiyonu kaydını oluştur. Stok İlklendi adında başka bir domain olayı bu şekilde ortaya çıkar. Bu domain olayını gözlemleyen gözlemci ise Stok Durumu kaydını sıfır değeri ile oluşturur. Bu sürecin sonunda Stok İlklendi durumuna ait entegrasyon olayı RabbitMq üzerinden tüm sisteme yayımlanır.
Stok İlklendi olayını gözleyen bir gözlemleyici, Stok sınıfına ait kaydı oluşturacaktır.
- Ürünün Stok Sayısının Arttırılması ve Azaltılması
Ürüne ait stok sayısının arttırılması için gönderilecek istek içerisinde ProductId, Count, CorrelationId gibi veriler yer almaktadır. Benzer istek modeliyle stok sayısının düşürülmesi için de talep gönderilebilir. Hangi ürün için ne kadarlık bir stok değişikliği yapılacağına dair isteğin yapısı makul görünmektedir. Bu işlemler, yazma modeli yani Stok Aksiyonu ve Stok Durumu sınıfları üzerinden kontroller yapılarak gerçekleştirilmektedir.
- Stok Sorgulaması
Stok üzerinde yürüttüğümüz işlemler tamamlandığında Stok Sayısı Arttı veya Stok Sayısı Azaldı mesajı RabbitMq üzerine gönderilmektedir. Mesaj RabbitMq ile gözlemleyiciye ulaştığı zaman Stok güncellenmektedir.
Stok sorgulama işleminin sonucunda ürün kodu, kalan stok sayısı, stoğun son güncellenme zamanı gibi verilere erişebilmekteyiz. CQRS yapısının artılarından biri işte bu durumdur. Yazma isteği olarak farklı model kullanırken okuma isteğinin sonucunda farklı bir modelle karşılaşmaktayız. Her iki varlık da kendi bağlamları için anlamlı modellerdir.
Stok verisinin senkronize olması için bir müddet vakit gerekmektedir. Bu sebeple sorgulama sonucunuzda stoğa sahip olduğunu düşündüğünüz bir ürün için stoktan düşme emri verdiğiniz zaman “stok bulunamadı” hatası alınabilmektedir. Eninde sonunda tutarlı sistemlerde buna benzer durumlarla karşılaşabilmekteyiz. Bir süre bekleyip stok değerini sorgularsak stok değerinin gerçek değerine ulaştığını görebiliriz.
Eninde sonunda tutarlı olacak bir sistem yerine stok üzerinde yürütülen aksiyonları doğrudan stok okuma modeline yansıtabilirdik. Böyle bir kurguda okuma işlemini yürütmek isteyen kişiler, o an yürütülen yazma istekleri varsa yazma işlemlerinin tamamlanmasını bekleyecekti. E-Ticaret sitelerini gözümüzün önünde canlandıralım. Bir ürüne bakılmak istendiğinde stok verisi de ekrana yansıtılmaktadır. Bir ürüne bakılmak istendiği zaman sepetine o ürünü ekleyen biri varsa, diğer kullanıcıların ürünün sepete eklenmesi işlemini sona ermesini beklemesi memnuniyetsizlik sebebi olacaktı. Biz de bu sebeple stok üzerindeki okuma modelimizi eninde sonunda tutarlı olacak şekilde tasarladık.
Kullanılan Araç ve Kütüphaneler
Bu noktaya kadar CQRS yazısında anlatılanlara ait bir örnek yapı kurguladık. Kurgumuzun isterleri ve çalışma şekli hakkında da konuştuk. Çalışma prensibini bildiğimiz kurgumuzu gerçeklerken, kullandığımız kütüphaneleri tanıyarak nasıl bir implementasyon yaptığımızı anlattığımız kısma geldik.
- MediatR: Uygulama içinde yer alan servislerin birbiri ile emirler ve olaylar üzerinden haberleşmesini sağlamak için kullanılmıştır.
- MassTransit: Mesajlaşma sistemine (RabbitMq) erişim için kullanılmıştır. Entegrasyon olayları bu kütüphane aracılığı ile mesajlaşma sistemine bildirilmiş ve mesajlaşma sisteminden öğrenilmiştir.
- EntityFrameworkCore: Verilerin saklanacağı ortam (Sql Server) ile haberleşmek için kullanılmıştır.
- DistirbutedLock: Yürütülen aksiyonlarda belirlenen kritik bölgelere giriş çıkışları denetlemek için kullanılmıştır.
- docker: Platform bağımlılıklarının (SqlServer ve RabbitMq) kolayca yönetilebilmesi için kullanılmıştır.
Konumuzla yakından ilintili olan MediatR ve MassTransit kütüphanelerini ise aşağıda daha detaylı olarak inceliyor olacağız.
- MediatR
Uygulamamızın yapısı olay güdümlü programlamaya uygun geliştirilmiştir. Bu sebeple uygulamamız içerisinde …Command ve …Event notasyonuna sahip bol miktarda sınıf görebiliriz. Emirler ve olayların dağıtımı sırasında MediatR kütüphanesinden faydalanmaktayız.
MediatR, mediator tasarım desenini implemente eden ve açık kaynak kodlu olarak kullanıma açılmış bir kütüphanedir. Mediator, nesnelerin birbiri ile olan ilişkisini yöneten objeye verilen isimdir ve tasarım deseninin adı da buradan gelmektedir. Nesneler arasındaki iletişimin mediator üzerinden gerçekleşmesi sayesinde zayıf bağlı sınıf ilişkileri (loosly coupling) elde ediyoruz.
Stok Yönetimi sırasında iş emirlerini ve olayları mediator nesnemiz aracılığı ile sisteme bildirerek düşük bağımlılığa sahip sınıflar geliştiriyoruz. Ürün yaratma süreci üzerinden MediatR kütüphanesinin nasıl kullanıldığını gözlemleyelim.
Öncelikle iş emrimizi, IRequest<T> arayüzünden türetiyoruz. Bu arayüz MediatR kütüphanesinden gelmektedir. <T> dönüş tipimizi göstermektedir. Bizim durumumuzda CreateProductCommand iş emri yürütülecektir. Ardından da geriye ProductResponse sınıfından bir obje dönülecektir. Bu yapıyı sağlayacak kod parçası aşağıdaki gibidir.
Not: MediatR içerisinde IRequest arayüzü de yer almaktadır. Bu arayüz dönüş tipi void olan emirler için kullanılmaktadır.
Şimdi emrimizi yerine getirecek sınıfımızı hazırlamamız gerekmektedir. Ürünler üzerinde yapılacak işlemleri bir araya toparlamak için aşağıdaki gibi IProductService adında bir arayüz hazırlayalım.
IProductService arayüzünü implemente eden ProductService sınıfını tasarladığımız zaman CreateProductCommand sınıfının bir örneğini parametre olarak alan, ProductResponse sınıfının bir örneğini sonuç olarak dönen Handle adında bir metod ortaya çıkacaktır.
İş kararlarımıza bağlı olarak development işlemlerimizi yukarıda göreceğiniz metodun içeresinde ele alıyoruz. İlk satırlarında request objesinden ürün modelini hazırlıyoruz ve veri tabanımıza kaydediyoruz. Bu noktada EntityFramework kütüphanesinden faydalanıyoruz.
await _dataContext.SaveChanges(cancellationToken); satırından sonra yapacağımız işlem ise yürüttüğümüz aksiyonun sonucunu domain içerisine bildirmek olacaktır. Bunun için MediatR kütüphanesinden faydalanıyoruz. MediatR kütüphanesi üzerinden aldığımız mediator objesine olayı bildiriyoruz ve gözlemleyicilerin işlemlerini bitirmesini bekliyoruz.
Stok Yönetimi içerisinde ürün yaratma emrini kullanıcıdan gelen istekle sadece ProductController sınıfı içerisinde oluşturuyoruz. Bu sebeple ProductController sınıfımız içerisinde MediatR kütüphanesinden gelen IMediator arayüzüne ait bir nesne örneği elde ediyoruz. Oluşturduğumuz iş emrini de bu obje aracılığı ile ilgili sınıflara iletiyoruz.
MediatR kütüphanesi üzerinde iş emirlerinin ve domain olaylarının dağıtımını yapıyoruz. Domain olaylarının aynı iş süreci (transaction) içerisinde ele alınması gerekmektedir. MediatR kütüphanesindeki IPipelineBehaviour arayüzünden yararlanarak ilk iş emrini yürütmeye başlamadan önce transaction başlatıp, süreç tamamlandığı zaman başlattığımız süreci sonuçlandırıyoruz. Bu işlemleri StockManagement/MediatRBehaviors dizininin altında yer alan TransactionalBehavior sınıfında gerçekleştirmekteyiz.
İş emirlerini yürüttükten sonra sistemde oluşan değişiklikleri MediatR aracılığı ile domainin içerisine bildiriyoruz. Bu işlem için Send(Gönder) değil Publish (Yayımla) eylemini kullanıyoruz. Bunun sebebi emirleri yürütecek tek bir gözlemleyici olabilmesidir. Mesajı o kişiye göndeririz fakat olan bir olayı birçok varlık izliyor olabilir. Bu sebeple olayları göndermiyoruz, yayımlıyoruz. Send eylemini izleyen IRequestHandler arayüzlerimiz varken Publish eylemiyle yayınladığımız olayları takip eden INotificationHandler arayüzünü implemente eden sınıflarımız olacaktır. Bunun için en iyi örneğimiz ise IStockSnapshotService arayüzümüz olacaktır. Stok verisi üzerinde oluşan değişiklikleri izleyip StockSnapshot modelini bu servis aracılığı ile güncelliyoruz.
- MassTransit
Olay güdümlü yapımızı kurgularken kullandığımız bir diğer kütüphane grubu da MassTransit ve MassTransit.RabbitMq kütüphaneleridir. MediatR bizim iş emirlerimizin ve domain olaylarımızın dağıtımından sorumludur. MassTransit kütüphanesi de entegrasyon olaylarımızın dağıtımından sorumlu olacaktır.
Öncelikle IIntegrationEventPublisher isminde bir arayüz tasarımı ile başlıyoruz. Bu arayüzün görevleri şunlardır: yayımlanmamış entegrasyon olaylarını listelemek, üzerinde atılması gereken entegrasyon olaylarını biriktirmek ve yayımlanmamış entegrasyon olaylarını yayımlamak.
Bu sınıfı bir önceki başlıktan hatırlayacağınız TransactionalBehaviour sınıfı içerisinde kullanıyoruz. Domain üzerinde yürüttüğümüz tüm işlemler tamamlandıktan sonra henüz yayınlanmamış entegrasyon olaylarını dış dünyaya duyuruyoruz.
MassTransit kütüphanesi, IIntegrationEventPublisher arayüzünü implemente eden sınıf içerisinde kullanılmaktadır. await _integrationEventPublisher.Publish(cancellationToken); satırı yürütüldüğü zaman MassTransit sahneye çıkarak entegrasyon olaylarımızı RabbitMq platformuna göndermektedir.
MassTransit aracılığı ile yürütebileceğimiz bir diğer operasyon da entegrasyon olaylarını takip etmektir. Bu durumun örneği StockManagement.Consumer projesi altında yer alan StockCreatorConsumer sınıfıdır. Bu sınıfımız MassTransit kütüphanesinden gelen IConsumer<T> arayüzünü implemente etmektedir. Bu sayede T sınıfının (bizim durumumuzda StockSnapshotCreatedIntegrationEvent) bir örneğine ait mesaj sistemde göründüğü zaman, kopyasını alıp işleyebileceğiz.
Not: MediatR kütüphanesinde de olduğu gibi olay gözlemleyicilerinin (event handler | consumer) yürüttükleri operasyonların dönüş tipi yoktur. Bunun sebebi gözlemleyicilerin tek yönlü akışlara sahip olmasıdır. Bir durum oluştuğu zaman kendi iş süreçlerini yürütmeye başlarlar. Onlardan cevap bekleyen herhangi bir nesne yoktur.
Okuma ve yazma modellerimizin farklı olduğu, okuma yaptığımız platform ile yazma yaptığımız platformun farklı olmasının gerektiği durumlarda MediatR ve MassTransit gibi kütüphaneler aracılığı ile CQRS tasarım kalıbını ve olay güdümlü programla tekniğini kullanabiliriz. Bu şekilde SOLID kurallarına uymak çok daha kolay olacaktır.
