MassTransit

مشکل: کد تکراری دور هر پیام

در درس RabbitMQ دیدیم که برای یک پیام ساده چقدر کد لازم است. در فروشگاه اینترنتی ما ده‌ها نوع پیام داریم. برای هر کدام باید این کارها را بنویسیم:

1. ساخت Exchange، صف و Binding.
2. تبدیل شیء به JSON و برعکس.
3. تلاش دوباره وقتی سرور ایمیل یک لحظه جواب نمی‌دهد.
4. فرستادن پیام خراب به صف خطا.
5. ثبت پیام در Outbox تا با دیتابیس هماهنگ باشد.
6. ساختن Scope برای Dependency Injection برای هر پیام.

هر تیم این‌ها را کمی متفاوت می‌نویسد و هر کدام چند باگ کوچک دارد. MassTransit همه این‌ها را یک بار و درست نوشته است.

ایده اصلی: کد تو فقط کسب‌وکار است

1. پیام یک record ساده است. هیچ وابستگی به RabbitMQ ندارد.
2. مصرف‌کننده یک کلاس با یک متد است. رابط IConsumer را پیاده می‌کند.
3. انتقال‌دهنده قابل تعویض است. امروز RabbitMQ، فردا Azure Service Bus. کد مصرف‌کننده تغییر نمی‌کند. برای تست هم یک انتقال‌دهنده داخل حافظه هست.

پیام و مصرف‌کننده

پیام‌ها را در یک پروژه مشترک (مثلاً Shop.Contracts) نگه دار. MassTransit نوع پیام را با namespace و نام کلاس می‌شناسد. پس namespace باید در فرستنده و گیرنده یکی باشد.

namespace Shop.Contracts;

public sealed record OrderPlaced(Guid OrderId, string CustomerEmail, decimal Total);

مصرف‌کننده سرویس ایمیل:

using MassTransit;
using Shop.Contracts;

public sealed class SendReceiptConsumer(IEmailSender email) : IConsumer<OrderPlaced>
{
    public async Task Consume(ConsumeContext<OrderPlaced> context)
    {
        var order = context.Message;
        // Idempotent: the sender skips orders that already got a receipt.
        await email.SendReceiptOnceAsync(order.OrderId, order.CustomerEmail, context.CancellationToken);
    }
}

اگر متد Consume بدون خطا تمام شود، MassTransit خودش پیام را Ack می‌کند. اگر خطا بدهد، مسیر تلاش دوباره شروع می‌شود.

راه‌اندازی

using MassTransit;

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddMassTransit(x =>
{
    x.SetKebabCaseEndpointNameFormatter();   // queue names like "send-receipt"
    x.AddConsumer<SendReceiptConsumer>();

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("localhost", "/", h =>
        {
            h.Username("guest");
            h.Password("guest");
        });

        cfg.UseMessageRetry(r =>
        {
            r.Intervals(TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(5)); // two retries
            r.Ignore<ArgumentException>();   // a bad message will not get better
        });

        cfg.ConfigureEndpoints(context);     // one queue per consumer, bound to its message types
    });
});

var app = builder.Build();
app.Run();

متد ConfigureEndpoints کارهای زیادی را انجام می‌دهد:

1. برای هر نوع پیام یک Exchange می‌سازد. اسمش از namespace و نام کلاس ساخته می‌شود.
2. برای هر مصرف‌کننده یک صف می‌سازد. اسم صف از نام کلاس مصرف‌کننده ساخته می‌شود.
3. صف را به Exchange پیام‌هایش وصل می‌کند. پس هر سرویس جدیدی که OrderPlaced را مصرف کند، خودکار نسخه خودش را می‌گیرد.

انتشار پیام: Publish یا Send؟

• متد Publish برای رویداد. «سفارش ثبت شد». فرستنده نمی‌داند چه کسی گوش می‌دهد. صفر، یک یا چند مصرف‌کننده می‌گیرند.
• متد Send برای دستور. «این ایمیل را بفرست». یک مقصد مشخص دارد و فقط یک مصرف‌کننده آن را انجام می‌دهد.

app.MapPost("/orders", async (
    PlaceOrder cmd, ShopDbContext db, IPublishEndpoint publisher, CancellationToken ct) =>
{
    var order = Order.Place(cmd.CustomerEmail, cmd.Items);
    db.Orders.Add(order);

    await publisher.Publish(new OrderPlaced(order.Id, order.CustomerEmail, order.Total), ct);
    await db.SaveChangesAsync(ct); // order and message: one transaction

    return Results.Created($"/orders/{order.Id}", order.Id);
});

تلاش دوباره و صف خطا

1. مصرف‌کننده خطا داد. MassTransit پیام را طبق تنظیم چند بار دیگر به همان مصرف‌کننده می‌دهد.
2. خطای موقت معمولاً حل می‌شود. مثلاً سرور ایمیل بعد از چند ثانیه جواب می‌دهد.
3. اگر همه تلاش‌ها شکست خورد، پیام به صفی با پسوند error می‌رود. جزئیات خطا هم در Header های پیام ذخیره می‌شود.
4. پیام گم نمی‌شود. یک نفر صف خطا را بررسی می‌کند. بعد از رفع مشکل، پیام را به صف اصلی برمی‌گرداند.

پیام و دیتابیس با هم: Outbox

مشکل Dual Write را در درس کافکا دیدیم: سفارش در دیتابیس ذخیره شد، ولی پیام هیچ وقت نرفت. MassTransit یک Outbox آماده با EF Core دارد:

builder.Services.AddMassTransit(x =>
{
    x.AddEntityFrameworkOutbox<ShopDbContext>(o =>
    {
        o.UseSqlServer();
        o.UseBusOutbox();   // Publish writes to the outbox table, not to the broker
    });

    // ... consumers and UsingRabbitMq as before ...
});

public sealed class ShopDbContext(DbContextOptions<ShopDbContext> options) : DbContext(options)
{
    public DbSet<Order> Orders => Set<Order>();

    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        modelBuilder.AddInboxStateEntity();
        modelBuilder.AddOutboxMessageEntity();
        modelBuilder.AddOutboxStateEntity();
    }
}

حالا قدم‌ها این شکلی است:

1. متد Publish پیام را فقط در جدول Outbox می‌گذارد. هنوز چیزی به RabbitMQ نرفته است.
2. متد SaveChangesAsync سفارش و پیام را در یک تراکنش ذخیره می‌کند. یا هر دو، یا هیچ کدام.
3. یک سرویس پس‌زمینه پیام‌های جدول را به RabbitMQ می‌فرستد. اگر RabbitMQ قطع باشد، بعداً می‌فرستد.

همین بسته یک Inbox هم برای سمت مصرف‌کننده دارد. این Inbox شناسه پیام‌های دیده‌شده را نگه می‌دارد و پیام تکراری را کنار می‌گذارد. باید آن را جدا برای صف مصرف‌کننده فعال کنی، با همان جدول‌هایی که بالا ساختیم.

Saga State Machine

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

اشتباه‌های رایج

چه وقت MassTransit؟