اشتراک میکروسرویس ها در رویدادها

اولین گام برای استفاده از گذرگاه رویداد، اشتراک میکروسرویس ها برای رویدادهایی است که می خواهند دریافت کنند. این عملکرد باید در میکروسرویس های گیرنده انجام شود.

کد ساده زیر نشان می‌دهد که هر میکروسرویس گیرنده باید چه چیزی را هنگام راه‌اندازی سرویس (یعنی در کلاس Startup) پیاده‌سازی کند تا در رویدادهایی که نیاز دارد مشترک شود. در این مورد، میکروسرویس basket-api باید در ProductPriceChangedIntegrationEvent و پیام‌های OrderStartedIntegrationEvent مشترک شود.

به عنوان مثال، هنگام اشتراک در رویداد ProductPriceChangedIntegrationEvent، میکروسرویس سبد را از هرگونه تغییر در قیمت محصول آگاه می‌کند و به کاربر اجازه می‌دهد در صورت وجود آن محصول در سبد کاربر، درباره تغییر هشدار دهد.

var eventBus = app.ApplicationServices.GetRequiredService<IEventBus>();

eventBus.Subscribe<ProductPriceChangedIntegrationEvent,

ProductPriceChangedIntegrationEventHandler>();

eventBus.Subscribe<OrderStartedIntegrationEvent,

OrderStartedIntegrationEventHandler>();

پس از اجرای این کد، میکروسرویس مشترک از طریق کانال های RabbitMQ گوش می دهد. هنگامی که هر پیامی از نوع ProductPriceChangedIntegrationEvent می رسد، کد کنترل کننده رویداد را که به آن ارسال می شود فراخوانی می کند و رویداد را پردازش می کند.

انتشار رویدادها از طریق اتوبوس رویداد

در نهایت، فرستنده پیام (میکروسرویس مبدا) رویدادهای یکپارچه سازی را با کدی شبیه به مثال زیر منتشر می کند. (این رویکرد یک مثال ساده شده است که اتمی را در نظر نمی گیرد.) هر زمان که یک رویداد باید در چندین میکروسرویس منتشر شود، معمولاً بلافاصله پس از انجام داده ها یا تراکنش ها از میکروسرویس مبدا، کد مشابهی را پیاده سازی می کنید.

ابتدا، شی پیاده سازی گذرگاه رویداد (بر اساس RabbitMQ یا بر اساس یک گذرگاه سرویس) مانند کد زیر به سازنده کنترلر تزریق می شود:

[Route("api/v1/[controller]")]

public class CatalogController : ControllerBase

{

private readonly CatalogContext _context;

private readonly IOptionsSnapshot<Settings> _settings;

private readonly IEventBus _eventBus;

public CatalogController(CatalogContext context,

IOptionsSnapshot<Settings> settings,

IEventBus eventBus)

{

_context = context;

_settings = settings;

_eventBus = eventBus;

}

// ...

}

سپس آن را از روش های کنترلر خود استفاده می کنید، مانند روش UpdateProduct:

[Route("items")]

[HttpPost]

public async Task<IActionResult> UpdateProduct([FromBody]CatalogItem product)

{

var item = await _context.CatalogItems.SingleOrDefaultAsync(

i => i.Id == product.Id);

// ...

if (item.Price != product.Price)

{

var oldPrice = item.Price;

item.Price = product.Price;

_context.CatalogItems.Update(item);

var @event = new ProductPriceChangedIntegrationEvent(item.Id,

item.Price,

oldPrice);

// Commit changes in original transaction

await _context.SaveChangesAsync();

// Publish integration event to the event bus

// (RabbitMQ or a service bus underneath)

_eventBus.Publish(@event);

// ...

}

// ...

}

در این مورد، از آنجایی که میکروسرویس مبدا یک میکروسرویس ساده CRUD است، آن کد مستقیماً در یک کنترلر Web API قرار می گیرد.

در میکروسرویس های پیشرفته تر، مانند زمانی که از رویکردهای CQRS استفاده می شود، می توان آن را در کلاس CommandHandler، در متد Handle() پیاده سازی کرد.

طراحی اتمی و انعطاف پذیری هنگام انتشار در اتوبوس رویداد

هنگامی که رویدادهای یکپارچه سازی را از طریق یک سیستم پیام رسانی توزیع شده مانند گذرگاه رویداد خود منتشر می کنید، با مشکل به روز رسانی اتمی پایگاه داده اصلی و انتشار یک رویداد مواجه می شوید (یعنی یا هر دو عملیات کامل می شوند یا هیچ کدام از آنها انجام نمی شود). به عنوان مثال، در مثال ساده‌شده‌ای، زمانی که قیمت محصول تغییر می‌کند، کد داده‌ها را به پایگاه داده ارسال می‌کند و سپس پیام ProductPriceChangedIntegrationEvent را منتشر می‌کند. در ابتدا، ممکن است ضروری به نظر برسد که این دو عملیات به صورت اتمی انجام شوند. با این حال، اگر از یک تراکنش توزیع‌شده شامل پایگاه داده و کارگزار پیام استفاده می‌کنید، همانطور که در سیستم‌های قدیمی‌تر مانند صف‌بندی پیام‌های مایکروسافت (MSMQ) انجام می‌دهید، این رویکرد به دلایلی که در قضیه CAP توضیح داده شده است، توصیه نمی‌شود.

اساسا، شما از میکروسرویس ها برای ساختن سیستم های مقیاس پذیر و بسیار در دسترس استفاده می کنید. تا حدودی ساده تر، قضیه CAP می گوید که شما نمی توانید یک پایگاه داده (توزیع شده) (یا یک میکروسرویس که مدل آن را در اختیار دارد) بسازید که به طور مداوم در دسترس، به شدت سازگار و قابل تحمل برای هر پارتیشنی باشد. شما باید دو مورد از این سه ملک را انتخاب کنید.

در معماری‌های مبتنی بر ریزسرویس‌ها، شما باید در دسترس بودن و تحمل را انتخاب کنید و باید بر ثبات قوی بی‌تأکید کنید. بنابراین، در اکثر برنامه‌های کاربردی مبتنی بر میکروسرویس مدرن، معمولاً نمی‌خواهید از تراکنش‌های توزیع‌شده در پیام‌رسانی استفاده کنید، مانند زمانی که تراکنش‌های توزیع‌شده را بر اساس هماهنگ‌کننده تراکنش توزیع‌شده ویندوز (DTC) با MSMQ پیاده‌سازی می‌کنید.

به موضوع اولیه و مثال آن برگردیم. اگر سرویس پس از به‌روزرسانی پایگاه داده از کار بیفتد (در این مورد، درست بعد از خط کد با _context.SaveChangesAsync())، اما قبل از انتشار رویداد یکپارچه‌سازی، سیستم کلی ممکن است ناسازگار شود. این رویکرد بسته به عملیات تجاری خاصی که با آن سروکار دارید، ممکن است برای کسب و کار حیاتی باشد.

همانطور که قبلا در بخش معماری ذکر شد، می توانید چندین رویکرد برای مقابله با این موضوع داشته باشید:

• با استفاده از الگوی کامل رویداد منبع یابی.

• استفاده از استخراج لاگ تراکنش

• با استفاده از الگوی صندوق خروجی این یک جدول تراکنشی برای ذخیره رویدادهای ادغام (توسعه تراکنش محلی) است.

برای این سناریو، استفاده از الگوی کامل رویداد منبع یابی (ES) یکی از بهترین رویکردها است، اگر بهترین نباشد. با این حال، در بسیاری از سناریوهای برنامه، ممکن است نتوانید یک سیستم کامل ES را پیاده سازی کنید. ES به این معنی است که به جای ذخیره داده های وضعیت فعلی، فقط رویدادهای دامنه را در پایگاه داده تراکنشی خود ذخیره کنید. ذخیره فقط رویدادهای دامنه می تواند مزایای بزرگی داشته باشد، مانند داشتن تاریخچه سیستم شما و توانایی تعیین وضعیت سیستم خود در هر لحظه از گذشته. با این حال، پیاده سازی یک سیستم کامل ES مستلزم آن است که بیشتر سیستم خود را مجدداً معماری کنید و بسیاری از پیچیدگی ها و الزامات دیگر را معرفی کنید. به عنوان مثال، می‌خواهید از پایگاه داده‌ای که به‌طور خاص برای منبع‌یابی رویدادها ساخته شده است، مانند فروشگاه رویداد، یا پایگاه‌داده‌ای سند محور مانند Azure Cosmos DB، MongoDB، Cassandra، CouchDB یا RavenDB استفاده کنید. ES یک روش عالی برای این مشکل است، اما ساده ترین راه حل نیست، مگر اینکه قبلاً با منبع یابی رویداد آشنا باشید.

گزینه استفاده از استخراج گزارش تراکنش در ابتدا شفاف به نظر می رسد. با این حال، برای استفاده از این رویکرد، میکروسرویس باید با گزارش تراکنش RDBMS شما، مانند گزارش تراکنش SQL Server، کوپل شود. این رویکرد احتمالاً مطلوب نیست. اشکال دیگر این است که به‌روزرسانی‌های سطح پایین ثبت‌شده در گزارش تراکنش ممکن است در سطح رویدادهای یکپارچه‌سازی سطح بالا شما نباشند. اگر چنین است، فرآیند مهندسی معکوس آن عملیات گزارش تراکنش می تواند دشوار باشد.

یک رویکرد متعادل ترکیبی از جدول پایگاه داده تراکنشی و یک الگوی ES ساده شده است. می‌توانید از حالتی مانند «آماده برای انتشار رویداد» استفاده کنید که در رویداد اصلی وقتی آن را به جدول رویدادهای یکپارچه‌سازی متعهد می‌کنید، تنظیم می‌کنید. سپس سعی کنید رویداد را در اتوبوس رویداد منتشر کنید. اگر اقدام انتشار-رویداد با موفقیت انجام شود، تراکنش دیگری را در سرویس مبدا شروع می‌کنید و وضعیت را از «آماده برای انتشار رویداد» به «رویداد قبلاً منتشر شده» منتقل می‌کنید.

اگر اقدام انتشار-رویداد در گذرگاه رویداد ناموفق باشد، داده‌ها همچنان در ریزسرویس مبدا متناقض نخواهند بود - همچنان به‌عنوان «آماده برای انتشار رویداد» علامت‌گذاری می‌شود و با توجه به بقیه سرویس‌ها، در نهایت انجام می‌شود. مقاوم باش. همیشه می توانید کارهای پس زمینه ای داشته باشید که وضعیت تراکنش ها یا رویدادهای یکپارچه سازی را بررسی می کنند. اگر شغل رویدادی را در وضعیت «آماده انتشار رویداد» بیابد، می‌تواند سعی کند آن رویداد را در اتوبوس رویداد منتشر کند.

اگر اقدام انتشار-رویداد در گذرگاه رویداد ناموفق باشد، داده‌ها همچنان در ریزسرویس مبدا متناقض نخواهند بود - همچنان به‌عنوان «آماده برای انتشار رویداد» علامت‌گذاری می‌شود و با توجه به بقیه سرویس‌ها، در نهایت انجام می‌شود. مقاوم باش. همیشه می توانید کارهای پس زمینه ای داشته باشید که وضعیت تراکنش ها یا رویدادهای یکپارچه سازی را بررسی می کنند. اگر شغل رویدادی را در وضعیت «آماده انتشار رویداد» بیابد، می‌تواند سعی کند آن رویداد را در اتوبوس رویداد منتشر کند.

توجه داشته باشید که با این رویکرد، شما فقط رویدادهای یکپارچه‌سازی را برای هر میکروسرویس مبدا ادامه می‌دهید، و فقط رویدادهایی را که می‌خواهید با سایر ریزسرویس‌ها یا سیستم‌های خارجی ارتباط برقرار کنید. در مقابل، در یک سیستم کامل ES، شما تمام رویدادهای دامنه را نیز ذخیره می کنید.

بنابراین، این رویکرد متعادل یک سیستم ES ساده شده است. شما به فهرستی از رویدادهای ادغام با وضعیت فعلی آنها نیاز دارید ("آماده برای انتشار" در مقابل "منتشر شده"). اما شما فقط باید این حالت ها را برای رویدادهای ادغام اجرا کنید. و در این رویکرد، مانند یک سیستم ES کامل، نیازی به ذخیره تمام داده های دامنه خود به عنوان رویداد در پایگاه داده تراکنش ندارید.

اگر قبلاً از یک پایگاه داده رابطه‌ای استفاده می‌کنید، می‌توانید از جدول تراکنشی برای ذخیره رویدادهای ادغام استفاده کنید. برای دستیابی به اتمی در برنامه خود، از یک فرآیند دو مرحله ای مبتنی بر تراکنش های محلی استفاده می کنید. اساسا، شما یک جدول IntegrationEvent در همان پایگاه داده ای که موجودیت های دامنه خود را دارید دارید. آن جدول به عنوان بیمه ای برای دستیابی به اتمی عمل می کند، به طوری که شما رویدادهای یکپارچه سازی مداوم را در همان تراکنش هایی که داده های دامنه شما را متعهد می کنند، قرار دهید.

مرحله به مرحله، روند به این صورت پیش می رود:

• برنامه یک تراکنش پایگاه داده محلی را آغاز می کند.
• سپس وضعیت موجودیت های دامنه شما را به روز می کند و یک رویداد را در جدول رویداد یکپارچه سازی درج می کند.
• در نهایت، تراکنش را انجام می دهد، بنابراین شما اتمی مورد نظر را دریافت می کنید و سپس
• شما رویداد را به نحوی منتشر می کنید.

هنگام اجرای مراحل انتشار رویدادها، این انتخاب ها را دارید:

• رویداد ادغام را بلافاصله پس از انجام تراکنش منتشر کنید و از تراکنش محلی دیگری برای علامت گذاری رویدادها در جدول به عنوان منتشر شده استفاده کنید. سپس، از جدول فقط به عنوان یک مصنوع برای ردیابی رویدادهای یکپارچه سازی در صورت بروز مشکلات در میکروسرویس های راه دور استفاده کنید و اقدامات جبرانی را بر اساس رویدادهای یکپارچه سازی ذخیره شده انجام دهید.

• از جدول به عنوان نوعی صف استفاده کنید. یک رشته یا فرآیند برنامه جداگانه جدول رویداد ادغام را پرس و جو می کند، رویدادها را در گذرگاه رویداد منتشر می کند و سپس از یک تراکنش محلی برای علامت گذاری رویدادها به عنوان منتشر شده استفاده می کند.

برای سادگی، نمونه eShopOnContainers از رویکرد اول (بدون فرآیندهای اضافی یا ریزسرویس‌های جستجوگر) به همراه گذرگاه رویداد استفاده می‌کند. با این حال، نمونه eShopOnContainers همه موارد خرابی احتمالی را مدیریت نمی کند. در یک برنامه واقعی که در فضای ابری مستقر شده است، باید این واقعیت را بپذیرید که در نهایت مشکلاتی به وجود می آیند و باید منطق بررسی و ارسال مجدد را اجرا کنید. استفاده از جدول به عنوان یک صف می تواند موثرتر از رویکرد اول باشد، اگر آن جدول را به عنوان منبع واحد رویدادها در هنگام انتشار آنها (با کارگر) از طریق اتوبوس رویداد داشته باشید.

پیاده سازی اتمی در هنگام انتشار رویدادهای یکپارچه سازی از طریق اتوبوس رویداد

کد زیر نشان می دهد که چگونه می توانید یک تراکنش منفرد شامل چندین شیء DbContext ایجاد کنید - یک زمینه مربوط به داده های اصلی در حال به روز رسانی و زمینه دوم مربوط به جدول IntegrationEventLog است.

اگر اتصالات به پایگاه داده در زمان اجرای کد مشکلی داشته باشند، تراکنش در کد مثال زیر انعطاف پذیر نخواهد بود. این می تواند در سیستم های مبتنی بر ابر مانند Azure SQL DB اتفاق بیفتد، که ممکن است پایگاه داده را در سرورها جابجا کند.

برای وضوح، مثال زیر کل فرآیند را در یک قطعه کد نشان می دهد. با این حال، پیاده سازی eShopOnContainers مجدداً ساخته شده است و این منطق را به چندین کلاس تقسیم می کند تا نگهداری آن آسان تر باشد.

// Update Product from the Catalog microservice

//

public async Task<IActionResult> UpdateProduct([FromBody]CatalogItem productToUpdate)

{

var catalogItem =

await _catalogContext.CatalogItems.SingleOrDefaultAsync(i => i.Id ==

productToUpdate.Id);

if (catalogItem == null) return NotFound();

bool raiseProductPriceChangedEvent = false;

IntegrationEvent priceChangedEvent = null;

if (catalogItem.Price != productToUpdate.Price)

raiseProductPriceChangedEvent = true;

if (raiseProductPriceChangedEvent) // Create event if price has changed

{

var oldPrice = catalogItem.Price;

priceChangedEvent = new ProductPriceChangedIntegrationEvent(catalogItem.Id,

productToUpdate.Price,

oldPrice);

}

// Update current product

catalogItem = productToUpdate;

// Just save the updated product if the Product's Price hasn't changed.

if (!raiseProductPriceChangedEvent)

{

await _catalogContext.SaveChangesAsync();

}

else // Publish to event bus only if product price changed

{

// Achieving atomicity between original DB and the IntegrationEventLog

// with a local transaction

using (var transaction = _catalogContext.Database.BeginTransaction())

{

_catalogContext.CatalogItems.Update(catalogItem);

await _catalogContext.SaveChangesAsync();

await _integrationEventLogService.SaveEventAsync(priceChangedEvent);

transaction.Commit();

}

// Publish the integration event through the event bus

_eventBus.Publish(priceChangedEvent);

_integrationEventLogService.MarkEventAsPublishedAsync(

priceChangedEvent);

}

return Ok();

}

پس از ایجاد رویداد یکپارچه سازی ProductPriceChangedIntegrationEvent، تراکنشی که عملیات دامنه اصلی را ذخیره می کند (به روز رسانی مورد کاتالوگ) همچنین شامل ماندگاری رویداد در جدول EventLog می شود. این باعث می شود که یک تراکنش واحد باشد و شما همیشه می توانید بررسی کنید که آیا پیام های رویداد ارسال شده است یا خیر.

جدول ثبت رویداد به صورت اتمی با عملیات پایگاه داده اصلی، با استفاده از یک تراکنش محلی در برابر همان پایگاه داده به روز می شود. اگر هر یک از عملیات شکست بخورد، یک استثنا ایجاد می‌شود و تراکنش هر عملیات تکمیل‌شده را به عقب برمی‌گرداند، بنابراین هماهنگی بین عملیات دامنه و پیام‌های رویداد ذخیره شده در جدول حفظ می‌شود.

دریافت پیام از اشتراک‌ها: کنترل‌کننده‌های رویداد در میکروسرویس‌های گیرنده

علاوه بر منطق اشتراک رویداد، باید کد داخلی را برای کنترل‌کننده‌های رویداد ادغام پیاده‌سازی کنید (مانند یک روش برگشتی). کنترل کننده رویداد جایی است که شما تعیین می کنید که در آن پیام های رویداد از نوع خاصی دریافت و پردازش شوند.

یک کنترل کننده رویداد ابتدا یک نمونه رویداد را از گذرگاه رویداد دریافت می کند. سپس مؤلفه‌ای را که باید پردازش شود مربوط به آن رویداد یکپارچه‌سازی را تعیین می‌کند و رویداد را به‌عنوان تغییر حالت در میکروسرویس گیرنده منتشر و ادامه می‌دهد. به عنوان مثال، اگر یک رویداد ProductPriceChanged از میکروسرویس کاتالوگ منشا گرفته شود، در میکروسرویس سبد مدیریت می شود و وضعیت را در این میکروسرویس سبد گیرنده نیز تغییر می دهد، همانطور که در کد زیر نشان داده شده است.

namespace Microsoft.eShopOnContainers.Services.Basket.API.IntegrationEvents.EventHandling

{

public class ProductPriceChangedIntegrationEventHandler :

IIntegrationEventHandler<ProductPriceChangedIntegrationEvent>

{

private readonly IBasketRepository _repository;

public ProductPriceChangedIntegrationEventHandler(

IBasketRepository repository)

{

_repository = repository;

}

public async Task Handle(ProductPriceChangedIntegrationEvent @event)

{

var userIds = await _repository.GetUsers();

foreach (var id in userIds)

{

var basket = await _repository.GetBasket(id);

await UpdatePriceInBasketItems(@event.ProductId, @event.NewPrice, basket);

}

}

private async Task UpdatePriceInBasketItems(int productId, decimal newPrice,

CustomerBasket basket)

{

var itemsToUpdate = basket?.Items?.Where(x => int.Parse(x.ProductId) ==

productId).ToList();

if (itemsToUpdate != null)

{

foreach (var item in itemsToUpdate)

{

if(item.UnitPrice != newPrice)

{

var originalPrice = item.UnitPrice;

item.UnitPrice = newPrice;

item.OldUnitPrice = originalPrice;

}

}

await _repository.UpdateBasket(basket);

}

}

}

}

کنترل کننده رویداد باید بررسی کند که آیا محصول در هر یک از نمونه های سبد وجود دارد یا خیر. همچنین قیمت اقلام را برای هر خط سبد مرتبط به‌روزرسانی می‌کند.

عدم توانایی در رویدادهای پیام به روز رسانی

یکی از جنبه های مهم رویدادهای پیام به روز رسانی این است که یک شکست در هر نقطه از ارتباط باید باعث شود که پیام دوباره امتحان شود. در غیر این صورت، یک کار پس‌زمینه ممکن است سعی کند رویدادی را منتشر کند که قبلاً منتشر شده است و شرایط مسابقه ایجاد می‌کند. اطمینان حاصل کنید که به‌روزرسانی‌ها یا فاقد قدرت هستند یا اطلاعات کافی را ارائه می‌دهند تا اطمینان حاصل شود که می‌توانید یک نسخه تکراری را شناسایی کنید، آن را دور بیندازید و تنها یک پاسخ را ارسال کنید.

همانطور که قبلا ذکر شد، ناتوانی به این معنی است که یک عمل می تواند چندین بار بدون تغییر نتیجه انجام شود. در یک محیط پیام‌رسانی، مانند هنگام برقراری ارتباط رویدادها، یک رویداد در صورتی بی‌توان است که بتواند چندین بار بدون تغییر نتیجه برای میکروسرویس گیرنده تحویل داده شود. این ممکن است به دلیل ماهیت خود رویداد یا به دلیل نحوه مدیریت رویداد توسط سیستم ضروری باشد. ناتوانی پیام در هر برنامه‌ای که از پیام‌رسانی استفاده می‌کند مهم است، نه فقط در برنامه‌هایی که الگوی اتوبوس رویداد را پیاده‌سازی می‌کنند.

نمونه ای از عملیات idempotent یک دستور SQL است که داده ها را تنها در صورتی وارد جدول می کند که آن داده ها قبلاً در جدول نباشد. مهم نیست که چند بار آن دستور SQL را اجرا کنید. نتیجه یکسان خواهد بود - جدول حاوی آن داده ها خواهد بود. اگر پیام‌ها به طور بالقوه می‌توانستند ارسال شوند و بنابراین بیش از یک بار پردازش شوند، چنین بی‌توانایی می‌تواند هنگام برخورد با پیام‌ها نیز ضروری باشد. به عنوان مثال، اگر منطق سعی مجدد باعث شود فرستنده دقیقاً همان پیام را بیش از یک بار ارسال کند، باید مطمئن شوید که آن پیام ضعیف است.

مثال دیگر ممکن است یک رویداد تکمیل شده با سفارش باشد که به چندین مشترک منتشر می شود. برنامه باید مطمئن شود که اطلاعات سفارش در سیستم های دیگر فقط یک بار به روز می شود، حتی اگر رویدادهای پیام تکراری برای همان رویداد تکمیل شده سفارش وجود داشته باشد.

مثال دیگر ممکن است یک رویداد تکمیل شده با سفارش باشد که به چندین مشترک منتشر می شود. برنامه باید مطمئن شود که اطلاعات سفارش در سیستم های دیگر فقط یک بار به روز می شود، حتی اگر رویدادهای پیام تکراری برای همان رویداد تکمیل شده سفارش وجود داشته باشد.

داشتن نوعی هویت در هر رویداد راحت است تا بتوانید منطقی ایجاد کنید که هر رویداد را تنها یک بار در هر گیرنده پردازش کند.

برخی از پردازش پیام‌ها ذاتاً بی‌توان هستند. برای مثال، اگر سیستمی تصاویر کوچک تصویر را تولید کند، ممکن است مهم نباشد که پیام مربوط به تصویر کوچک تولید شده چند بار پردازش می شود. نتیجه این است که ریز عکسها تولید می شوند و هر بار یکسان هستند. از سوی دیگر، عملیاتی مانند فراخوانی درگاه پرداخت برای شارژ کارت اعتباری ممکن است اصلاً بی تأثیر نباشد. در این موارد، باید اطمینان حاصل کنید که پردازش چندباره یک پیام، تأثیری را که انتظار دارید داشته باشد.

کپی برداری از پیام های رویداد یکپارچه سازی

می توانید مطمئن شوید که رویدادهای پیام تنها یک بار برای هر مشترک در سطوح مختلف ارسال و پردازش می شوند. یکی از راه‌ها استفاده از ویژگی کپی‌برداری است که توسط زیرساخت پیام‌رسانی که استفاده می‌کنید ارائه می‌شود. دیگری این است که منطق سفارشی را در میکروسرویس مقصد خود پیاده سازی کنید. داشتن اعتبارسنجی هم در سطح حمل و نقل و هم در سطح برنامه بهترین گزینه است.

کپی کردن رویدادهای پیام در سطح EventHandler

یک راه برای اطمینان از اینکه یک رویداد فقط یک بار توسط هر گیرنده پردازش می شود، اجرای منطق خاصی در هنگام پردازش رویدادهای پیام در کنترل کننده رویداد است. برای مثال، این رویکردی است که در برنامه eShopOnContainers استفاده می‌شود، همانطور که می‌توانید در کد منبع کلاس UserCheckoutAcceptedIntegrationEventHandler هنگام دریافت رویداد ادغام UserCheckoutAcceptedIntegrationEvent مشاهده کنید. (در این مورد، CreateOrderCommand با یک IdentifiedCommand، با استفاده از eventMsg.RequestId به عنوان شناسه، قبل از ارسال آن به کنترل کننده فرمان پیچیده می شود).

کپی کردن پیام ها هنگام استفاده از RabbitMQ

هنگامی که خرابی های متناوب شبکه رخ می دهد، پیام ها می توانند تکرار شوند و گیرنده پیام باید آماده رسیدگی به این پیام های تکراری باشد. در صورت امکان، گیرنده‌ها باید پیام‌ها را به شیوه‌ای بی‌توان مدیریت کنند، که بهتر از مدیریت صریح آن‌ها با کپی‌برداری است.

با توجه به مستندات RabbitMQ، "اگر پیامی به یک مصرف کننده تحویل داده شود و سپس درخواست داده شود (به عنوان مثال، به دلیل اینکه قبل از قطع اتصال مصرف کننده تایید نشده است) پس RabbitMQ پرچم تحویل مجدد را روی آن تنظیم می کند که دوباره تحویل داده شود (خواه به مصرف کننده مشابه یا مصرف کننده متفاوت).

اگر پرچم "تحویل مجدد" تنظیم شده باشد، گیرنده باید آن را در نظر بگیرد، زیرا ممکن است پیام قبلا پردازش شده باشد. اما این تضمین شده نیست. ممکن است پیام پس از خروج از واسطه پیام هرگز به گیرنده نرسد، شاید به دلیل مشکلات شبکه. از سوی دیگر، اگر پرچم "تحویل مجدد" تنظیم نشود، تضمین می شود که پیام بیش از یک بار ارسال نشده است. بنابراین، گیرنده نیاز دارد پیام‌ها را کپی کند یا پیام‌ها را به روشی غیرقابل پردازش پردازش کند، تنها در صورتی که پرچم «تحویل مجدد» در پیام تنظیم شده باشد.