در این مقاله، ما قصد داریم به نحوه استفاده از کتابخانه کاربردی و منبع باز MassTransit ,  در ارتباط با RabbitMQ در یک برنامه ASP.NET Core نگاهی بیندازیم.
ابتدا، ما می‌خواهیم برخی از ویژگی‌های پیشرفته‌تر RabbitMQ و همچنین برخی از مفاهیمی را که در کتابخانه MassTransit با آنها مواجه خواهیم شد، پوشش دهیم. در نهایت، نحوه استفاده از این کتابخانه ها را در یک برنامه ASP.NET Core Web API یاد خواهیم گرفت.
در این مقاله از Docker برای اجرای سرور RabbitMQ ما به لوکال  استفاده می کنیم.
پس از اتمام این مقاله، از شما انتظار داریم که درک خوبی از چیستی MassTransit، مزایای آن و نحوه استفاده از آن در ارتباط با RabbitMQ داشته باشید.

RabbitMQ چیست؟
به طور خلاصه، RabbitMQ یک message brokerاست که پذیرش، ذخیره و ارسال پیام ها را بین برنامه های ما انجام می دهد. استفاده از message broker به ما امکان می دهد تا برنامه های مستقل (decoupl)  بر پایه ارتباطات async بین برنامه های خود ایجاد کنیم.

Exchanges در RabbitMQ چیست؟
هنگام کار با RabbitMQ، فرستنده می تواند پیام هایی را به چند نقطه پایانی مختلف ارسال کنند:
•    -Queues صف ها
•    -Exchanges مبادلات

چون در مقاله قبلی به queues  پرداختیم، الان فقط بر روی exchanges تمرکز می‌کنیم. هنگامی که فرستنده پیام رو مستقیماً به یک صف ارسال می کنه، این پیام توسط همه consumers آن صف دریافت می شه. اما اگر بخواهیم به صورت انتخابی پیام‌ها را بر اساس metadata موجود در پیام به صف‌های مختلف ارسال کنیم، چه؟ اینجاست که exchanges وارد عمل می شوند.

یک exchange پیام‌هایی را از producers دریافت می‌کنه و بسته به پیکربندی آن، پیام را به یک یا چند صف ارسال می‌کنه. ما باید یک اتصال(binding) ایجاد کنیم، که مطمئن شود پیام ها از exchange  به یک یا چند صف ارسال می شه.
ما می تونیم exchanges  را از یکی از انواع زیر تعریف کنیم:

•    Direct مستقیم
•    Topic موضوع
•    Headers سرصفحه ها
•    Fanout

در این مقاله، ما بر روی نوع Fanout تمرکز می کنیم، چون  MassTransit به طور پیش فرض از اون استفاده می کنه. نوع تبادل fanout بسیار ساده است. فقط تمام پیام هایی که دریافت می کنه رو به تمام صف هایی که با آن اتصال ایجاد کرده اند(بایند شده اند) پخش می کنه.
 

MassTransit چیست؟
MassTransit یک فریمورک رایگان، منبع باز و توزیع شده برای برنامه های NET ه. این منطق زیربنایی مورد نیاز برای کار با واسطه‌های پیام مانند RabbitMQ را از بین می‌بره و ایجاد برنامه‌های کاربردی مبتنی بر پیام را آسان‌تر می‌کند.

چند مفهوم بنیادی هست که ابتدا باید بدونیم:
Service Bus که معمولاً به Bus کوتاه می‌شود، اصطلاحی است که به نوع برنامه‌ای گفته می‌شود که حرکت پیام‌ها را مدیریت می‌کند.

Transports  انواع message brokersهایی هستند که MassTransit با آنها کار می کند، از جمله RabbitMQ، InMemory، Azure Service Bus و غیره.

Message  یک قرارداد است که با ایجاد یک کلاس دات نت یا interface تعریف می شوند.

Command نوعی پیام است که به طور خاص برای دستور دادن به یک سرویس برای انجام کاری استفاده می شود. این نوع پیام ها به یک endpoint  (صف) ارسال می شوند و با استفاده از یک دنباله فعل-اسم بیان می شوند.

Events  نوع دیگری از پیام هستند که نشان می دهند چیزی اتفاق افتاده است. Events  برای یک یا چند مصرف کننده منتشر می شوند و با استفاده از توالی اسم-فعل (زمان گذشته) بیان می شوند.

 

چرا از MassTransit استفاده کنیم؟
انتخاب استفاده از کتابخانه ای مانند MassTransit به جای کار با کتابخانه های بومی message broker , چند مزیت دارد. اولاً، با انتزاع منطق زیربنایی message broker ، می‌توانیم با چند message broker کار کنیم، بدون اینکه نیازی به بازنویسی کامل کدمان باشد. این به ما امکان می دهد هنگام دولوپ با InMemory transport کار کنیم، سپس هنگام استقرار کد خود، از transport  دیگری مانند Azure Service Bus یا Amazon Simple Queue Service استفاده کنیم.
علاوه بر این، زمانی که ما با یک معماری مبتنی بر پیام کار می کنیم، الگوهای خاصی وجود دارد که باید از آنها آگاه باشیم و آن ها را پیاده سازی کنیم، مانند تلاش مجدد، قطع کننده مدار، صندوق خروجی. MassTransit همه اینها را به همراه بسیاری از ویژگی های دیگر مانند رسیدگی به استثناء، تراکنش های توزیع شده و نظارت انجام می دهد.
اکنون که متوجه شدیم MassTransit چیست و چرا باید از آن استفاده کنیم، بیایید ببینیم چگونه می‌توانیم همراه با RabbitMQ در ASP.NET Core  ازش استفاده کنیم.

پیاده سازی MassTransit با RabbitMQ در Core ASP.NET
نصب RabbitMQ
قبل از شروع ایجاد برنامه خود، ابتدا باید یک سرور RabbitMQ را با استفاده از Docker ایحاد کنیم. با توجه به اینکه Docker نصب شده است، یک ترمینال خط فرمان باز می کنیم و از دستور docker run برای چرخش سرور خود استفاده می کنیم:

docker run -d --hostname my-rabbitmq-server --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management

ما از ایمیج rabbitmq:3-management از DockerHub استفاده می‌کنیم که یک UI در پورت 15672 در اختیار ما قرار می‌دهد. همچنین باید یک نگاشت پورت برای 5672 اضافه کنیم، که پورت پیش‌فرض RabbitMQ برای ارتباط است. برای اینکه بتوانیم به UI مدیریت دسترسی پیدا کنیم، یک پنجره مرورگر باز می کنیم و با استفاده از ورود پیش فرض guest/guest به localhost:15672 پیمایش می کنیم. بعداً به این رابط کاربری مدیریت باز خواهیم گشت تا ببینیم MassTransit در RabbitMQ چه چیزی برای ما ایجاد می کند.

ایجاد یک کتابخانه کلاس مشترک
اگر به مفاهیم MassTransit برگردید، وقتی از Messages استفاده می کنیم، باید یک کلاس یا interface دات نتی تعریف کنیم. MassTransit شامل فضای نام برای قراردادهای پیام است، بنابراین می‌توانیم از یک class/inteface مشترک برای تنظیم صحیح پیوندهایمان استفاده کنیم.

با در نظر گرفتن این موضوع، اولین کاری که می‌خواهیم انجام دهیم این است که یک class library ایجاد کنیم که حاوی interface  مشترکی باشد که برای برنامه‌های Producer  و Consumer  خود استفاده خواهیم کرد.

ابتدا یک class library ایجاد می کنیم که آن را SharedModels می نامیم و در آن یک interface تعریف می کنیم:

public interface OrderCreated 
{
    int Id { get; set; }
    string ProductName { get; set; }
    decimal Price { get; set; }
    int Quantity { get; set; }
}

ما از زمان گذشته برای نام interface   استفاده می کنیم که نشون می ده این یک نوع event  ه.  این همه آن چیزیه که ما برای کتابخانه کلاس SharedModels خود نیاز داریم. در مرحله بعد، ما Producer خود را پیاده سازی میکنیم.

ایجاد یک تولید کننده(Producer) با استفاده از MassTransit
بیایید Producer خود را ایجاد کنیم که آن را به عنوان یک ASP.NET Core Web API پیاده سازی می کنیم. اولین کاری که می خواهیم انجام دهیم اینه که یک رفرنس به کتابخانه SharedModels خود اضافه کنیم.
در مرحله بعد، باید چند بسته Nuget را برای MassTransit اضافه کنیم:
•    MassTransit
•    MassTransit.AspNetCore
•    MassTransit.RabbitMQ
اکنون اجازه دهید MassTransit را برای استفاده از RabbitMQ در Program.cs پیکربندی کنیم:

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddMassTransit(x =>
{
    x.UsingRabbitMq();
});

builder.Services.AddMassTransitHostedService();

ابتدا WebApplicationBuilder خود را ایجاد می کنیم.
در مرحله بعد MassTransit را برای استفاده از RabbitMQ کانفیگ می کنیم. از آنجایی که این Producer است، و RabbitMQ قراره روی لوکال هاست اجرا بشه، نیازی به تعریف تنظیمات بیشتری در اینجا نداریم. مرحله آخر AddMassTransitHostedService رو کال میکنیم که به طور خودکار شروع/توقف bus را کنترل می کنه.
قبل از ایجاد یک کنترلر API، ابتدا باید یک کلاس OrderDto ایجاد کنیم که به عنوان پارامتری برای متد نقطه پایانی ما استفاده می شه:

public class OrderDto
{
    public string ProductName { get; set; }
    public decimal Price { get; set; }
    public int Quantity { get; set; }
}

بعد از پیکربندی MassTransit برای استفاده از RabbitMQ و تعریف DTO ، بیایید حالا یک API controller ایجاد کنیم که با استفاده از اینترفیس OrderCreated , یک message یا به طور مشخصا یک event رو پابلیش می کنه:

[ApiController]
[Route("api/[controller]")]
public class OrdersController : ControllerBase
{
    private readonly IPublishEndpoint _publishEndpoint;

    public OrdersController(IPublishEndpoint publishEndpoint)
    {
        _publishEndpoint = publishEndpoint;
    }
}

اولین کاری که باید انجام دهیم اینه که IPublishEndpoint را به controller اینجکت کنیم، که برای پابلیش event از آن استفاده کنیم.
الان می توانیم یک endpoint برای ایجاد یک سفارش ایجاد کنیم:

[HttpPost]
public async Task<IActionResult> CreateOrder(OrderDto orderDto)
{
    await _publishEndpoint.Publish<OrderCreated>(new
    {
        Id = 1,
        orderDto.ProductName,
        orderDto.Quantity,
        orderDto.Price
    });

    return Ok();
}

یک anonymous  ایجاد کردیم .فقط چک کنید که نام پراپرتیها از همان نام‌هایی که در اینترفیس  OrderCreated تعریف شده‌اند, استفاده کرده باشید. این تمام چیزی است که ما برای پابلیش یک event در  RabbitMQ نیاز داریم.
در نهایت یک OkResult برمی گردانیم.

 

ایجاد یک Consumer با استفاده از    MassTransit
الان که Producer رو در اختیار داریم، نگاهی به ایجاد یک Consumer بسیار ساده با استفاده از یک برنامه کنسول دات نت بندازیم.
تو همون solution، یک برنامه کنسول به نام Consumer ایجاد میکنیم. مانند برنامه Producer ، باید یک رفرمس پروژه را به کتابخانه SharedModels خود اضافه میکنیم. به همراه Nuget MassTransit:

•    MassTransit
•    MassTransit.RabbitMQ

💡این بار، به پکیج MassTransit.AspNetCore نیاز نداریم، زیرا از dependency injection در این پروژه استفاده نخواهیم کرد.

 

ابتدا باید یک پیاده ساز Consumer ایجاد کنیم، که حاوی منطق کاری است که می خواهیم با هر پیام دریافتی انجام دهیم.

class OrderCreatedConsumer : IConsumer<OrderCreated>
{
    public async Task Consume(ConsumeContext<OrderCreated> context)
    {
        var jsonMessage = JsonConvert.SerializeObject(context.Message);
        Console.WriteLine($"OrderCreated message: {jsonMessage}");
    }
}

ابتدا، کلاسی به نام OrderCreatedConsumer ایجاد می‌کنیم، و مطمئن می‌شویم که اینترفیس IConsumer ارائه شده توسط MassTransit را با استفاده از اینترفیس OrderCreated که در کتابخانه ShareModels خودمون تعریف شده است، پیاده‌سازی می‌کنیم. در روش Consume، می‌توانیم به سادگی message را serialize کنیم در کنسول لاگ کنیم.

حالاکه کلاس Consumer ما تعریف شده است، بیاید Consumer رو برای استفاده از MassTransit در Program.cs پیکربندی کنیم:

var busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
{
    cfg.ReceiveEndpoint("order-created-event", e =>
    {
        e.Consumer<OrderCreatedConsumer>();
    });

})

با استفاده از کلاس استاتیک Bus ارائه شده توسط MassTransit ، که قراره برای استفاده از   RabbitMQ  کانفیگ بشه ، یه IBusControl ایجاد کردیم. سپس ReceiveEndpoint را پیکربندی کردیم، که پیام‌هایی را از صف رویداد order-created-event دریافت می‌کند. در نهایت، ما از OrderCreatedConsumer  برای consume پیام‌های این صف استفاده کردیم.

حالا آخرین کاری که باید انجام دهیم این است که BUS را راه اندازی کنیم:

await busControl.StartAsync(new CancellationToken());

try
{
    Console.WriteLine("Press enter to exit");

    await Task.Run(() => Console.ReadLine());
}
finally
{
    await busControl.StopAsync();
}