MassTransit kullanarak RabbitMQ ile Messaging Altyapısı Oluşturma

Merhaba arkadaşlar.

Bir süredir sizlere messaging sistemleri üzerinde çalıştığımdan daha önceki makalelerimde bahsetmiştim. Messaging konusundaki diğer makalelerim üzerinden sizlerden gelen feedback’ler doğrultusunda MQ(Messaging Queue) yapısı ile beraber bir ESB(Enterprise Service Bus) kullanarak, büyük ölçekli uygulamaları nasıl daha iyi scale edebiliriz konusundaki bilgilerimi sizlere aktarmaya çalışacağım.

Bu makale kapsamındaki konu başlıklarımız ise sırasıyla aşağıdaki gibi olacaktır.

1. Messaging Architecture
2. Örnek Senaryo
3. ESB Nedir ve MassTransit’e Giriş
4. RabbitMQ ile MassTransit Implementasyonu

Messaging concept’leri hakkında diğer makalelerimde de sizlere olabildiğince bahsetmeye çalıştığım için dilerseniz bu bilgileri hatırlamak adına, Messaging Architecture’a genel bir bakış atalım.

1) Messaging Architecture

Messaging yapıları ile uygulamaları loosely coupled olarak, asenkron bir şekilde birbirleri ile iletişime geçirebildiğimizden daha önceki konularda bahsetmiştim. Daha çok enterprise düzeyde geliştirilen ve ihtiyaçlar doğrultusunda birbirlerinden bağımsız platformlarda yer alan uygulamalar, distributed bir şekilde günümüzde çalıştırılmaktadır. Messaging yapılarının tercih edilmesinin key point’i olarak, scalability ve flexibility diyebilirim en azından kendi adıma.

Scalability tarafından baktığımızda olaya büyük verilerle çalıştığımız ortamlarda, bu verileri process eden uygulamayı single instance ve aynı ortamda host etmek yerine farklı ortamdaki service’lere dağıtarak, distributed bir şekilde işlemleri hem daha hızlı hem de daha az yükle scale edebilmek mümkün oluyor. Monolith architecture’lar gibi her şeyi tek bir instance’a yükleyerek yapmak, büyük ölçekli enterprise uygulamalar için çok efektif olmayacaktır.

Messaging architecture basic concept olarak gayet basittir. Message’ları queue ya gönderen bir adet client vardır ve Producer olarak adlandırılır. Birde queue’daki message’ları dinleyip, process eden bir uygulama vardır, buda Consumer olarak adlandırılır. Messaging architecture implementasyon kısımlarında ise referans olarak alınabilecek Enterprise Integration Patterns olarak adlandırılan, çeşitli pattern ve messaging channel’lar mevcuttur. Biz ise gerçekleştirecek olduğumuz örneğimizdeki ihtiyaçdan dolayı, messaging channel olarak Publish-Subscribe pattern’ını implemente edeceğiz.

2) Örnek Senaryo

Messaging architecture ile ilgili concept’i genel olarak tazelememizin ardından, makale konusunda gerçekleştirecek olduğumuz örnek senaryomuza bir bakalım.

Konunun daha iyi anlaşılabilmesi için en çok transactional işlemlerin gerçekleştiği, e-commerce sistemlerindeki sipariş oluşturma örneğini gerçekleştireceğiz.

Bu tarz işlemlerde genelde bir sipariş oluşur, kullanıcıya bir e-mail gönderilir ve faturalandırma gibi vb. servisler çalışır. Biz ise gerçekleştirecek olduğumuz örnekte yukarıdaki flowchart’a baktığımızda, producer görevini üstlenecek olan bir Order UI olacak. Bu ekrandan düşen siparişler bir queue’ya aktarılacak. Bu queue’yu ise consume edecek olan bir OrderService olacak. OrderService ise gerekli order işlemlerini gerçekleştirecek. Bu aradaki iki uygulama, tamamen birbirlerinden bağımsız bir şekilde haberleşeceklerdir.

Makalenin 2. serisinde yer vermeyi düşündüğüm bir diğer senaryoya değinmek gerekirse, OrderService ilgili order işlemini gerçekleştirir ve sonra bir event fırlatır. Bu event ile ilgilenen herhangi bir XService, bu event’ı yakalar ve ilgili işlemlerini tamamlar. Örneğin order işlemi OrderService tarafından gerçekleştirildikten sonra, ilgili kullanıcıya bir bilgilendirme e-mail’ı gönderilebilmesi için, NotificationService’e bir event fırlatılır. NotificationService bu event’ı yakalar ve ilgili e-mail gönderim işlemini gerçekleştirir.

Bu tarz n sayıda farklı servis, birbirleri ile loosely coupled bir şekilde haberleşebilmektedirler.

3) ESB Nedir ve MassTransit’e Giriş

Enterprise Service Bus (ESB) için genel olarak transport işlemleri için bir gateway’dir diyebiliriz. ESB’nin görevi aslında high level olarak özünde, transport işlemlerini higher abstraction seviyesinde client api için halletmektedir.

Olaya biraz daha anlaşılır bir bakış açısından da bakmak gerekirse, SOA mimarilerinin kolay ve reliable bir şekilde uygulanabilmesi için içerisinde bir çok altyapısal fonksiyonları barındıran bir altyapı mimarisidir diyebiliriz. Bunlara ek olarak distributed olan uygulamaları çalıştırmayı ve yönetebilmeyi de kolaylaştırmaktadır.

Şuanda .Net stack’i altında popüler olan NServiceBus ve MassTransit gibi ESB’ler mevcuttur. MassTransit open-source bir projedir ve bir çok MQ sistemine desteği de bulunmaktadır. Makale içeriğinde gerçekleştirecek olduğumuz örnekte MassTransit ile ilerleyeceğimiz için, dilerseniz sağlıyor olduğu bazı benefit’lere bir bakalım.

MassTransit Benetifs:

• Transport işlemlerinin complexity’sini gizler
• Multiple transport desteği sunar
• Build-in olarak içerisinde retries policy’leri mevcuttur (ki MQ yapılarında olmazsa olmaz bir yapıdır)
• Failure management sağlar
• MassTransit Test Framework paketi ile kolay unit testing sağlar
• Message  scheduling mevcuttur. (periodic ve publishing yapılabilmektedir)
• Request/Response pattern’larını destekler
• Verimlilik için kendisi exchange’leri yönetir

gibi faydaları bulunmaktadır. Bunlara ek olarak da message’ları intercept edebilmeye de olanak sağlamaktadır.

3) RabbitMQ ile MassTransit Implementasyonu

Gerekli concept’lere göz attıktan sonra, artık yavaş yavaş implementasyon işlemlerine geçebiliriz. “LightMessagingCore.Boilerplate” isminde boş bir solution oluşturarak, ilk adımımızı atalım. Solution içerisine “LightMessagingCore.Boilerplate.Common” isminde bir class library ekleyelim ve burada bus için configuration işlemlerimizi gerçekleştirmeye başlayalım.

“MqConstants” isminde bir class ekleyelim ve aşağıdaki gibi kodlayalım.

using System.Configuration;

namespace LightMessagingCore.Boilerplate.Common
{
    public class MqConstants
    {
        public static string RabbitMQUri => ConfigurationManager.AppSettings["RabbitMQUri"];
        public static string RabbitMQUserName => ConfigurationManager.AppSettings["RabbitMQUserName"];
        public static string RabbitMQPassword => ConfigurationManager.AppSettings["RabbitMQPassword"];
    }
}

Tanımlamış olduğumuz bu class’da, daha sonra ihtiyaç duyacağımız bazı property’leri ekledik. Burada kullanacak olduğumuz RabbitMQ broker’ı için URI adresi, varsa credential bilgileri ve bir queue ismi yer almaktadır.

MassTransit implementasyonuna başlamadan önce hemen Nuget Package Manager üzerinden aşağıdaki gibi “MassTransit.RabbitMQ” paketini kuralım.

Paket kurulumunu tamamladıktan sonra artık ESB configuration’larını gerçekleştirecek olduğumuz “BusConfigurator” class’ını aşağıdaki gibi kodlamaya başlayalım.

using System;
using MassTransit;
using MassTransit.RabbitMqTransport;

namespace LightMessagingCore.Boilerplate.Common
{
    public class BusConfigurator
    {
        private static readonly Lazy<BusConfigurator> _Instance = new Lazy<BusConfigurator>(() => new BusConfigurator());

        private BusConfigurator()
        {

        }

        public static BusConfigurator Instance => _Instance.Value;

        public IBusControl ConfigureBus(
    Action<IRabbitMqBusFactoryConfigurator, IRabbitMqHost> registrationAction = null)
        {
            return Bus.Factory.CreateUsingRabbitMq(cfg =>
            {
                var host = cfg.Host(new Uri(MqConstants.RabbitMQUri), hst =>
                {
                    hst.Username(MqConstants.RabbitMQUserName);
                    hst.Password(MqConstants.RabbitMQPassword);
                });

                registrationAction?.Invoke(cfg, host);
            });
        }
    }
}

Singleton olarak tasarladığımız “BusConfigurator” class’ı içerisinde, “ConfigureBus” method’u ile MassTransit’in “IBusControl” interface’ini implemente eden bir RabbitMQ BusControl instance’ı oluşturuyoruz. Bu işlem için ise “Bus.Factory.CreateUsingRabbitMq” method’unu call ederek, “IRabbitMqBusFactoryConfigurator” tipinde bir action oluşturuyoruz ve içerisinde RabbitMQ’ya connect olabilmek için gerekli olan host credential bilgilerini tanımlıyoruz.

Bir diğer yandan “registrationAction” parametresi ile ise “ConfigureBus” method’unu çağırırken ihtiyaç duyulduğu noktalarda, RabbitMQ bus’ı için gerekli olan “queue name”, “endpoint” ve “consumer” gibi bilgileri veriyor olacağız.

En basit haliyle “BusConfigurator” class’ı şimdilik bu kadar.

Not: Makale içerisinde gerçekleştirecek olduğumuz örneği, sonrasında lightweight bir messaging core boilerplate’i olarak github üzerinden ilerletmeyi düşünüyorum. En kısa zamanda bus configurator için retries policy’leri, circuit breaker, rate limiter vb. gibi özellikleri de fluent bir şekilde implemente etmeyi düşünüyorum.

Şimdi producer ve consumer’ın exchange işlemi sırasında kullanacak olduğu, messaging interface’lerini tanımlamaya başlayabiliriz. Bunun için öncelikle solution üzerine “LightMessagingCore.Boilerplate.Messaging” isminde bir class library daha ekleyelim.

Library eklemeyi tamamladıktan sonra içerisine aşağıdaki gibi “IOrderCommand” isminde bir interface tanımlayalım.

namespace LightMessagingCore.Boilerplate.Messaging
{
    public interface IOrderCommand
    {
        int OrderId { get; set; }
        string OrderCode { get; set; }
    }
}

Oluşturmuş olduğumuz bu interface sadece “OrderId” ve “OrderCode” bilgilerini tutmaktadır.

Not: Bu kısımda önemli bir nokta bulunmaktadır. ESB olarak MassTransit kullandığımız için, producer ve consumer’ın aynı messaging interface‘ini exchange ediyor olmaları önemlidir. Aksi takdirde producer’ın queue’ya gönderdiği message’ları, consumer consume edemeyecektir. Bunun sebebi ise MassTransit’in verimliliği arttırabilmek için exchange’leri kendisinin yönetmesidir. Yönetim işlemini ise MassTransit, interface’lere göre gerçekleştirmektedir. Bu konuya makalenin ilerleyen bölümlerinde görsel olarak değineceğim.

3.1) Producer’ın Hazırlanması

Producer kısmını yani Order UI’ı oluşturabilmek için gerekli altyapımız hazır durumda. Solution üzerine “LightMessagingCore.Boilerplate.OrderUI” isminde, template olarak ise empty seçili bir  Asp.Net MVC projesi ekleyelim. Projenin ekleme işleminden sonra ise solution structure’ın son hali aşağıdaki gibi olacaktır.

Projeyi ekledikten sonra ise “Controllers” kısmına “OrderController” isminde, empty bir MVC controller’ı ekleyelim.

Not: “LightMessagingCore.Boilerplate” solution’ı altında, sanki monolith architecture’a doğru gidiyor gibi olabiliriz. Fakat gerçek bir ortamda oluşturmuş olduğumuz “LightMessagingCore.Boilerplate.Common” ve ortak paylaşmamız gereken contract’ın da olduğu “LightMessagingCore.Boilerplate.Messaging” library’lerini, nuget server’lar üzerinden birer package olarak yönetebilmek mümkündür. Sonrasında ise producer ve consumer’lar monolithlikten çıkarak, birer microservice olarak geliştirilebilir. Ben örnek gereği tek bir solution üzerinden ilerleyeceğim.

Artık projeye ilgili referansları eklemeye başlayabiliriz. “LightMessagingCore.Boilerplate.OrderUI” projesinin referans kısmına gelerek “LightMessagingCore.Boilerplate.Common”, “LightMessagingCore.Boilerplate.Messaging” library’lerini referans olarak ekleyelim ve Nuget Package Manager üzerinden de “MassTransit.RabbitMQ” paketini kuralım.

Referans ekleme işlemlerinin ardından order model’ini oluşturabiliriz. Order model’ine “LightMessagingCore.Boilerplate.Messaging” library’sinde oluşturmuş olduğumuz “IOrderCommand” interface’ini implemente edeceğiz. “Models” klasörüne “OrderModel” isminde yeni bir class tanımlayalım ve “IOrderCommand” interface’ini aşağıdaki gibi implemente edelim.

using LightMessagingCore.Boilerplate.Messaging;

namespace LightMessagingCore.Boilerplate.OrderUI.Models
{
    public class OrderModel : IOrderCommand
    {
        public string OrderCode { get; set; }

        public int OrderId { get; set; }
    }
}

Implementasyon işleminden sonra ise “OrderController” a gelerek, “Index” action’ı üzerine sağ tıklayıp yeni bir view ekleyelim. View ekleme işlemi sırasında bize kolaylık sağlayabilmesi adına template olarak “Create” ve model class olarak ise oluşturmuş olduğumuz “OrderModel” class’ını seçelim.

View’ın eklenmesinin ardından “Views>Order>Index.cshtml” path’ini takip edelim ve aşağıdaki gibi güncelleyelim.

@model LightMessagingCore.Boilerplate.OrderUI.Models.OrderModel

@{
    ViewBag.Title = "Order Create";
}

<h2>Order Create</h2>


@using (Html.BeginForm())
{
    @Html.AntiForgeryToken()

    <div class="form-horizontal">
        <h4>Order</h4>
        <hr />
        @Html.ValidationSummary(true, "", new { @class = "text-danger" })
        <div class="form-group">
            @Html.LabelFor(model => model.OrderCode, htmlAttributes: new { @class = "control-label col-md-2" })
            <div class="col-md-10">
                @Html.EditorFor(model => model.OrderCode, new { htmlAttributes = new { @class = "form-control" } })
                @Html.ValidationMessageFor(model => model.OrderCode, "", new { @class = "text-danger" })
            </div>
        </div>
        <div class="form-group">
            @Html.LabelFor(model => model.OrderId, htmlAttributes: new { @class = "control-label col-md-2" })
            <div class="col-md-10">
                @Html.EditorFor(model => model.OrderId, new { htmlAttributes = new { @class = "form-control" } })
                @Html.ValidationMessageFor(model => model.OrderId, "", new { @class = "text-danger" })
            </div>
        </div>

        <div class="form-group">
            <div class="col-md-offset-2 col-md-10">
                <input type="submit" value="Create" class="btn btn-default" />
            </div>
        </div>
    </div>
}

<div>
    @Html.ActionLink("Back to List", "Index")
</div>

<script src="~/Scripts/jquery-1.10.2.min.js"></script>
<script src="~/Scripts/jquery.validate.min.js"></script>
<script src="~/Scripts/jquery.validate.unobtrusive.min.js"></script>

İlgili paketlerin ve view’ın tamamlanmasının ardından artık “OrderController” ı kodlamaya devam edebiliriz.  “OrderController” ı aşağıdaki gibi kodlayalım.

using System;
using System.Configuration;
using System.Web.Mvc;
using LightMessagingCore.Boilerplate.Common;
using LightMessagingCore.Boilerplate.OrderUI.Models;
using MassTransit;

namespace LightMessagingCore.Boilerplate.OrderUI.Controllers
{
    public class OrderController : Controller
    {
        private readonly ISendEndpoint _bus;

        public OrderController()
        {
            var busControl = BusConfigurator.Instance.ConfigureBus();
            var sendToUri = new Uri($"{MqConstants.RabbitMQUri}{ConfigurationManager.AppSettings["OrderQueueName"]}");

            _bus = busControl.GetSendEndpoint(sendToUri).Result;
        }

        // GET: Order
        public ActionResult Index(OrderModel orderModel)
        {
            if (orderModel.OrderId > 0)
                CreateOrder(orderModel);

            return View();
        }

        private void CreateOrder(OrderModel orderModel)
        {
            _bus.Send(orderModel).Wait();
        }
    }
}

Burada öncelikle constructor üzerinden “busControl” ü initialize ediyoruz. Bu işlem için common library’si üzerinde oluşturmuş olduğumuz “BusConfigurator” class’ını kullanıyoruz. Devamında ise bir queue endpoint’i belirleyerek, “busControl” ün “GetSendEndpoint” method’una gönderiyoruz. Bu işlemin ardından artık tüm messaging işlemleri, initialize ettiğimiz “ISendEndpoint” i üzerinden devam edecektir. “CreateOrder” method’una baktığımızda ise artık çok straightforward bir hale geldiğini görebiliriz. Artık tek yapmamız gereken bus üzerindeki “Send” method’unu çağırarak, içerisine ilgili “IOrderCommand” interface’ini implemente etmiş model’i vermek yeterli olacaktır.

Artık producer hazır durumdadır. Dilerseniz consumer kısmına başlamadan önce “MqConstants” ve “OrderController” içerisinde tanımlamış olduğumuz key’leri, config dosyasına aşağıdaki gibi ekleyelim.

<add key="RabbitMQUri" value="rabbitmq://localhost/"/>
<add key="RabbitMQUserName" value="guest"/>
<add key="RabbitMQPassword" value="guest"/>
<add key="OrderQueueName" value="lightmessagingcore.boilerplate.order"/>

Not: Bu makale kapsamında RabbitMQ kurulumuna değinmeyeceğim. Buradan daha önce yazmış olduğum makaleye ulaşarak,windows üzerine kurulum bilgilerine erişebilirsiniz veya docker üzerine kurup, kullanabilirsiniz.

3.2) Consumer’ın Hazırlanması

Consumer için solution üzerine “LightMessagingCore.Boilerplate.OrderService” isminde bir console application ekleyelim. Bu kısım microservice yaklaşımında olup, oldukça lightweight tutmaya çalışacağız. Diğer projelerde olduğu gibi buraya da “LightMessagingCore.Boilerplate.Common”, “LightMessagingCore.Boilerplate.Messaging” library’lerini referans olarak ekeyerek, Nuget Package Manager üzerinden “MassTransit.RabbitMQ” paketini de dahil edelim.

Şimdi “Program.cs” e gelelim ve burada bus control’ü initialize edelim.

using System;
using LightMessagingCore.Boilerplate.Common;
using System.Configuration;
using MassTransit;

namespace LightMessagingCore.Boilerplate.OrderService
{
    class Program
    {
        static void Main(string[] args)
        {
            Console.Title = "OrderService";

            var bus = BusConfigurator.Instance
                .ConfigureBus((cfg, host) =>
                {
                    cfg.ReceiveEndpoint(host, ConfigurationManager.AppSettings["OrderQueueName"], e =>
                    {
                        e.Consumer<OrderCommandConsumer>();
                    });
                });

            bus.Start();

            Console.WriteLine("Listening order command..");
            Console.ReadLine();
        }
    }
}

Burada da common projesi içerisinde daha önce oluşturmuş olduğumuz “BusConfigurator” class’ını kullanarak, yeni bir bus initialize ediyoruz. Bu sefer farklı olarak initializing kısmında bir action method kullanıyoruz ve oluşturuyor olduğumuz configuration’ın message’ları consume edeceği endpoint’i, “ReceiveEndpoint” method’u ile set ediyoruz ve ardından endpoint’in hangi consumer aracılığı ile consuming işlemini gerçekleştireceğini belirliyoruz. Konuyu toplamak gerekirse bu kısım artık bu uygulamanın kendisine set edilen endpoint üzerinden, “OrderCommandConsumer” type’ı ile consume işlemini gerçekleştireceğini gösteriyor.

Dilerseniz şimdi “OrderCommandConsumer” ı aşağıdaki gibi kodlayalım.

using System;
using System.Threading.Tasks;
using LightMessagingCore.Boilerplate.Messaging;
using MassTransit;

namespace LightMessagingCore.Boilerplate.OrderService
{
    public class OrderCommandConsumer : IConsumer<IOrderCommand>
    {
        public async Task Consume(ConsumeContext<IOrderCommand> context)
        {
            var orderCommand = context.Message;

            await Console.Out.WriteAsync($"Order code: {orderCommand.OrderCode} Order id: {orderCommand.OrderId}");

            //do something..
        }
    }
}

Gördüğümüz gibi bu kısımda oldukça basit bir durumda. Tek yapmamız gereken şey MassTransit içerisinde bulunan “IConsumer<T>” interface’ini, ilgili contract’ımız ile implemente etmektir. “Consume” method’u ile de consume işlemlerini gerçekleştirmektedir.

3.3) Test

Evet consumer’da şuan hazır durumdadır. Sizlerde fark ettiyseniz eğer, MassTransit bir çok konuda bizlere kolaylık ve hız sağlamaktadır. Dilerseniz artık test işlemlerine geçebiliriz. Bunun için solution ayarlarından multiple startup projects olarak, “OrderService” ve “OrderUI” projelerini seçelim ve start tuşuna basalım.

Bu işlemin ardından öncelikle consumer görevini görecek olan “LightMessagingCore.Boilerplate.OrderService” projesi, aşağıdaki gibi açılacaktır.

Order service artık “IOrderCommand” interface’ine sahip message’ları, consume durumundadır. Consumer’ın ve Producer’ın initialize olmasıyla beraber dilerseniz RabbitMQ Management ekranı üzerinden, exchange’lere bir bakalım.

Biz her iki uygulamada da sadece queue name olarak “lightmessagingcore.boilerplate.order” set etmiştik.

Makalenin üst kısımlarında hatırlarsak MassTransit’in verimlilik için, exchange’leri kendisinin yönettiğinden bahsetmiştik. İşte bu noktada yukarıdaki görselde olduğu gibi “LightMessagingCore.Boilerplate.Messaging:IOrderCommand” isimli exchange’i oluşturmuştur. Burada “IOrderCommand” contract’ı ile gelen message’ları otomatik olarak kendisi aşağıda bulunan queue’ya bind etmektedir.

Şimdi “OrderUI” ın olduğu tab’a gelelim ve aşağıdaki gibi bir order create edelim.

Create işleminin ardından “OrderService” ilgili message’ı yakalayıp, consume işlemini aşağıdaki gibi gerçekleştirmiştir.

Bu makale kapsamında sizlerle service bus olarak MassTransit kullanıp, RabbitMQ ile messaging işlemlerini nasıl gerçekleştirebileceğimizi ele aldık ve messaging architecture hakkındaki temel bilgimizi tazeledik. Bir sonraki messaging serisinde ise reliability, fault management ve HA gibi konulara değinmeyi düşünüyorum.

İlgili projeye github hesabım üzerinden aşağıdan erişebilir, sizlerde bu core library’nin geliştirilmesinde katkıda bulunabilirsiniz. Takipte kalın.

http://github.com/GokGokalp/lightmessagingcore-boilerplate