Azure Storage Serisi #8 – Azure Queue Storage ve Asp.NET Core İle Kullanımı

Merhaba,

Bu içeriğimizde Azure Storage Yazı Dizisinin sonuncu makalesi olan Azure Queue Storage üzerine konuşuyor olacağız.

Azure Queue Storage Nedir?

Azure Queue Storage, bir mesaj kuyruk sistemidir. Mesaj kuyruk sistemleri; uzun süren işlemlerin arka planda bir kuyruğa atılarak o kuyruğu dinleyen başka uygulamalar tarafından eşzamansız bir şekilde işlenmesini sağlayan yapılardır. Bknz: RabbitMQ

Azure Queue Storage’a gönderilen mesajın boyutu en fazla 64 KB olabilir. Her bir mesajın default ömrü 7 gündür. 7 gün sonrasında mesajlar otomatik silinir. Mesaj ömürlerini custom bir şekilde değiştirebilir yahut -1 değerini vermek kaydıyla ömür boyu kuyrukta tutabilirsiniz.

Azure Queue Storage Kavramları Nelerdir?

URL Format
Azure Queue Storage’da ki mesaj kuyruklarını aşağıdaki formatta endpoint ile url biçiminde temsil edebilirsiniz.
https://<account_name>.queue.core.windows.net/<queue_name>

Örneğin yukarıda paylaşılan şemadaki mesaj kuyruğunu şöyle temsil edebilirsiniz;
https://myaccount.queue.core.windows.net/images-to-download

Storage account
Azure Queue Storage, bir Azure Storega servisidir. Dolayısıyla tüm işlemler bir storage hesabı üzerinden gerçekleştirilir.

Queue
Bir dizi mesaj içeren yapıdır. Queue isimleri komple küçük harf olmalıdır.

Message
Herhangi bir formatta 64 KB’a kadar varan mesajlardır.

Visibility Özelliği(Görünürlük)

Bir kuyruğa verilen mesajın aynı anda birden fazla consumer(tüketici) tarafından tüketilmemesi için default 30 saniye kuyrukta görünmez olmasıdır. Dolayısıyla ilgili mesajı 30 saniye içerisinde işlemek gerekmekte aksi taktirde başka consumerlar’a görünür olacağından dolayı eşzamanlı olarak farklı bir consumer tarafından işlenebilir olacaktır. Buradaki işleme süresinin yapılacak işin boyutuna göre tutarlı bir şekilde verilmesine özen gösterilmelidir. Görünürlük süresine ‘Visibility Time Out‘ denmektedir ve maksimum 7 gün almaktadır.

Asp.NET Core İle Azure Queue Storage Kullanımı

Asp.NET Core ile Azure Queue Storage kullanımı için aşağıdaki adımları sırasıyla gerçekleştirmeniz yeterli olacaktır;

• Adım 1
  Bir Asp.NET Core uygulaması oluşturunuz ve Azure.Storage.Queues kütüphanesini uygulamaya entegre ediniz.
  
• Adım 2
  ‘AzQueue’ isminde bir sınıf oluşturunuz ve içeriğini aşağıdaki gibi geliştiriniz.

      public class AzQueue
      {
          readonly QueueClient _queueClient;
          public AzQueue(string queueName)
          {
              _queueClient = new QueueClient("DefaultEndpointsProtocol=https;AccountName=gencayy******2Zsyk7Rk54Bs1BxP4MT7tTeHbNe/WljTCZQt7RfM****************ndows.net", queueName);
              _queueClient.CreateIfNotExists();
          }
          public async Task SendMessageAsync(string message)
          {
              if (await _queueClient.ExistsAsync())
                  await _queueClient.SendMessageAsync(message);
              else
                  throw new Exception("Mesaj gönderilmeye çalışılırken beklenmeyen bir hatayla karşılaşıldı.");
          }
          /// <summary>
          /// Mesajı kuyruktan alır ve siler.
          /// </summary>
          public async Task<QueueMessage> RetrieveNextMessageAsync()
          {
              if (await _queueClient.ExistsAsync())
              {
                  Response<QueueProperties> response = await _queueClient.GetPropertiesAsync();
                  //ApproximateMessagesCount : Önbelleğe alınan yaklaşık mesaj sayısını döner.
                  if (response.Value.ApproximateMessagesCount > 0)
                  {
                      QueueMessage[] queueMessages = await _queueClient.ReceiveMessagesAsync();
                      if (queueMessages.Any())
                          return queueMessages[0];
                  }
                  return null;
              }
              else
                  throw new Exception("Mesaj okunurken(Retrieve) beklenmeyen bir hatayla karşılaşıldı.");
          }
          /// <summary>
          /// Mesajı kuyruktan alır lakin silmez!
          /// </summary>
          public async Task<PeekedMessage> PeekMessageAsync()
          {
              if (await _queueClient.ExistsAsync())
              {
                  Response<QueueProperties> response = await _queueClient.GetPropertiesAsync();
                  //ApproximateMessagesCount : Önbelleğe alınan yaklaşık mesaj sayısını döner.
                  if (response.Value.ApproximateMessagesCount > 0)
                  {
                      PeekedMessage[] peekedMessages = await _queueClient.PeekMessagesAsync();
                      if (peekedMessages.Any())
                          return peekedMessages[0];
                  }
                  return null;
              }
              else
                  throw new Exception("Mesaj okunurken(Peek) beklenmeyen bir hatayla karşılaşıldı.");
          }
          public async Task UpdateMessageAsync(string newMessage)
          {
              if (await _queueClient.ExistsAsync())
              {
                  QueueMessage[] queueMessages = await _queueClient.ReceiveMessagesAsync();
                  //TimeSpan.FromSeconds(60) ile 60 saniye daha visiblty özelliğini genişleterek ilgili mesajı görünmez yapıyoruz.
                  await _queueClient.UpdateMessageAsync(queueMessages[0].MessageId, queueMessages[0].PopReceipt, newMessage, TimeSpan.FromSeconds(60));
              }
              else
                  throw new Exception("Mesaj güncellenirken beklenmeyen bir hatayla karşılaşıldı.");
          }
          public async Task DeleteMessageAsync()
          {
              if (await _queueClient.ExistsAsync())
              {
                  QueueMessage[] queueMessages = await _queueClient.ReceiveMessagesAsync();
                  await _queueClient.DeleteMessageAsync(queueMessages[0].MessageId, queueMessages[0].PopReceipt);
              }
              else
                  throw new Exception("Mesaj silinirken beklenmeyen bir hatayla karşılaşıldı.");
          }
          public async Task DeleteQueueAsync()
          {
              if (await _queueClient.ExistsAsync())
                  await _queueClient.DeleteAsync();
              else
                  throw new Exception("Kuyruk silinirken beklenmeyen bir hatayla karşılaşıldı.");
          }
      }
  

  Yukarıdaki kod bloğunu izah etmemiz gerekirse eğer;

  • 6. satırda, storage account’un connection string’i verilmektedir.
  • 12. satırda, ‘SendMessageAsync’ metodu ile kuyruğa mesaj gönderilmektedir. İlgili metodun 3. overload’ına göz atarsak eğer;
    
    ‘visibilityTimeout’ ve ‘timeToLive’ parametreleri mevcuttur. Bu parametreleri sırasıyla açıklarsak eğer;

    visibilityTimeout, kuyruktan mesaj alındığında o mesajın kuyrukta ne kadar süre görünmez olmasını istiyorsanız eğer buradan belirleyebilirsiniz. Default 30 saniyedir.

    timeToLive, mesajın yaşam süresini belirler. Default olarak 7 gündür. TimeSpan.FromSecond(-1) değeri ile ömür boyu mesaj tutulabilir.

  • 25. satırda, ‘ApproximateMessagesCount’ property’si önbelleğe alınan yaklaşık mesaj sayısını belirtmektedir. Böylece, yaklaşık olarak ne kadarlık mesaj üzerinde işlem yapabildiğimizi ölçmüş oluyoruz.
  • 27. satırda, ‘ReceiveMessagesAsync’ metodu ile kuyruktaki mesaj elde edilmekte ve kuyruktan silinmektedir.
  • 47. satırda, ‘PeekMessagesAsync’ metodu ise kuyruktaki mesajı elde etmekte lakin kuyruktan silmemektedir.
• Adım 3
  Geliştirilen bu sınıfı uygulamaya servis olarak ekleyiniz.

      public class Startup
      {
          .
          .
          .
          public void ConfigureServices(IServiceCollection services)
          {
              .
              .
              .
              services.AddSingleton<AzQueue>(new AzQueue("mymessage"));
              .
              .
              .
          }
  
          .
          .
          .
      }
  

  Yukarıdaki kod bloğuna bakarsanız eğer ilgili sınıf servis olarak eklenirken biryandan da constructor üzerinden mesaj kuyruğunun ismide belirtilmektedir.

• Adım 4
  Ardından isimleri ‘ProducerController’ ve ‘ConsumerController’ olmak üzere iki adet controller sınıfı oluşturunuz ve içeriklerini aşağıdaki gibi geliştiriniz.

  ProducerController;

      [Route("api/[controller]")]
      [ApiController]
      public class ProducerController : ControllerBase
      {
          readonly AzQueue _azQueue;
          public ProducerController(AzQueue azQueue)
          {
              _azQueue = azQueue;
          }
          [HttpGet("[action]/{messageText}")]
          public async Task<IActionResult> SendMessageAsync(string messageText)
          {
              try
              {
                  await _azQueue.SendMessageAsync(messageText);
                  return Ok(true);
              }
              catch (Exception ex)
              {
                  return Ok(ex.Message);
              }
          }
          [HttpGet("[action]/{newMessageText}")]
          public async Task<IActionResult> UpdateMessageAsync(string newMessageText)
          {
              try
              {
                  await _azQueue.UpdateMessageAsync(newMessageText);
                  return Ok(true);
              }
              catch (Exception ex)
              {
                  return Ok(ex.Message);
              }
          }
          [HttpGet("[action]")]
          public async Task<IActionResult> DeleteMessageAsync()
          {
              try
              {
                  await _azQueue.DeleteMessageAsync();
                  return Ok(true);
              }
              catch (Exception ex)
              {
                  return Ok(ex.Message);
              }
          }
          [HttpGet("[action]")]
          public async Task<IActionResult> DeleteQueueAsync()
          {
              try
              {
                  await _azQueue.DeleteQueueAsync();
                  return Ok(true);
              }
              catch (Exception ex)
              {
                  return Ok(ex.Message);
              }
          }
      }
  

  ConsumerController;

      [Route("api/[controller]")]
      [ApiController]
      public class ConsumerController : ControllerBase
      {
          readonly AzQueue _azQueue;
          public ConsumerController(AzQueue azQueue)
          {
              _azQueue = azQueue;
          }
          [HttpGet("[action]")]
          public async Task<IActionResult> RetrieveMessageAsync()
          {
              try
              {
                  QueueMessage queueMessage = await _azQueue.RetrieveNextMessageAsync();
                  return Ok(queueMessage?.MessageText);
              }
              catch (Exception ex)
              {
                  return Ok(ex.Message);
              }
          }
          [HttpGet("[action]")]
          public async Task<IActionResult> PeekMessageAsync()
          {
              try
              {
                  PeekedMessage peekedMessage = await _azQueue.PeekMessageAsync();
                  return Ok(peekedMessage?.MessageText);
              }
              catch (Exception ex)
              {
                  return Ok(ex.Message);
              }
          }
      }
  

Bu işlemlerden sonra artık uygulamayı test etmek için derleyip, çalıştırabilirsiniz.

Test Edelim

Süreçte kuyruğu Azure portal üzerinden takip edebilmek için storage account’unuzdan ‘Queues’e ve onun da ardından ilgili kuyruğa tıklayınız.

Görüldüğü üzere… Kuyruktaki tüm mesajlar burada listelenmektedir.

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

Not : Örnek projeyi indirmek için buraya tıklayınız.