Mikroservis + RabbitMQ uygulaması - 4

Spring cloud ile yazdığımız mikroservis uygulamamızın üzerine rabbitmq ve messajlaşma özelliği eklemeye kaldığımız yerden devam ediyoruz. Yazı dizisinin bu 4. yazısında rabbitmq 'da kuyruklanmış mesajları alabilmek için dinleyici servisi oluşturacağız. Spring initializr ile bir servis oluşturacağız ve spring AMQP kütüphanesi ile @RabbitListener metodlar yazacağız. Her zaman olduğu gibi büyük resmimiz burada dursun.

Mail servisi projesini oluşturmak

Sistemimizde Eureka 'ya bağlayacağımız fakat gateway ile erişmeyeceğimiz mail gönderimi servisimiz olacak. Bu servis için spring initializr 'dan aşağıdaki gibi bir proje oluşturarak başlayabiliriz.

Önceden bildiğiniz bağımlılıklardan hızlıca geçeyim. Spring web services bağımlılığı ile projemiz restful çağrılara cevap verebilir hale geliyor. Config client ile spring boot projesi ayarları config server 'dan ve dolayısı ile private github repo 'dan alabiliyor. Eureka discovery client sayesinde servisimiz eureka 'ya kayıt olacak ve ayakta olup olmadığını bileceğiz. Lombok ise sınıflara getter ve setter metodlarını otomatik ekleyecek. Ayrıca loglama için bize default SLF4J de sağlayacak bonus olarak. Bootstrap sayesinde config server 'a bağlanmadan önce set edilmesi gereken bilgileri boot aşamasında set edebileceğiz. Son olarak Spring RabbitMQ ise AMQP görevini üstleniyor. Bu kütüphane sayesinde rabbitmq 'daki kuyruklar dinlenebiliyor ve geri bildirimler gönderilebiliyor.

Pom dosyasına util projesi olarak kullandığımız ve constant değerleri tanımladığımız minik projeyi de ekleyeceğiz. Bu sayede alınan custom class 'larımızı da tanımış oluyoruz.

			
<dependency>
	<groupId>com.aldimbilet</groupId>
	<artifactId>util</artifactId>
	<version>0.0.1-SNAPSHOT</version>
</dependency>		
			
		

AMQP işine girişmeden önce config repository 'de ab-mailservice-local.properties isimli cloud config dosyasını oluşturmanız gerekiyor. Dilerseniz kendi repo 'nuza aktarmak için config repository 'yi indirebilirsiniz. Bu dosyanın içeriği ise eureka adresini tanımlamak için aşağıdaki gibi olmalı:

			
eureka.client.service-url.defaultZone=http://aldimbilet:eureka@localhost:4442/eureka
			
		

Sonrasında bootstrap.properties dosyasında profil adı, config server bağlantı bilgileri ve uygulama ismi vereceğiz. Aşağıdaki gibi.

			
spring.application.name=ab-mailservice
eureka.instance.instance-id=${spring.application.name}:${random.int(1,10000)}
spring.cloud.config.discovery.service-id=ab-config-server
spring.cloud.config.fail-fast=true
spring.cloud.config.username=aldimbilet
spring.cloud.config.password=config
spring.profiles.active=local
			
		

Application.properties dosyasında ise en önemli özelliklerden birini set edeceğiz. Mesajların alınıp işlenmesi sırasında hata alındığında aynı kuyrukta kalmaması ve deadletter kuyruğuna geçebilmesi için tekrar kuyruklama işlemini iptal edeceğiz. Aksi takdirde mesaj rabbitmq ile bu servis arasında gelir gider sonsuz döngüye gider.

			
server.port = 0
spring.rabbitmq.listener.simple.default-requeue-rejected=false
			
		

Bu sisteme lombok ile kolayca ekleyebileceğimiz slf4j 'yi de bonus olarak eklemek istiyorum. Rabbitmq içerisinde herhangi bir loglama veya ileri seviye monitorin yapmadığımız için console 'da mesajların ne durumda olduğunu görmek istiyorum. Bunun için logback-spring.xml isimli bir xml dosyasını resources klasörüne yani application.properties dosyasının yanına koyabilirsiniz. Aşağıdaki gibi bu loglama işlemine format verebilirsiniz.

			
<configuration debug="false">
	<include resource="/org/springframework/boot/logging/logback/base.xml" />
	<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
		<encoder>
			<pattern>My custom logger -> %msg%n</pattern>
		</encoder>
	</appender>
	<root level="info">
		<!-- Uygulamada info seviyesi ve üstündeki bütün log işlemleri için -->
		<!-- Yukarıdaki STDOUT isimli appender kendi formatı ile kullanılacak -->
		<appender-ref ref="STDOUT" />
	</root>
</configuration>	
			
		

Bu noktada sistemi bu yazımdan yardım alarak ayağa kaldırabilirsiniz. Hatta mail servisini de çalıştırıp Eureka 'da görebilirsiniz. Fakat bu servisin amacı rabbitmq dinlemek olduğu için esas işi şimdi başlıyor.

RabbitListener metodlar

Bu serviste mail gönderimi işlemini yapıyormuş gibi yapacağımız için tek bir class olması yeterli oluyor. Ayrıca DB işlemleri veya güvenlik önemleri almadığımız için config veya repo sınıflara da ihtiyaç duymuyoruz. Konuyu rabbitmq üzerinde yoğunlaştırmak adına bu servisi olabildiğince basit tutuyorum. Bu servise MVC uygulamasından mail gönderilecek kişi bilgisini göndermiştik. Bu bizim custom sınıfımızdı ve util projesinde bu sınıf tanımlı idi. Zaten bu projenin bağımlılığını da pom doyasına yukarıda ekledik.

Main sınıfında etkinlik iptali ve ödeme yapılması durumları için gelen mesajları dinleyecek 2 metodumuz olacak. Detayları bu sefer kodun içerisinde comment 'ler olarak yazdım. Ayrıca SLF4J 'yi Lombok ile inject ettim ve kullanmış oldum. Bu servisin bütün kodları aşağıdaki gibi :)

			
import java.util.Random;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.handler.annotation.Payload;
import com.aldimbilet.pojos.UserInfoPojo;
import lombok.extern.slf4j.Slf4j;

// Lombok SLF4J logger sınıfını inject ediyor
// Bu anotasyon lombok.extern.slf4j.Slf4j anotasyonu
// Bu şekilde istediğiniz yerde log.info veya log.error şeklinde direct olarak kullanabiliyorsunuz
@Slf4j
@SpringBootApplication
public class AldimbiletMailServicesApplication
{
	public static void main(String[] args)
	{
		SpringApplication.run(AldimbiletMailServicesApplication.class, args);
	}

	// MVC uygulamasında olduğu gibi burada da custom sınıfları mesajlardan çevirebilmek gerekiyor
	// Bu bean 'i unutursanız "Listener method could not be invoked with the incoming message" hatası alırsınız
	// Root cause olarak "Cannot convert from [[B] to [<your custom class>] for GenericMessage" hatasını görürsünüz
	@Bean
	public Jackson2JsonMessageConverter producerJackson2MessageConverter()
	{
		return new Jackson2JsonMessageConverter();
	}

	// Bu metod bir rabbitmq kuyruk dinleyicisidir ve direkt olarak exchange 'leri veya routing key 'leri dinleyemezsiniz
	// "emailReceiptQueue" isimli kuyruğu dinler
	// İlk giren ilk çıkar (FIFO) mantığı ile mesajları alır
	// "emailReceiptQueue" bir topic exchange kuyruğudur
	// "email.receipt.*" formatındaki routing key 'ler ile gelen mesajlar bu kuyruğa atılır
	// Bu sayede bu metod "email.receipt.special" veya "email.receipt.normal" routing key 'i ile gelen mesajları dinlemiş olur
	@RabbitListener(queues = "emailReceiptQueue")
	public void sendReceiptEmail(@Payload UserInfoPojo userInfo)
	{
		log.info("sending email receipt to '" + userInfo.getEmail() + "'");
		while (new Random().nextInt(99999999) != 55555555)
		{
			// Burada mail gönderimi için bir süre bekleniyor
			// Tamamen simulasyon amaçlı bir bekleme işlemi
			// Burada bir thread oluşturup sonucunu bekleyebilirsiniz
		}
		log.info("Handled the message and done with it");
		// RabbitMQ ack bekler yani bu metodun bitmesini bekler
		// Bu metod herhangi bir sonuç döndürmese bile ack gönderilir
		// Aspect oriented programming örneği gibi
		// Sonrasında rabbitmq sıradaki mesajı gönderir
	}

	// İptal maili bildirimlerini yani "emailCancelationQueue" kuyruğunu dinler
	// Bu kuyruk direct exchange kuyruğudur
	// Sadece "email.cancelation" routing key 'i ile gelen mesajlar buraya ulaşır
	@RabbitListener(queues = "emailCancelationQueue")
	public void sendCancelationEmail(@Payload UserInfoPojo userInfo) throws Exception
	{
		log.info("sending cancelation email to '" + userInfo.getEmail() + "'");
		while (new Random().nextInt(99999999) != 55555555)
		{
			// Yine bekleme işlemi
		}
		if (new Random().nextInt(10) == 3)
		{
			// Dead letter kuyruğa gitmesi için rastgele bir hata üretiliyor
			// Retry mekanizması olmadığı için ilk hatada kuyruktan çıkacak ve deadletter 'a ulaştırılacak
			// Rabbitmq tarafında kuyruk ayarlarını yapmıştık
			throw new Exception("bişeyler bişeyler");
		}
		log.info("Handled the message and done with it");
	}
}
			
		

Kodlardaki açıklamalardan da anlayacağınız gibi 3 kuyruktan 2 tanesini dinliyoruz. 3. kuyruk olan deadletter kuyruğunu rabbitmq kendisi yönetiyor ve exception döndüren mesajları retry etmeden direkt olarak oraya yönlendiriyor. Rabbitmq 'nun bir mesajı undeliverable yani iletilemez olarak değerlendirmesi demek bu mesajın alınmış fakat işlem sırasında hata almış olması gibi görünüyor. Çünkü bu hata alınması durumunda bile bir ack bilgisi gidiyor rabbitmq tarafına. Yani kuyruk dinlenmiyor olsaydı henüz iletim yapılmyacağı için iletilmez kategorisine girmeyecekti. Kuyruk dinlenmeye başlandı ve mesajlar gittiği halde hata ile geri döndü ise ve retry ayarı ile tekrar denemeler yapılmasına rağmen sonuç alınamıyorsa iletilemez mesajdır. Bunların yeri ise deadletter queue 'dur çünkü tekrar tekrar denenmesinin bir anlamı olmayacaktır. Best practice olmasa da belli bir süreden uzun süren işlemler de timeout ayarları ile undeliverable varsayılabilir.

Consumer tarafı bu şekilde tamamlanmış oluyor. Bu servisi ayağa kaldırdığınızda bir önceki yazıda gönderdiğiniz mesajları almaya başlar. Bazıları hata verir ve deadletter 'a düşer. Satış işleminde 2 farklı routing key olmasına rağmen aynı metoddan mail gönderilir. Tabi ki burada bir if koşulu işletilebilirdi farklı türden mail atılması için. Fakat bu durumda aynı kuyruğu kullanmak mantıksız olurdu. Amacımız bir topic 'e farklı routing key 'lerin ulaşabileceğini göstermekti. Bu servis ayağa kalktığında aşağıdaki gibi loglar görebilirsiniz.

Bu mesajlar alınırken beklemeler olacağı için bu kuyrukları rabbitmq yönetim konsolunda çalışıyor durumunda görebilirsiniz.

Konsolda logları gördüyseniz ve rabbitmq tarafında ready mesajlar kalmadıysa operasyon başarılı geçmiş ve mesajlar alınmış demektir. Bu şekilde rabbitmq 'yu utilize etmiş ve mesajları dinlemiş olduk. Sırada gerçekten doğru kararları aldık mı ve doğru şekilde kullanıyor muyuz gibi soruları incelemek ve savunmamızı yapmak geliyor. Rabbitmq gibi büyük bir ürünü aldım kullandım şeklinde açıklayamazsınız. Neden aldınız ve nasıl kullanıyorsunuz diye sorarlar. Yazacağımız kodlar bitti. Bir sonraki yazıda görüşmek üzere :)