Consumer ile Error Handling<\/h2>\n
Bir e-commerce sistemi \u00fczerinde\u00a0\u00e7al\u0131\u015ft\u0131\u011f\u0131m\u0131z\u0131 d\u00fc\u015f\u00fcnelim. Sistem \u00fczerinde bir sipari\u015f ba\u015far\u0131yla ger\u00e7ekle\u015fti\u011fi an, sistem taraf\u0131ndan “IPaymentApproved” tipinde bir event f\u0131rlat\u0131ld\u0131\u011f\u0131n\u0131 varsayal\u0131m. “IPaymentApproved” event’\u0131n\u0131 dinleyen farkl\u0131 consumer’lar olabilece\u011fi gibi,\u00a0bunlardan birisinin \u00f6rne\u011fimiz gere\u011fi sipari\u015f detaylar\u0131n\u0131 m\u00fc\u015fteriye e-mail g\u00f6nderen bir servis oldu\u011funu d\u00fc\u015f\u00fcnerek ele alal\u0131m.<\/p>\n “IPaymentApproved” event’\u0131, a\u015fa\u011f\u0131daki gibi\u00a0tan\u0131ml\u0131 olsun.<\/p>\n “IPaymentApproved” event’\u0131n\u0131 dinleyen consumer ise, \u015fimdilik a\u015fa\u011f\u0131daki gibi implemente edilmi\u015f olsun.<\/p>\n Peki consume i\u015flemi s\u0131ras\u0131nda, herhangi bir exceptional durum ile kar\u015f\u0131la\u015f\u0131rsak ne olacak? Bu exception hatal\u0131 kodlamadan kaynakl\u0131 olabilece\u011fi gibi,\u00a0eksik<\/strong>\/tutars\u0131z<\/strong> produce edilen bir message’dan kaynakl\u0131 da olabilir.\u00a0Bu tarz durumlarda\u00a0kullan\u0131lmak \u00fczere MassTransit i\u00e7erisinde implemente edilmi\u015f baz\u0131 \u00e7\u00f6z\u00fcmler mevcuttur.<\/p>\n Default bus ile exception MassTransit middleware’i taraf\u0131ndan yakalan\u0131r ve ilgili message, “{queue_name}_error<\/em>” isimli bir queue’ya otomatik olarak ta\u015f\u0131n\u0131r. Exception’\u0131n detaylar\u0131na ise ta\u015f\u0131nan message’\u0131n header’\u0131 \u00fczerinden eri\u015febilmek m\u00fcmk\u00fcnd\u00fcr. Bu tarz problemlerden haberdar olabilmek i\u00e7in ise olu\u015fan “{queue_name}<\/em>_error<\/em>” gibi \u00f6nemli queue’lar\u0131, \u00f6zellikle production ortamlar\u0131 i\u00e7in network monitoring tool’lar\u0131 ile sensor<\/strong> ekleyerek takip etmek, do\u011fru bir hareket olacakt\u0131r.<\/p>\n Baz\u0131 durumlarda bir web service’in o an cevap verememesi, database taraf\u0131nda deadlock olu\u015fabilmesi gibi transient exceptionlar ger\u00e7ekle\u015febilmektedir. Bu tarz durumlarda ilgili message’\u0131 o an kaybetmemek i\u00e7in kullanabilece\u011fimiz baz\u0131 retrying<\/em> function’lar\u0131 bulunmaktad\u0131r.<\/p>\n \u00d6rne\u011fin:<\/p>\n Yukar\u0131daki kod blo\u011funa bakal\u0131m ve e-mail g\u00f6nderim i\u015flemi\u00a0s\u0131ras\u0131nda kullan\u0131yor\u00a0oldu\u011fumuz e-mail provider’\u0131na ait web service’in, cevap vermedi\u011fini ve bu y\u00fczden transient bir exception olu\u015ftu\u011funu d\u00fc\u015f\u00fcnelim. Bu tarz durumlar genelde ikinci defa denemeler sonucunda ba\u015far\u0131yla ger\u00e7ekle\u015febilmektedir.<\/p>\n Retry i\u015flemini devreye sokabilmek i\u00e7in a\u015fa\u011f\u0131daki gibi consumer’\u0131 initialize ederken, “UseRetry” method’unu kullanmak yeterli olacakt\u0131r.<\/p>\n “e.UseRetry(r => r.Immediate(5));<\/em>” k\u0131sm\u0131na dikkat edersek, “Immediate” policy’si ile herhangi bir exception an\u0131nda ilgili message’\u0131 error queue<\/strong>‘ya ta\u015f\u0131madan \u00f6nce, “5” kere daha denemesi gerekti\u011fini bildirdik. “Immediate” gibi birden fazla retry policy’leri bulunmaktad\u0131r. Bu policy’ler aras\u0131nda benim en yayg\u0131n olarak kulland\u0131\u011f\u0131m ise, “Incremental” retry policy olan\u0131d\u0131r. Bu policy ile ne kadar retry i\u015fleminin tekrar yap\u0131laca\u011f\u0131n\u0131, her denemenin ard\u0131ndan ne kadar s\u00fcre beklenece\u011fi ve her bekleme s\u00fcresinin de incremental olarak ne kadar artmas\u0131 gerekti\u011fi gibi bilgileri de set edebilmekteyiz.<\/p>\n Bununla ilgili \u00f6rne\u011fi ise, MassTransit’in s\u0131k kullan\u0131yor oldu\u011fum\u00a0function’lar\u0131n\u0131 wrapledi\u011fim MetroBus<\/a><\/em> library’si \u00fczerinden g\u00f6stermek istiyorum.<\/p>\n Bunlara ek olarak MassTransit \u00fczerinde bulunan di\u011fer retry policy’ler ise:<\/p>\n \u015feklindedir.<\/p>\n Bir di\u011fer yandan\u00a0retrying i\u015flemleri s\u0131ras\u0131nda meydana gelen exception’lar\u0131, retry filter<\/strong>‘lar\u0131n\u0131 kullanarak handle ve ignore edebilmek de m\u00fcmk\u00fcnd\u00fcr.<\/p>\n \u00d6rne\u011fin:<\/p>\n Daha detayl\u0131 bilgilere\u00a0ise buradan<\/a><\/em> ula\u015fabilirsiniz. Dilerseniz retry i\u015flemi i\u00e7in bir \u00f6rnek ger\u00e7ekle\u015ftirelim.<\/p>\n NOT<\/strong>: Makale boyunca consumer’a y\u00f6nelik \u00f6rnekleri ger\u00e7ekle\u015ftirebilmek i\u00e7in, queue \u00fczerine “IPaymentApproved” interface’ini kullanarak bir ka\u00e7 message produce edece\u011fim.<\/p>\n<\/blockquote>\n Bunun i\u00e7in \u00f6ncelikle “PaymentApprovedConsumer” class’\u0131n\u0131, a\u015fa\u011f\u0131daki gibi initialize edelim.<\/p>\n ve ard\u0131ndan “Program.cs” i\u00e7erisini a\u015fa\u011f\u0131daki gibi kodlay\u0131p, start’a basal\u0131m.<\/p>\n Console \u00e7\u0131kt\u0131s\u0131na bakt\u0131\u011f\u0131m\u0131zda, consume ederken bir exception throw oldu ve bu i\u015flem sonucunda retry<\/strong> mekanizmas\u0131 devreye girdi.<\/p>\n Queue \u00fczerine bakt\u0131\u011f\u0131m\u0131zda ise:<\/p>\n message’\u0131n “unacked” olarak bekledi\u011fini g\u00f6rebiliriz.<\/p>\n Incremental retry policy ile “1” er dakika arayla toplam “3” kere retry i\u015flemi ger\u00e7ekle\u015ftirilecektir\u00a0ve her bir denemenin ard\u0131ndan interval time’a\u00a0“1” er dakika daha eklenerek, retry i\u015flemi retry limit’e kadar\u00a0devam\u00a0edecektir. Retry limit’e ula\u015f\u0131ld\u0131\u011f\u0131nda hala failure durumu mevcut ise a\u015fa\u011f\u0131daki gibi message, error queue’ya ta\u015f\u0131nacakt\u0131r.<\/p>\n MassTransit i\u00e7erisindeki exception handling unsurlar\u0131ndan \u00f6nemli birisi de, Circuit Breaker pattern’\u0131d\u0131r. Kullan\u0131m amac\u0131 ise belirli bir zaman diliminde sistem hata vermeye ba\u015flad\u0131\u011f\u0131nda ve failure threshold de\u011ferine ula\u015f\u0131ld\u0131\u011f\u0131nda, resource’lar\u0131 bu failure state durumlar\u0131ndaki overloaded’dan\u00a0koruyabilmek i\u00e7indir diyebilirim.<\/p>\n Bu pattern’\u0131n genel ak\u0131\u015f\u0131 ise a\u015fa\u011f\u0131daki gibidir:<\/p>\n \u00d6rne\u011fin “IPaymentApproved” event’\u0131n\u0131 consume ederken, e-mail g\u00f6nderim i\u015flemi s\u0131ras\u0131nda kulland\u0131\u011f\u0131m\u0131z \u00a0e-mail provider’\u0131na ait web service’in bir s\u00fcre cevap vermedi\u011fini ve request’lerin 30 sn gibi bir s\u00fcre sonunda timed out’a u\u011frad\u0131\u011f\u0131n\u0131 d\u00fc\u015f\u00fcnelim. Peki ya t an\u0131nda bir \u00e7ok event olu\u015fursa ne olacak? T\u00fcm request’ler 30 sn timeout s\u00fcreleri sonunda failure state’ine u\u011frayacaklar ve resource’u gereksiz bir yere me\u015fgul etmi\u015f olaca\u011f\u0131z ve ayn\u0131 zamanda cascading failure<\/strong>‘lara da sebebiyet vermi\u015f olabilece\u011fiz.<\/p>\n Peki ya circuit breaker’\u0131 a\u00e7arsak bu durumda ne olacak?<\/p>\n Bu durumda circuit breaker bizim i\u00e7in failure durumlar\u0131n\u0131 monitor edecektir. Failures durumlar\u0131 belirli bir threshold seviyesine ula\u015ft\u0131\u011f\u0131nda, circuit breaker devreye girecek ve repeated failures’lara sebebiyet veren durumlar\u0131 ge\u00e7ici olarak engelleyecektir. \u00c7al\u0131\u015fma mant\u0131\u011f\u0131 da ise\u00a0reset interval<\/strong>\u00a0time expire oldu\u011funda,\u00a0consume i\u015flemi yava\u015f bir \u015fekilde \u00e7al\u0131\u015fmaya devam edecektir. E\u011fer bu durum kar\u015f\u0131s\u0131nda failure tekrar devam ederse,\u00a0timeout interval\u00a0reset’lenir ve circuit breaker tekrar a\u00e7\u0131k moda getirilir. E\u011fer herhangi bir failure ile kar\u015f\u0131la\u015f\u0131lmazsa, t\u00fcm consume flow’u normal hale getirilir.<\/p>\n MassTransit \u00fczerinde\u00a0circuit breaker’\u0131, a\u015fa\u011f\u0131daki gibi etkinle\u015ftirebilmek m\u00fcmk\u00fcnd\u00fcr.<\/p>\n ayr\u0131ca MetroBus taraf\u0131nda ise, a\u015fa\u011f\u0131daki gibi etkinle\u015ftirilebilinir:<\/p>\n Circuit breaker hakk\u0131nda daha detayl\u0131 bilgiye, buradan<\/a><\/em> ula\u015fabilirsiniz.<\/p>\n Consumer initialize ederken dikkat edilmesi gereken bir di\u011fer unsur ise,\u00a0rate limiter kullan\u0131m\u0131d\u0131r. Kullan\u0131m\u0131ndaki as\u0131l ama\u00e7 ise: consumer i\u00e7erisinde e-mail g\u00f6nderim i\u015flemi i\u00e7in kulland\u0131\u011f\u0131m\u0131z provider, saniye ba\u015f\u0131na belirli bir say\u0131da call limiti kabul ediyor olabilir. Bu gibi durumlarda rate limiter kullan\u0131labilmektedir.<\/p>\n A\u015fa\u011f\u0131daki gibi etkinle\u015ftirebilmek m\u00fcmk\u00fcnd\u00fcr.<\/p>\n MetroBus \u00fczerinde ise:<\/p>\n “UseRateLimiter()” \u0131 kullanarak fluently bir \u015fekilde etkinle\u015ftirebilmek m\u00fcmk\u00fcnd\u00fcr.<\/p>\n Son zamanlarda baz\u0131 business case’ler do\u011frultusunda kar\u015f\u0131la\u015ft\u0131\u011f\u0131m belkide en \u00f6nemli concern’lerden birisi de, hen\u00fcz i\u015flemememiz gereken\u00a0message’lar\u0131 nas\u0131l handle edece\u011fiz konusu olmu\u015ftur.<\/p>\n Build-in olarak RabbitMQ i\u00e7erisinde AMQP standartlar\u0131nda deferred<\/strong> i\u015flemler bulunmamaktad\u0131r. Fakat MassTransit i\u00e7erisinde bunu ger\u00e7ekle\u015ftirebilmenin bir ka\u00e7 yolu vard\u0131r.<\/p>\n Bu konudaki \u00f6rne\u011fi benimde kulland\u0131\u011f\u0131m “RabbitMQ Delayed Message Plugin” ile ger\u00e7ekle\u015ftirece\u011fim. Bunun i\u00e7in \u00f6ncelikle buradan<\/a><\/em>\u00a0“rabbitmq_delayed_message_exchange” plugin’ini indirelim ve kurulumunu ger\u00e7ekle\u015ftirelim.<\/p>\n NOT<\/strong>: \u0130ndirilen ilgili plugin’i RabbitMQ i\u00e7erisindeki “plugins” klas\u00f6r\u00fcne atarak, “rabbitmq-plugins enable rabbitmq_delayed_message_exchange” komutu ile CLI \u00fczerinden etkinle\u015ftirebilmek m\u00fcmk\u00fcnd\u00fcr.<\/p>\n<\/blockquote>\n Plugin’i etkinle\u015ftirmemizin ard\u0131ndan, MassTransit i\u00e7erisinden a\u015fa\u011f\u0131daki gibi initialize edebiliriz.<\/p>\n MetroBus i\u00e7erisinde ise:<\/p>\n yukar\u0131daki gibi initialize etmek, yeterli olacakt\u0131r.<\/p>\n “PaymentApprovedConsumer” class’\u0131n\u0131 ise, a\u015fa\u011f\u0131daki gibi refactor edelim.<\/p>\n Deferred i\u015flemlerde dikkat etmemiz gereken bir di\u011fer nokta ise, “MT-Redelivery-Count” key’i ile message’\u0131n header’\u0131 \u00fczerinden ald\u0131\u011f\u0131m\u0131z redelivery count<\/strong> bilgisidir. Bu bilgi sayesinde message’\u0131 ka\u00e7 kere retry etti\u011fimizi \u00f6\u011frenebilir, infinite retry <\/strong>i\u015flemlerinden de ka\u00e7\u0131nabilmi\u015f oluruz.<\/p>\n Consumer’\u0131 start edelim ve sonucuna bir bakal\u0131m.<\/p>\n Console \u00e7\u0131kt\u0131s\u0131na bakt\u0131\u011f\u0131m\u0131zda, “context.Defer(TimeSpan.FromMinutes(1))<\/em>”\u00a0kod sat\u0131r\u0131 ile set etmi\u015f oldu\u011fumuz “1” er dakikal\u0131k defer i\u015flemi, ba\u015far\u0131yla ger\u00e7ekle\u015fmi\u015f ve sonunda i\u015flemi tamamlayamad\u0131\u011f\u0131 i\u00e7in ilgili message’\u0131 error queue \u00fczerine y\u00f6nlendirmi\u015ftir.<\/p>\n Bu i\u015flem s\u0131ras\u0131nda RabbitMQ’nun management ekran\u0131na bakt\u0131\u011f\u0131m\u0131zda ise:<\/p>\n![\"\"](\"\/wp-content\/uploads\/2017\/04\/masstransit-error-handling.png\"){kind=spacer}
<\/a><\/p>\nnamespace ErrorAndRedeliverHandlingSample.Contracts\r\n{\r\n public interface IPaymentApproved\r\n {\r\n string OrderNumber { get; set; }\r\n }\r\n}<\/pre>\nnamespace ErrorAndRedeliverHandlingSample\r\n{\r\n public class PaymentApprovedConsumer : IConsumer1. Retrying Messages<\/h4>\n
namespace ErrorAndRedeliverHandlingSample\r\n{\r\n public class PaymentApprovedConsumer : IConsumerIBusControl busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>\r\n{\r\n cfg.ReceiveEndpoint(host, \"queue_name\", e =>\r\n {\r\n e.UseRetry(r => r.Immediate(5));\r\n e.Consumer(() => new PaymentApprovedConsumer());\r\n });\r\n}<\/pre>\nIBusControl busControl = MetroBusInitializer.Instance.UseRabbitMq(\"rabbitMqUri\", \"rabbitMqUserName\", \"rabbitMqPassword\")\r\n .UseIncrementalRetryPolicy(retryLimit: 5, initialIntervalFromMinute: 10, intervalIncrementFromMinute: 10)\r\n .InitializeConsumer
\n
cfg.UseRetry(r => \r\n{\r\n c.Ignore(typeof(InvalidOperationException), typeof(InvalidCastException));\r\n});<\/pre>\n\n
namespace ErrorAndRedeliverHandlingSample\r\n{\r\n public class PaymentApprovedConsumerService\r\n {\r\n private readonly IBusControl _consumerBusControl;\r\n private readonly string _rabbitMqUri;\r\n private readonly string _rabbitMqUserName;\r\n private readonly string _rabbitMqPassword;\r\n private readonly string _queueName;\r\n\r\n public PaymentApprovedConsumerService()\r\n {\r\n _rabbitMqUri = ConfigurationManager.AppSettings[\"RabbitMqUri\"];\r\n _rabbitMqUserName = ConfigurationManager.AppSettings[\"RabbitMqUserName\"];\r\n _rabbitMqPassword = ConfigurationManager.AppSettings[\"RabbitMqPassword\"];\r\n _queueName = ConfigurationManager.AppSettings[\"FooQueue\"];\r\n\r\n _consumerBusControl =\r\n MetroBusInitializer.Instance.UseRabbitMq(_rabbitMqUri, _rabbitMqUserName, _rabbitMqPassword)\r\n .UseIncrementalRetryPolicy(3, 1, 1)\r\n .InitializeConsumernamespace ErrorAndRedeliverHandlingSample\r\n{\r\n class Program\r\n {\r\n static void Main(string[] args)\r\n {\r\n var paymentApprovedConsumerService = new PaymentApprovedConsumerService();\r\n\r\n paymentApprovedConsumerService.Start();\r\n }\r\n }\r\n}<\/pre>\n<\/a><\/p>\n<\/a><\/p>\n<\/a><\/p>\n2. Circuit Breaker Kullan\u0131m\u0131<\/h4>\n
<\/a><\/p>\nIBusControl busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>\r\n{\r\n cfg.ReceiveEndpoint(host, \"queue_name\", e =>\r\n {\r\n e.UseCircuitBreaker(cb =>\r\n {\r\n cb.TripThreshold = 15;\r\n cb.ActiveThreshold = 10;\r\n cb.ResetInterval = TimeSpan.FromMinutes(5);\r\n });\r\n e.Consumer(() => new PaymentApprovedConsumer());\r\n });\r\n}<\/pre>\nIBusControl busControl = MetroBusInitializer.Instance.UseRabbitMq(\"rabbitMqUri\", \"rabbitMqUserName\", \"rabbitMqPassword\")\r\n .UseCircuitBreaker(tripThreshold: 15, activeThreshold: 10, resetInterval: 5)\r\n .InitializeConsumer
3. Rate Limiter Kullan\u0131m\u0131<\/h4>\n
IBusControl busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>\r\n{\r\n cfg.ReceiveEndpoint(host, \"queue_name\", e =>\r\n {\r\n e.UseRateLimit(20, TimeSpan.FromSeconds(5));\r\n e.Consumer(() => new PaymentApprovedConsumer());\r\n });\r\n}<\/pre>\nIBusControl busControl = MetroBusInitializer.Instance.UseRabbitMq(\"rabbitMqUri\", \"rabbitMqUserName\", \"rabbitMqPassword\")\r\n .UseRateLimiter(rateLimit: 20, interval: 5000)\r\n .InitializeConsumer
Redeliver Messages<\/h2>\n
<\/a><\/p>\n\n
\n
IBusControl busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>\r\n{\r\n cfg.UseDelayedExchangeMessageScheduler();\r\n}<\/pre>\nIBusControl busControl = MetroBusInitializer.Instance.UseRabbitMq(\"rabbitMqUri\", \"rabbitMqUserName\", \"rabbitMqPassword\")\r\n .UseDelayedExchangeMessageScheduler()\r\n .InitializeConsumer
namespace ErrorAndRedeliverHandlingSample\r\n{\r\n public class PaymentApprovedConsumer : IConsumer<\/a><\/p>\n