Amazon SQS & Asp.NET Core – Scalable Messaging

Merhaba,

Biliyorsunuz ki, bir yazılım için önem arz eden birçok faktör arasından ölçeklenebilirlik ve güvenilir mesajlaşma yapılanması oldukça kritik arz etmektedir. Bu ihtiyaçlara karşın kullanabileceğimiz muhtelif teknolojiler olmasının yanında Amazon SQS’de imdadımıza yetişmekte ve oldukça esnek ve güçlü bir şekilde web ve mobil uygulamalar için alternatif olarak değerlendirilebilmektedir. Bu içeriğimizde Amazon SQS hakkında temel kavramları ve mimarisel yapıyı inceleyecek, .NET uygulamalarıyla nasıl entegre edileceği üzerine bir çalışma gerçekleştiriyor olacağız. O halde buyurun Amazon SQS ile .NET’in gücünü birlikte keşfetmeye başlayalım.

Amazon SQS(Amazon Simple Queue Service) Nedir?

Amazon SQS, AWS’nin genellikle microservice, distributed yahut event-driven mimarilerini kullanan uygulamalarda mesajların güvenli bir şekilde gönderilmesi, alınması ve saklanması için geliştirmiş olduğu yönetilebilir, yüksek düzeyde kullanılabilir, ölçeklenebilir, serverless bir servisidir. Oldukça uygun maliyetlerde kullanılabilir olan Amazon SQS, standart kuyruk yapılanmaları yanında FIFO davranış sergileyen kuyruk mimarisi sunabilmekte; AWS Lambda, EC2 ve diğer AWS servisleri ile kolayca entegre edilebilmektedir.

Amazon SQS’in Temel Kavramları Nelerdir?

Bu görsel https://codewithmukesh.com/blog/amazon-sqs-and-aspnet-core/ adresinden alınmıştır.

• Queue
  Queue, bir mesajın gönderildiği ve alındığı depolama alanıdır. Queue’lar sayesinde mesajlar bir sistemden diğerine güvenli bir şekilde iletilebilmektedir.

  Standard Queues ve FIFO (First-In-First-Out) Queues olmak üzere iki temel queue türü mevcuttur;

  1. Standard Queues
     İlk piyasaya sürülen queue türü olan standard queue’lar; mesajların en az bir kez teslim edildiği, büyük hacimli mesajların yüksek hızda işlenebildiği ve mesajların işlenme sırasının bir öneminin olmadığı sistemlerde kullanılan queue’lardır. Son ibareye odaklanırsak eğer birden fazla mesajın söz konusu olduğu durumlarda queue’da ki mesajların işlenmesi sürecinde sırasının korunması garanti edilmemektedir. Buna örnek vermemiz gerekirse eğer; 1, 2, 3, 4, 5 şeklinde gelen mesajlar consumer’lar tarafından 1, 5, 3, 2, 4 şeklinde işlenebilmesi olasıdır.
  2. FIFO (First-In-First-Out) Queues
     Adından da anlaşılacağı üzere bu queue’lar da mesajlar sırayla işlenmektedir. Ve standard queue’ların aksine her mesaj yalnızca bir kez işlenir. Bir düzen sağladığı için ister istemez standard queue’lara nazaran daha maliyetli olabilmektedir. Bu yüzden performansta biraz azalma gözlemlenebilir.
• Message Processing
  Message Processing ise bir kuyruğa gönderilen mesajların alınması ve işlenmesi sürecidir. Bu süreçte bilinmesi gereken kritik bir davranış vardır. Bu davranış, iki farklı consumer tarafından bir queue’dan aynı mesajın alınamamasıdır. Bir consumer, bir mesajı aldığında bu mesaj diğer consumer’lar tarafından görülemeyecek şekilde kilitlenmektedir. Böylece bir mesaja yalnızca bir thread’in erişebilmesinin garantisi sağlanmış olacaktır. Tabi mesaj alındıktan sonra belirli bir süre zarfında işlenmezse mesajın farklı bir consumer tarafından tüketilebilmesi için yeniden görünür hale getirilmesi gerekecek ve bunun için de mesajın kilidi açılarak tekrar queue’ya geri koyulacaktır. Buradaki süre terminolojik olarak Visibility Timeout olarak nitelendirilmektedir. Bu süre default olarak 30 saniyedir ve her bir mesaj için özelleştirilememektedir!

  Ayrıca bir mesaj düzgün bir şekilde işlendiği taktirde kuyruktan kalıcı olarak silinecektir. Amma velakin belirli bir sayıda başarısız denemeden sonra(ki bu geliştirici tarafından Maximum Receives parametresi ile belirlenmektedir) işlenemeyen mesajlar ayrı bir kuyruk olan Dead Letter Queue (DLQ)‘ya taşınacaktır.

Amazon SQS’in Avantajları Nelerdir?

Amazon SQS’in, RabbitMQ ve Kafka gibi diğer mesajlaşma sistemlerine kıyasla çeşitli avantajları mevcuttur. Bu avantajlar, belirli senaryolarda Amazon SQS’i daha cazip hale getirebilmektedirler. İşte Amazon SQS’in birkaç öne çıkan avantajları;

• Tam Yönetilen Hizmet
  Amazon SQS, AWS tarafından tamamen yönetildiği için sunucu yönetimi, yazılım güncellemeleri, ölçeklendirme ve bakım gibi sorumluluklardan bizleri münezzeh kılmaktadır. Haliyle uygulama sahipleri tarafından altyapıyı yönetmek ve sorumluluğunu üstlenmek ciddi maliyet ve kalem olacağından dolayı bu açıdan büyük avantaj sağlamaktadır.
• Kolay Entegrasyon ve Kullanım
  Amazon SQS, AWS ekosisteminin bir parçası olduğundan diğer AWS servisleriyle(Lambda, EC2, S3 vs.) kolayca entegre edilebilmektedir. Ayrıca SQS’in kullanımı, basit bir API ile sağlanabilmekte ve bu da hızlı bir kullanım ve öğrenim eğrisi sağlamaktadır.
• Otomatik Ölçeklendirme
  Amazon SQS, message queue trafiği arttığında otomatik olarak ölçeklendirmeyi gerçekleştirmektedir. Haliyle geliştiriciler tarafından kapasitenin önceden planlanması gerekmemekte ve böylece sistemde anlık olası trafik artışlarına karşı uyum doğal bir şekilde sağlanabilmektedir.
• Güvenilirlik ve Dayanıklılık
  Amazon SQS, AWS’in yüksek güvenilirlik ve dayanıklılık sağlayan altyapısı üzerinde çalıştığı için mesajlar birden fazla veri merkezinde çoğaltılarak güvence altına alınmaktadır. Dolayısıyla bu açıdan olayı değerlendirdiğimizde de herhangi bir risk durumu söz konusu değildir.
• Fiyatlandırma ve Maliyet Verimliliği
  Amazon SQS’de kullanıcılar yalnızca kullandıkları kadar ödeme yaparlar. Özellikle küçük ölçekli ve değişken trafik senaryoları için maliyet oldukça verimlidir diyebiliriz.
• FIFO ve Dead Letter Queue(DLQ) Desteği
  FIFO queue’ları sayesinde tam sıralı mesaj teslimi sağlarken, DLQ ile de başarısız mesajları kolayca yönetilebilir hale getirmektedir.
• Güvenlik
  Amazon SQS, AWS IAM(Identity and Access Management) ile entegrasyon, güvenlik ve erişim denetimi için gelişmiş yetenekler sunmaktadır. Ayrıca mesajlar varsayılan olarak şifrelenmektedir.

Amazon SQS’in Kullanımı

Şimdi Amazon SQS’i nasıl kullanabileceğimizi inceliyor olacağız. Bu incelemeyi hem AWS üzerinden gerçekleştireceğiz hem de Asp.NET Core uygulamaları üzerinden. İlk olarak AWS’de bir kuyruğun nasıl oluşturulup kullanılabileceğini değerlendireceğiz. Bu aşamadan sonra artık olayı programatik ele alacak ve biri producer diğeri consumer olmak üzere iki Asp.NET Core uygulaması oluşturacağız. Haliyle planlama belli. O halde hadi başlayalım.

• AWS Management Console Aracılığıyla Amazon SQS Oluşturma

  İlk olarak Amazon Management Console üzerinden SQS servisine gidelim.Create queue butonuna tıklayıp ardından Standard türünü seçerek bir kuyruk adı belirleyelim.Tabi bizler içeriğimiz boyunca standard türü üzerinden örneklendirmede bulunmayı tercih ediyoruz ama siz isterseniz burada FIFO türünden de kuyruk oluşturabilirsiniz.

  Ardından devamında ilgili kuyruğu aşağıdaki gibi yapılandıralım.
  Burada yaptığımız yapılandırmada önceki satırlarda bahsettiğimiz Visibility Timeout değerini yapılandırabilmekteyiz. Neydi la bu? diye hatırlamak maksatlı bu suali cevaplandırırsak eğer bir mesajın bir consumer tarafından okunduktan sonra diğer consumer’ların onu okuyabilmesi için kilitlendiği süreyi temsil ediyordu. Haliyle önceden de ifade ettiğimiz gibi görselden de varsayılan olarak 30 saniye olduğunu görmekteyiz.

  Bunun dışında bir mesajın kuyrukta görünmesine rağmen consumer’lar tarafından gecikmeli bir şekilde işlenebilmesi için Delivery Delay değerini belirleyebilir, henüz silinmemiş bir mesajın ne kadar periyot sonrasında silinebileceğini Retention Period ile ayarlayabilir ve kuyruktaki mesajların kullanılabilir olması için Receive Message Wait Time değeriyle oynayabiliriz.

  Devamında ise oluşturacağımız kuyruğa dair diğer yapılandırmaları varsayılan ayarlarıyla kabul ederek kuyruğu oluşturalım.Son olarak oluşturduğumuz bu kuyruğa aşağıdaki ekran görüntüsünde olduğu gibi AWS Management Console üzerinden mesaj yayınlayabilir ve subscribe olarak yayınlanan mesajları elde edebiliriz.AWS üzerinden kuyruk yapılanmasını kullanmak işte bu kadar 🙂

• Asp.NET Core – Producer ve Consumer Altyapısını Geliştirme

  Şimdi sıra Asp.NET Core kısmına gelmiştir. Burada ilk olarak producer ve consumer uygulamalarında kullanacağımız altyapıyı geliştirmeye odaklanalım. Bunun için solution içerisinde Shared isimli bir class library oluşturalım ve aşağıdaki paketlerin ilgili projeye yüklenmesiyle başlayalım.

  Install-Package AWSSDK.SQS
Install-Package AWSSDK.Extensions.NETCore.Setup

  Ardından bu class library içerisinde aşağıdaki producer ile consumer arasındaki haberleşmeyi örnek/misal olarak modelleyecek olan OrderCreatedEvent‘i tasarlayalım.

      public class OrderCreatedEvent : IEvent
      {
          public Guid OrderId { get; set; }
          public Guid CustomerId { get; set; }
          public DateTime CreatedDate { get; } = DateTime.UtcNow;
      }
  

  Ve son olarak bu Shared isimli class library’i producer ve consumer uygulamalarında referans edelim.

  Artık producer ya da consumer uygulamalarını geliştirmeye odaklanabiliriz. Burada opsiyonel olarak bizler ilk producer uygulamasını geliştirmeyle başlayalım.

  using Amazon.SQS;
  using Amazon.SQS.Model;
  using Shared.Events;
  using System.Text.Json;
  
  var builder = WebApplication.CreateBuilder(args);
  
  builder.Services.AddDefaultAWSOptions(builder.Configuration.GetAWSOptions());
  builder.Services.AddAWSService<IAmazonSQS>();
  
  var app = builder.Build();
  
  app.MapGet("/create-order", async (IAmazonSQS amazonSQS) =>
  {
      string queueName = "order-created-queue";
  
      OrderCreatedEvent orderCreatedEvent = new()
      {
          CustomerId = Guid.NewGuid(),
          OrderId = Guid.NewGuid()
      };
  
      string queueUrl = string.Empty;
  
      try
      {
          GetQueueUrlResponse getQueueUrlResponse = await amazonSQS.GetQueueUrlAsync(queueName);
          queueUrl = getQueueUrlResponse.QueueUrl;
      }
      catch (QueueDoesNotExistException)
      {
          CreateQueueResponse createQueueResponse = await amazonSQS.CreateQueueAsync(queueName);
          queueUrl = createQueueResponse.QueueUrl;
      }
  
      SendMessageRequest sendMessageRequest = new()
      {
          QueueUrl = queueUrl,
          MessageBody = JsonSerializer.Serialize(orderCreatedEvent)
      };
  
      await amazonSQS.SendMessageAsync(sendMessageRequest);
      return Results.Ok();
  });
  
  app.Run();
  

  Producer’da yapılacak işlemi yukarıdaki gibi tasarlayabiliriz. Yapılan çalışmaya göz atarsak eğer 8 ve 9. satırlarda AWS Service entegrasyonu gerçekleştirilmekte ve 25 ile 34. satır aralığında ise ‘order-created-queue’ adında bir queue tanımlanmakta yahut önceden tanımlanmışsa var olan kullanılarak mesaj gönderimi sağlanmaktadır.

  Producer’ı tamamladığımıza göre sıra consumer’a gelmiştir. Producer tarafından yayınlanmış mesajları consumer’da yakalayabilmek için aşağıdaki gibi bir BackgroundService oluşturabiliriz.

  
  using Amazon.SQS;
  using Amazon.SQS.Model;
  
  namespace Consumer.BackgroundServices
  {
      public class OrderCreatedEventConsumer(IAmazonSQS amazonSQS) : BackgroundService
      {
          protected override async Task ExecuteAsync(CancellationToken stoppingToken)
          {
              string queueName = "order-created-queue";
              string queueUrl = string.Empty;
  
              try
              {
                  GetQueueUrlResponse getQueueUrlResponse = await amazonSQS.GetQueueUrlAsync(queueName);
                  queueUrl = getQueueUrlResponse.QueueUrl;
              }
              catch (QueueDoesNotExistException)
              {
                  CreateQueueResponse createQueueResponse = await amazonSQS.CreateQueueAsync(queueName);
                  queueUrl = createQueueResponse.QueueUrl;
              }
  
              ReceiveMessageRequest receiveMessageRequest = new()
              {
                  QueueUrl = queueUrl
              };
  
              while (!stoppingToken.IsCancellationRequested)
              {
                  ReceiveMessageResponse receiveMessageResponse = await amazonSQS.ReceiveMessageAsync(receiveMessageRequest);
                  if (receiveMessageResponse.Messages.Count > 0)
                      foreach (Message message in receiveMessageResponse.Messages)
                      {
                          //OrderCreatedEvent _message = JsonSerializer.Deserialize<OrderCreatedEvent>(message.Body);
                          Console.WriteLine(message.Body);
                          await Task.Delay(1000);
                          await amazonSQS.DeleteMessageAsync(queueUrl, message.ReceiptHandle);
                      }
              }
          }
      }
  }
  

  Devamında ise ‘Program.cs’ dosyasında aşağıdaki gibi yapılandırmada bulunalım.

  using Amazon.SQS;
  using Consumer.BackgroundServices;
  
  var builder = WebApplication.CreateBuilder(args);
  
  builder.Services.AddDefaultAWSOptions(builder.Configuration.GetAWSOptions());
  builder.Services.AddAWSService<IAmazonSQS>();
  builder.Services.AddHostedService<OrderCreatedEventConsumer>();
  
  var app = builder.Build();
  
  app.MapGet("/", () => "Hello World!");
  
  app.Run();
  

  Evet, böylece artık consumer’da hazır kıta modundadır diyebiliriz.

  Artık tek yapılması gereken her iki uygulamanın ‘appsettings.json’ dosyasında aşağıdaki yapılandırmanın sağlanmasıdır.

    "AWS": {
      "Profile": "sqs-profile",
      "Region": "ap-south-1"
    }
  

  Nihai olarak yaptığımız bu çalışmayı derleyip çalıştırdığımızda producer ile consumer arasındaki haberleşmenin başarıyla sağlandığını gözlemlemiş oluyoruz.

MassTransit Entegrasyonu

Tabi eğer ki sahada bir message broker kullanılıyorsa bunu burada yaptığımız gibi manuel bir şekilde yapılandırıp kullanmaktansa türlü avantajlar ve sebeplerden dolayı bir Enterprise Service Bus (ESB) kullanmayı tercih ederiz. Ee haliyle ESB deyince de akla ilk gelen MassTransit kütüphanesi olacağından dolayı Amazon SQS’i MassTransit’le nasıl kullanabileceğimizi de inceleyip öyle içeriğimizi noktalayacağız. Aksi taktirde Amazon SQS ile MassTransit’i tecrübe etmeksizin bu içeriği noktalamak ciddi eksiklik olacaktır.

Bunun için öncelikle aşağıdaki paketi Shared projesine yükleyelim.
Install-Package MassTransit.AmazonSQS

Ardından producer’ın ‘Program.cs’ dosyasında aşağıdaki yapılandırmada bulunalım.

#region MassTransit
builder.Services.AddMassTransit(configurator =>
{
    configurator.UsingAmazonSqs((context, _configurator) =>
    {
        _configurator.Host(Amazon.RegionEndpoint.APSouth1.OriginalSystemName, hostConfigurator =>
        {
            hostConfigurator.AccessKey(builder.Configuration["AWS:AccessKey"]);
            hostConfigurator.SecretKey(builder.Configuration["AWS:SecretKey"]);
        });
    });
});
#endregion

Ve mesaj gönderebilmek için de aşağıdaki gibi bir action oluşturalım.

#region MassTransit
app.MapGet("/create-order-masstransit", async (IPublishEndpoint publishEndpoint, ISendEndpointProvider sendEndpointProvider) =>
{
    OrderCreatedEvent orderCreatedEvent = new()
    {
        CustomerId = Guid.NewGuid(),
        OrderId = Guid.NewGuid()
    };

    #region Publish
    //await publishEndpoint.Publish(orderCreatedEvent);
    #endregion

    #region Send
    var sendEndpoint = await sendEndpointProvider.GetSendEndpoint(new Uri($"queue:masstransit-order-created-event-queue"));
    await sendEndpoint.Send(orderCreatedEvent);
    #endregion

    return Results.Ok();
});
#endregion

Benzer mantıkla consumer’ın ‘Program.cs’ dosyasında da yapılandırmada bulunmamız gerekmektedir.

#region MassTransit
builder.Services.AddMassTransit(configurator =>
{
    configurator.AddConsumer<Consumers.OrderCreatedEventConsumer>().Endpoint(e => e.InstanceId = "queue");

    configurator.UsingAmazonSqs((context, _configurator) =>
    {
        _configurator.Host(Amazon.RegionEndpoint.APSouth1.OriginalSystemName, hostConfigurator =>
        {
            hostConfigurator.AccessKey(builder.Configuration["AWS:AccessKey"]);
            hostConfigurator.SecretKey(builder.Configuration["AWS:SecretKey"]);
        });

        _configurator.ConfigureEndpoints(context, new KebabCaseEndpointNameFormatter("masstransit", false));
    });
});
#endregion

Burada MassTransit konfigürasyonu açısından önceki içeriklerimizden tecrübe ettiklerimizin dışında Endpoint ve ConfigureEndpoints metotlarıyla birlikte KebabCaseEndpointNameFormatter sınıfını görmekteyiz. Hemen bunların izahını yaparsak eğer;

• Endpoint, ilgili consumer için oluşturulacak kuyruk endpoint’ine bir InstanceId değeri ekler. Böylece oluşturulan endpoint benzersiz bir hal alacaktır. Dolayısıyla bir consumer’ı birden fazla yerde kullanıyorsanız, onları ayırt etmek için InstanceId kullanılmaktadır diyebiliriz. Ayrıca bu değer kuyruğun sonuna yansıyacaktır.
• ConfigureEndpoints, consumer’ların MassTransit’in default kurallarına göre otomatik olarak yapılandırılmasını sağlar. Bu yapılandırma sürecinde formatter olarak verdiğimiz sınıfların niteliğine göre davranış sergiler. Örnekte KebabCaseEndpointNameFormatter sınıfından istifade edilmiştir. Bu sınıf, endpoint(yani kuyruk) isimlendirmesini consumer’ın event’ini baz alarak oluşturacak ve kebab-case formatında olmasını sağlayacaktır. Misal olarak OrderCreatedEvent‘i consume eden bir consumer, bu sınıf sayesinde ‘order-created-event’ kuyruğuna sahip olacaktır. Ayrıca ConfigureEndpoints metodu, kuyruk isimlerine önek eklememizi de sağlamaktadır. Buradaki ‘masstransit’ değeri ile oluşturulacak kuyruk
  ‘masstransit-order-created-event’ şeklinde olacaktır. Ha bir yandan da Endpoint metodu InstanceId olarak ‘queue’ değerini ekleyeceği için nihai olarak kuyruk adı
  ‘masstransit-order-created-event-consumer-queue’ şeklinde olacaktır.

Velhasıl, şimdi de OrderCreatedEvent‘i tüketecek olan consumer’ı oluşturalım.

    public class OrderCreatedEventConsumer : IConsumer<OrderCreatedEvent>
    {
        public Task Consume(ConsumeContext<OrderCreatedEvent> context)
        {
            Console.WriteLine($"{context.Message.OrderId} - {context.Message.CustomerId}");
            return Task.CompletedTask;
        }
    }

Makale boyunca AWS kanadında işlemlerin gerçekleştirilebilmesi için IAM servisinde gerekli olan AmazonSNSFullAccess ve AmazonSQSFullAccess izinlerini kapsayan politikanın yapılandırılması ve permissions boundary olarak da AdministratorAccess belirlenmesi gerekmektedir. Ya da aşağıdaki gibi komple IAM Politikalarının kullanılması yeterlidir…

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "SqsAccess",
            "Effect": "Allow",
            "Action": [
                "sqs:SetQueueAttributes",
                "sqs:ReceiveMessage",
                "sqs:CreateQueue",
                "sqs:DeleteMessage",
                "sqs:SendMessage",
                "sqs:GetQueueUrl",
                "sqs:GetQueueAttributes",
                "sqs:ChangeMessageVisibility",
                "sqs:PurgeQueue",
                "sqs:DeleteQueue",
                "sqs:TagQueue"
            ],
            "Resource": "arn:aws:sqs:*:YOUR_ACCOUNT_ID:*"
        },{
            "Sid": "SnsAccess",
            "Effect": "Allow",
            "Action": [
                "sns:GetTopicAttributes",
                "sns:CreateTopic",
                "sns:Publish",
                "sns:Subscribe"
            ],
            "Resource": "arn:aws:sns:*:YOUR_ACCOUNT_ID:*"
        },{
            "Sid": "SnsListAccess",
            "Effect": "Allow",
            "Action": [
                "sns:ListTopics"
            ],
            "Resource": "*"
        }
    ]
}

Ve son olarak her iki uygulamanın ‘appsettings.json’ dosyasında aşağıdaki yapılandırmalarda bulunalım;

  "AWS": {
    "Profile": "sqs-profile",
    "Region": "ap-south-1",
    "AccessKey": "***",
    "SecretKey": "***"
  }

İşte bu yapılandırmalar ve çalışmalardan sonra uygulamayı topyekün derleyip çalıştırdığımızda MassTransit aracılığıyla başarılı bir haberleşmenin sağlandığını tecrübe etmiş olacağız.

İçerik boyunca oluşturduğumuz kuyruklara göz atarsak eğer aşağıdaki gibi gözükecektir.

Nihai olarak;
Amazon SQS hakkında bilgi edinmiş ve kuyruk yapısına dair temel kavramları keşfetmiş bulunuyoruz. Artık ihtiyaç noktalarında RabbitMQ ve Kafka teknolojilerine karşın Amazon SQS’i bir alternatif olarak değerlendirebilir ve Amazon’un sunduğu avantajlardan istifade edebiliriz.

İlgilenenlerin faydalanması dileğiyle…
Sonraki yazılarımda görüşmek üzere…
İyi çalışmalar..

Not : Örnek çalışmanın github adresine aşağıdan erişebilirsiniz.
https://github.com/gncyyldz/AmazonSQS.Example