RabbitMQ در دات‌نت: retry، پیام تکراری و Outbox

مهدی مولوی ۱۴ مهر ۱۴۰۵ ۱۴ دقیقه مطالعه

RabbitMQ با ack دستی و تأیید ناشر تحویل «دست‌کم یک‌باره» را ممکن می‌کند،‏ اما تحویل دقیقاً یک‌باره را تضمین نمی‌کند. برای همین پردازش قابل‌اعتماد پیام در دات‌نت چند لایه دارد: صف durable و پیام persistent،‏ تأیید ناشر (publisher confirms)،‏ ack دستی با prefetch محدود،‏ مصرف‌کننده‌ای که پیام تکراری را بی‌خطر کنار می‌گذارد،‏ retry با تأخیر از راه dead-letter exchange،‏ صف پارکینگ برای پیام‌های سمی و الگوی Outbox برای وقتی که باید هم در پایگاه‌داده بنویسید و هم پیام منتشر کنید. نمونه‌کدها با API ناهمگام RabbitMQ.Client 7 نوشته شده‌اند.

پیام کجا ممکن است گم شود؟

پیام در سه جا در خطر است: میان ناشر و broker،‏ داخل خود broker و میان broker و مصرف‌کننده. اول خود broker: صف باید durable باشد و پیام persistent. این دو از هم مستقل‌اند و پیام transient،‏ حتی در صف durable،‏ هنگام راه‌اندازی دوبارهٔ گره دور ریخته می‌شود. برای persistent کردن پیام در RabbitMQ.Client 7 کافی است در BasicProperties مقدار Persistent را true بگذارید. quorum queue همیشه durable است،‏ روی چند گره تکثیر می‌شود و همهٔ پیام‌هایش را،‏ فارغ از delivery mode،‏ ماندگار نگه می‌دارد؛‏ نمونه‌های این راهنما از همین نوع صف استفاده می‌کنند.

سمت ناشر،‏ اگر تأیید ناشر روشن نباشد،‏ تمام شدن BasicPublishAsync فقط یعنی کلاینت پیام را فرستاده است،‏ نه این‌که broker آن را پذیرفته باشد. در RabbitMQ.Client 7 این تأیید را هنگام ساختن کانال با CreateChannelOptions روشن می‌کنید. اگر publisherConfirmationTrackingEnabled هم فعال باشد،‏ await روی BasicPublishAsync تا رسیدن تأیید broker صبر می‌کند. اگر broker پیام را nack کند،‏ یا پیامی که با mandatory: true منتشر شده به هیچ صفی نرسد و برگردد،‏ استثنای PublishException پرتاب می‌شود. برای پیام persistent در صف durable،‏ تأیید یعنی پیام روی دیسک نوشته شده و در quorum queue یعنی اکثریت replicaها آن را پذیرفته‌اند. مقدار پیش‌فرض mandatory برابر false است و پیامی که به هیچ صفی نرسد،‏ اگر alternate exchange هم تعریف نشده باشد،‏ بی‌صدا دور ریخته می‌شود.

اگر اتصال پیش از رسیدن تأیید قطع شود،‏ ناشر نمی‌داند پیام رسیده یا نه و تنها راه امن،‏ انتشار دوباره است. این اولین منبع پیام تکراری است.

ack دستی،‏ autoAck و prefetch

در حالت autoAck،‏ broker پیام را همان لحظهٔ ارسال تحویل‌شده حساب می‌کند و اگر پروسه در میانهٔ پردازش از کار بیفتد یا اتصال قطع شود،‏ پیام از دست رفته است. برای پیام‌های مهم،‏ BasicConsumeAsync را با autoAck: false صدا بزنید و BasicAckAsync را فقط بعد از ثبت نتیجه در پایگاه‌داده بفرستید. هر تحویلی که ack نشده باشد،‏ با بسته شدن کانال یا اتصال خودبه‌خود به صف برمی‌گردد.

BasicQosAsync تعیین می‌کند چند پیام ack‌نشده هم‌زمان در اختیار مصرف‌کننده باشد و RabbitMQ با global: false این سقف را برای هر مصرف‌کننده جداگانه اعمال می‌کند. مقدار صفر یعنی بی‌سقف،‏ و مستندات هشدار می‌دهد که مصرف بدون ack و بدون سقف،‏ حافظهٔ گرهٔ broker را بالا می‌برد. طبق مستندات،‏ بازهٔ ۱۰۰ تا ۳۰۰ معمولاً بهترین توان عملیاتی را می‌دهد و مقدار ۱ محافظه‌کارانه‌ترین انتخاب است که توان عملیاتی را به‌طور محسوسی کم می‌کند؛‏ عدد مناسب را با اندازه‌گیری روی بار واقعی پیدا کنید.

دو پیش‌فرض دیگر هم مهم‌اند. مهلت ack به‌طور پیش‌فرض ۳۰ دقیقه است. اگر تحویلی در این مدت ack نشود،‏ بسته به نسخهٔ broker و نوع صف،‏ کانال با خطای PRECONDITION_FAILED بسته می‌شود یا (در quorum queue از نسخهٔ 4.3) مصرف‌کننده لغو می‌شود؛‏ در هر دو حالت تحویل‌های ack‌نشده به صف برمی‌گردند. ConsumerDispatchConcurrency هم در RabbitMQ.Client 7 به‌طور پیش‌فرض ۱ است،‏ یعنی handlerهای یک کانال یکی‌یکی اجرا می‌شوند. با مقدار بیشتر،‏ پردازش موازی می‌شود،‏ اما ترتیب پردازش پیام‌ها دیگر تضمین نمی‌شود و handler باید thread-safe باشد.

چرا پیام تکراری می‌رسد؟

با ack دستی،‏ تحویل دست‌کم یک‌باره است و پیام تکراری را باید جزو رفتار عادی سیستم حساب کرد. سه سناریوی رایج:

  • مصرف‌کننده تراکنش پایگاه‌داده را commit کرده،‏ اما پیش از رسیدن ack به broker از کار افتاده است.
  • ناشر تأیید را دریافت نکرده و پیام را دوباره منتشر کرده،‏ در حالی که نسخهٔ اول به صف رسیده بود.
  • relay در الگوی Outbox پیام را فرستاده،‏ اما پیش از علامت زدن ردیف به‌عنوان ارسال‌شده متوقف شده است.

به پرچم Redelivered نمی‌شود تکیه کرد. مستندات RabbitMQ آن را فقط نشانه‌ای می‌داند که شاید پیام قبلاً دیده شده باشد،‏ و در سناریوی دوم اصلاً روشن نمی‌شود،‏ چون نسخهٔ دوم برای broker پیامی تازه است. توصیهٔ خود مستندات این است که مصرف‌کننده را idempotent طراحی کنید.

راه اول این است که خود عملیات idempotent باشد: «وضعیت سفارش ۴۲ را پرداخت‌شده کن» را می‌شود چند بار اجرا کرد،‏ اما «۱۰۰ واحد به موجودی اضافه کن» را نه. راه دوم،‏ برای عملیاتی که ذاتاً idempotent نیستند،‏ جدولی از پیام‌های پردازش‌شده است که MessageId کلید اصلی آن است و در همان تراکنشی پر می‌شود که تغییر اصلی را ثبت می‌کند. شرطش این است که تولیدکننده MessageId را هنگام ساختن پیام تعیین کند و در انتشار دوباره عوضش نکند؛‏ اگر هر انتشار یک Guid تازه بسازد،‏ این جدول بی‌فایده می‌شود.

public sealed class OrderPlacedHandler(BillingDbContext db)
{
    public async Task HandleAsync(string messageId, OrderPlaced order, CancellationToken ct)
    {
        if (await db.ProcessedMessages.AnyAsync(p => p.MessageId == messageId, ct))
            return;

        var invoice = Invoice.CreateFor(order);
        db.Invoices.Add(invoice);
        db.ProcessedMessages.Add(new ProcessedMessage(messageId, DateTime.UtcNow));
        db.OutboxMessages.Add(OutboxMessage.Create("invoice.created", new InvoiceCreated(invoice.Id)));

        try
        {
            await db.SaveChangesAsync(ct);
        }
        catch (DbUpdateException ex) when (ex.InnerException is SqlException { Number: 2627 or 2601 })
        {
            // Same MessageId already saved by a concurrent delivery: treat as processed.
        }
    }
}

هر سه درج در یک فراخوانی SaveChangesAsync انجام می‌شود و EF Core به‌طور پیش‌فرض همهٔ تغییرات یک SaveChanges را در یک تراکنش اعمال می‌کند؛‏ پس فاکتور،‏ رکورد پیام و ردیف Outbox یا با هم نوشته می‌شوند یا هیچ‌کدام. AnyAsync فقط میان‌بری ارزان است و تضمین اصلی از کلید اصلی می‌آید: اگر دو مصرف‌کننده هم‌زمان یک پیام را پردازش کنند،‏ ذخیرهٔ دومی با خطای ۲۶۲۷ یا ۲۶۰۱ (نقض کلید اصلی یا ایندکس یکتا) شکست می‌خورد. catch خالی عمدی است: تراکنش دوم کامل برگشت خورده،‏ تراکنش اول همین پیام را ثبت کرده و handler بدون خطا برمی‌گردد تا مصرف‌کننده پیام را ack کند. اگر جدول‌های دیگر این تراکنش هم قید یکتا دارند،‏ نام قید را هم بررسی کنید تا خطای واقعی پنهان نماند. برای هر پیام هم DbContext تازه‌ای بسازید،‏ چون بعد از شکست SaveChanges موجودیت‌های افزوده‌شده هنوز در همان DbContext ردگیری می‌شوند.

retry با تأخیر: DLX و صف انتظار

ساده‌ترین واکنش به خطا BasicNackAsync با requeue: true است،‏ اما پیام برگشتی ممکن است فوراً دوباره تحویل شود. اگر علت خطا فوراً برطرف نشود،‏ مثلاً پایگاه‌داده چند دقیقه در دسترس نباشد،‏ حلقه‌ای از تحویل و برگشت ساخته می‌شود که به گفتهٔ مستندات RabbitMQ از نظر پهنای باند شبکه و CPU پرهزینه است. میان تلاش‌ها باید فاصله انداخت.

الگوی کلاسیکی که روی همهٔ نسخه‌های پشتیبانی‌شده کار می‌کند از dead-letter exchange (DLX) و TTL ساخته می‌شود. صف کار پیام‌های ردشده را به DLX خودش می‌سپارد و این exchange به صف انتظاری وصل است که مصرف‌کننده ندارد،‏ TTL ثابت دارد و DLX آن دوباره همان exchange اصلی است. وقتی پردازش شکست می‌خورد،‏ مصرف‌کننده پیام را با requeue: false رد می‌کند؛‏ پیام تا پایان TTL در صف انتظار می‌ماند و بعد به صف کار برمی‌گردد. کد زیر ابتدای متد ExecuteAsync یک BackgroundService است؛‏ نام کاربری و رمز را از پیکربندی بخوانید و در کد ننویسید.

var factory = new ConnectionFactory
{
    HostName = "rabbitmq.internal",
    ClientProvidedName = "billing-worker"
};
await using var connection = await factory.CreateConnectionAsync(stoppingToken);
await using var channel = await connection.CreateChannelAsync(cancellationToken: stoppingToken);

await channel.ExchangeDeclareAsync("orders", ExchangeType.Direct, durable: true);
await channel.ExchangeDeclareAsync("orders.retry", ExchangeType.Fanout, durable: true);

await channel.QueueDeclareAsync("orders.work", durable: true, exclusive: false, autoDelete: false,
    arguments: new Dictionary<string, object?>
    {
        ["x-queue-type"] = "quorum",
        ["x-dead-letter-exchange"] = "orders.retry"
    });
await channel.QueueBindAsync("orders.work", "orders", "order.placed");

await channel.QueueDeclareAsync("orders.wait", durable: true, exclusive: false, autoDelete: false,
    arguments: new Dictionary<string, object?>
    {
        ["x-queue-type"] = "quorum",
        ["x-message-ttl"] = 30_000,
        ["x-dead-letter-exchange"] = "orders"
    });
await channel.QueueBindAsync("orders.wait", "orders.retry", "");

await channel.QueueDeclareAsync("orders.parking", durable: true, exclusive: false, autoDelete: false,
    arguments: new Dictionary<string, object?> { ["x-queue-type"] = "quorum" });

برای صف انتظار x-dead-letter-routing-key تعیین نشده،‏ پس پیام با همان routing key اصلی،‏ یعنی order.placed،‏ به exchange orders و از آن‌جا به orders.work برمی‌گردد. RabbitMQ چرخه‌های dead-letter را تشخیص می‌دهد،‏ اما پیام را فقط وقتی حذف می‌کند که در کل چرخه هیچ ردی (rejection) نبوده باشد؛‏ این‌جا هر دور با رد مصرف‌کننده شروع می‌شود. صف orders.parking نه مصرف‌کنندهٔ خودکار دارد و نه TTL،‏ تا پیام‌ها برای بررسی بمانند. چند نکتهٔ دیگر:

  • TTL را روی صف بگذارید،‏ نه روی تک‌تک پیام‌ها. پیام منقضی فقط وقتی به سر صف برسد dead-letter می‌شود؛‏ با TTL جداگانه برای هر پیام،‏ پیام‌های منقضی ممکن است پشت پیام‌هایی بمانند که هنوز منقضی نشده‌اند. برای تأخیر پله‌ای چند صف انتظار با TTLهای مختلف بسازید و مصرف‌کننده پیام را خودش در صف مناسب منتشر کند.
  • مستندات RabbitMQ برای DLX و TTL به‌جای x-argument،‏ policy را توصیه می‌کند،‏ چون x-argument را بدون استقرار دوبارهٔ برنامه نمی‌شود عوض کرد،‏ اما policy را هر وقت بخواهید. در کد بالا برای خوانایی x-argument گذاشته‌ام؛‏ فقط x-queue-type را باید هنگام تعریف صف داد.
  • dead-lettering به‌طور پیش‌فرض بدون publisher confirms داخلی انجام می‌شود (راهبرد at-most-once) و در کلاستر ممکن است پیام در همین مسیر گم شود. در quorum queue با گذاشتن dead-letter-strategy روی at-least-once و overflow روی reject-publish در policy،‏ این مسیر دست‌کم یک‌باره می‌شود،‏ به قیمت مصرف بیشتر حافظه و CPU.
  • از RabbitMQ 4.3،‏ quorum queue خودش retry با تأخیر خطی دارد: با آرگومان‌های x-delayed-retry-type،‏ x-delayed-retry-min و x-delayed-retry-max (یا کلیدهای متناظر بدون پیشوند x- در policy)،‏ پیامی که مصرف‌کننده به صف برمی‌گرداند پیش از تحویل دوباره نگه داشته می‌شود و تأخیر بر اساس شمار تحویل‌های ناموفق بیشتر می‌شود. اگر نسخهٔ broker اجازه می‌دهد،‏ این قابلیت می‌تواند جای صف انتظار را بگیرد.
  • طبق README مخزن افزونهٔ delayed message exchange،‏ تیم RabbitMQ دیگر آن را نگهداری نمی‌کند؛‏ طراحی تازه را روی آن بنا نکنید.

پیام سمی و صف پارکینگ

پیام سمی هیچ‌وقت با موفقیت پردازش نمی‌شود: قالبش خراب است،‏ به رکوردی ناموجود اشاره می‌کند یا به باگی در handler می‌خورد. retry بی‌پایانش فقط منابع را هدر می‌دهد و مشکل را پنهان می‌کند؛‏ باید برای تلاش‌ها سقف گذاشت و بعد پیام را به صف پارکینگ فرستاد تا کسی بررسی‌اش کند.

quorum queue یک تور ایمنی داخلی هم دارد. از RabbitMQ 4.0 مقدار پیش‌فرض delivery-limit برابر ۲۰ است و تعداد تحویل‌های ناموفق در هدر x-delivery-count می‌آید. پیامی که از این حد بگذرد،‏ اگر DLX تعریف شده باشد dead-letter و در غیر این صورت حذف می‌شود؛‏ یعنی quorum queue بدون DLX پیام سمی را بی‌صدا پاک می‌کند. از نسخهٔ 4.3 این شمارنده فقط با شکست واقعی،‏ مثل basic.reject یا قطع کانال و اتصال،‏ بالا می‌رود و basic.nack را نمی‌شمارد. طبق مستندات،‏ غیرفعال کردن حد با مقدار -1 توصیه نمی‌شود.

در این طراحی شمار تلاش‌ها از هدر x-death می‌آید: هر بار که پیام از orders.work dead-letter می‌شود،‏ RabbitMQ شمارندهٔ همان صف و همان دلیل را در این هدر یکی زیاد می‌کند. مصرف‌کننده پیش از پردازش این عدد را می‌خواند و اگر به سقف رسیده باشد،‏ پیام را بدون اجرا به پارکینگ می‌فرستد. پیامی که قابل خواندن نیست یا MessageId ندارد هم مستقیم به همان‌جا می‌رود. این تکه ادامهٔ همان متد است:

const int MaxAttempts = 5;

await using var parking = await connection.CreateChannelAsync(
    new CreateChannelOptions(publisherConfirmationsEnabled: true, publisherConfirmationTrackingEnabled: true));
await channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 20, global: false);

var consumer = new AsyncEventingBasicConsumer(channel);
consumer.ReceivedAsync += async (_, ea) =>
{
    var id = ea.BasicProperties.MessageId;
    OrderPlaced? order = null;
    try { order = JsonSerializer.Deserialize<OrderPlaced>(ea.Body.Span); }
    catch (JsonException) { }

    if (id is null || order is null || DeathCount(ea.BasicProperties, "orders.work") >= MaxAttempts)
    {
        var props = new BasicProperties(ea.BasicProperties) { Persistent = true };
        await parking.BasicPublishAsync("", "orders.parking", mandatory: true,
            basicProperties: props, body: ea.Body.ToArray());
        await channel.BasicAckAsync(ea.DeliveryTag, multiple: false);
        return;
    }

    try
    {
        await using var scope = scopes.CreateAsyncScope();
        var handler = scope.ServiceProvider.GetRequiredService<OrderPlacedHandler>();
        await handler.HandleAsync(id, order, stoppingToken);
        await channel.BasicAckAsync(ea.DeliveryTag, multiple: false);
    }
    catch (Exception)
    {
        await channel.BasicNackAsync(ea.DeliveryTag, multiple: false, requeue: false);
    }
};

await channel.BasicConsumeAsync("orders.work", autoAck: false, consumer: consumer);
await Task.Delay(Timeout.Infinite, stoppingToken);

static long DeathCount(IReadOnlyBasicProperties props, string queue)
{
    if (props.Headers is null || !props.Headers.TryGetValue("x-death", out var raw)
        || raw is not IList<object> deaths)
        return 0;

    long total = 0;
    foreach (var d in deaths.OfType<IDictionary<string, object>>())
        if (d.TryGetValue("queue", out var q) && q is byte[] name && Encoding.UTF8.GetString(name) == queue)
            total += Convert.ToInt64(d["count"]);
    return total;
}

انتشار در پارکینگ روی کانال جداگانه‌ای با تأیید ناشر انجام می‌شود،‏ پس ack پیام اصلی فقط بعد از تأیید broker فرستاده می‌شود؛‏ اگر پروسه میان این دو از کار بیفتد،‏ پیام دو بار پارک می‌شود که بی‌ضرر است. scopes همان IServiceScopeFactory تزریق‌شده است. DeathCount همهٔ دلیل‌ها را برای صف کار جمع می‌زند،‏ پس پیامی که با رسیدن به delivery-limit به صف انتظار رفته هم شمرده می‌شود. بدنه با ToArray کپی می‌شود،‏ چون طبق مستندات کلاینت،‏ استفاده از حافظهٔ Body بیرون از ReceivedAsync بدون کپی امن نیست.

با MaxAttempts = 5،‏ پیامی که مدام خطا می‌دهد پنج بار پردازش می‌شود و در تحویل ششم بدون اجرا پارک می‌شود. روی طول صف پارکینگ هشدار بگذارید؛‏ پارکینگی که کسی سراغش نرود،‏ در عمل همان حذف بی‌صداست.

Outbox: ذخیره در پایگاه‌داده و انتشار پیام با هم

فرض کنید سرویسی فاکتوری ذخیره می‌کند و باید رویداد InvoiceCreated را هم منتشر کند. SQL Server و RabbitMQ تراکنش مشترکی ندارند. اگر اول ذخیره کنید و پروسه پیش از انتشار از کار بیفتد،‏ فاکتور بدون رویداد می‌ماند؛‏ اگر اول منتشر کنید و ذخیره شکست بخورد،‏ رویدادی دربارهٔ داده‌ای که وجود ندارد پخش شده است.

در الگوی transactional outbox،‏ پیام به‌جای انتشار مستقیم در جدول Outbox و در همان تراکنش تغییر اصلی نوشته می‌شود؛‏ همان کاری که OrderPlacedHandler با OutboxMessages.Add کرد. فرایندی جداگانه،‏ که معمولاً relay نامیده می‌شود،‏ ردیف‌های ارسال‌نشده را می‌خواند،‏ با تأیید ناشر منتشر می‌کند و بعد علامت ارسال می‌زند. پس پیام فقط وقتی منتشر می‌شود که تراکنش commit شده باشد. کد زیر بدنهٔ ExecuteAsync یک BackgroundService جداست:

await using var channel = await connection.CreateChannelAsync(
    new CreateChannelOptions(publisherConfirmationsEnabled: true, publisherConfirmationTrackingEnabled: true),
    stoppingToken);

while (!stoppingToken.IsCancellationRequested)
{
    await using var scope = scopes.CreateAsyncScope();
    var db = scope.ServiceProvider.GetRequiredService<BillingDbContext>();

    var batch = await db.OutboxMessages
        .Where(m => m.SentAt == null)
        .OrderBy(m => m.CreatedAt)
        .Take(100)
        .ToListAsync(stoppingToken);

    foreach (var m in batch)
    {
        var props = new BasicProperties
        {
            MessageId = m.Id.ToString(),
            Type = m.Type,
            ContentType = "application/json",
            Persistent = true
        };
        await channel.BasicPublishAsync("billing", m.Type, mandatory: true,
            basicProperties: props, body: m.Payload, cancellationToken: stoppingToken);
        m.SentAt = DateTime.UtcNow;
        await db.SaveChangesAsync(stoppingToken);
    }

    if (batch.Count == 0)
        await Task.Delay(TimeSpan.FromSeconds(1), stoppingToken);
}

این‌جا connection و scopes از DI می‌آیند. این relay هر پیام را جداگانه منتشر می‌کند و منتظر تأییدش می‌ماند. آموزش رسمی RabbitMQ نشان می‌دهد انتشار دسته‌ای سریع‌تر است،‏ اما وقتی خطایی رخ دهد،‏ فهمیدن این‌که کدام پیام نرسیده سخت‌تر می‌شود. شناسهٔ ردیف Outbox همان MessageId است؛‏ اگر relay بعد از انتشار و پیش از SaveChangesAsync متوقف شود،‏ بعد از راه‌اندازی دوباره همان پیام را باز می‌فرستد و مصرف‌کننده نسخهٔ دوم را با جدول پیام‌های پردازش‌شده کنار می‌گذارد. به همین دلیل Outbox بدون مصرف‌کنندهٔ idempotent کامل نیست. در کد واقعی،‏ PublishException و خطای اتصال را بگیرید،‏ لاگ کنید و بعد از مکثی کوتاه ادامه دهید.

اگر چند نمونه از سرویس اجرا می‌شود،‏ یا فقط یکی از آن‌ها relay را اجرا کند یا ردیف‌ها را هنگام خواندن قفل کنید. روش رایج در SQL Server خواندن با hintهای UPDLOCK و READPAST داخل یک تراکنش است؛‏ مستندات SQL Server هم READPAST را اساساً برای صف کار روی جدول معرفی می‌کند. در EF Core این hintها را باید با SQL خام (FromSql) نوشت،‏ و اگر READ_COMMITTED_SNAPSHOT روشن باشد،‏ READPAST در سطح READ COMMITTED فقط همراه READCOMMITTEDLOCK پذیرفته می‌شود.

یک ایندکس filtered روی ردیف‌های ارسال‌نشده پرس‌وجوی relay را سبک نگه می‌دارد (راهنمای کارایی SQL Server) و ردیف‌های ارسال‌شده را هم باید دوره‌ای پاک یا بایگانی کرد. اگر نمی‌خواهید این قطعه‌ها را خودتان بسازید،‏ MassTransit و NServiceBus هم retry،‏ Outbox و کنار گذاشتن پیام تکراری را آماده دارند؛‏ شرایط مجوز این دو با هم فرق دارد و پیش از انتخاب باید بررسی شود.

چک‌لیست پیش از محیط عملیاتی

در «گزینه دو»،‏ RabbitMQ کنار .NET 6،‏ SQL Server و Redis بخشی از پشتهٔ ما بود. فهرست زیر خلاصهٔ این راهنماست و پیش از راه‌اندازی هر مصرف‌کنندهٔ تازه می‌شود مرورش کرد:

  • صف quorum یا دست‌کم durable،‏ پیام با Persistent = true،‏ تأیید ناشر و mandatory: true برای پیام‌های مهم.
  • autoAck: false،‏ ack فقط بعد از commit و prefetch محدود.
  • پردازشی که پیش از پایان مهلت ack (به‌طور پیش‌فرض ۳۰ دقیقه) تمام شود.
  • MessageId پایدار از سمت تولیدکننده و جدول پیام‌های پردازش‌شده در همان تراکنش.
  • retry با تأخیر،‏ سقف تلاش و صف پارکینگ با هشدار روی طولش.
  • DLX برای هر quorum queue،‏ تا delivery-limit پیامی را بی‌صدا حذف نکند.
  • Outbox هر جا که ذخیره و انتشار باید با هم انجام شوند.

هر سیستمی هم به broker نیاز ندارد. وقتی لازم نیست پیام‌ها بعد از ری‌استارت پروسه باقی بمانند،‏ صف درون‌پروسه‌ای ساده‌تر است؛‏ در خط لولهٔ زمان تماشا از یک Channel محدود در حافظه با پاسخ‌های ۲۰۲ و ۵۰۳ استفاده کردم. اگر برای طراحی یا بازبینی چنین مسیری کمک می‌خواهید،‏ در صفحهٔ همکاری توضیح داده‌ام چه کمکی از من برمی‌آید.

پرسش‌های رایج

آیا RabbitMQ تحویل دقیقاً یک‌باره را تضمین می‌کند؟

خیر. با صف و پیام ماندگار،‏ تأیید ناشر و ack دستی،‏ تحویل دست‌کم یک‌باره به دست می‌آید و پیام تکراری هم ممکن است برسد. اثر دقیقاً یک‌باره را باید در خود برنامه ساخت،‏ مثلاً با جدول MessageIdهای پردازش‌شده که در همان تراکنش تغییر اصلی نوشته می‌شود.

autoAck یا ack دستی؟

در autoAck پیام همان لحظهٔ ارسال تحویل‌شده حساب می‌شود و اگر مصرف‌کننده در میانهٔ پردازش از کار بیفتد،‏ پیام گم می‌شود. برای پیام‌های مهم،‏ ack را دستی و بعد از ثبت نتیجه بفرستید و تعداد پیام‌های ack‌نشده را با BasicQosAsync محدود کنید.

چرا با requeue: true دوباره تلاش نکنیم؟

پیام برگشتی ممکن است فوراً دوباره تحویل شود و تا وقتی علت خطا برطرف نشده،‏ حلقه‌ای پرهزینه از تحویل و برگشت ساخته می‌شود. بهتر است پیام را با requeue: false به صف انتظاری با TTL بفرستید،‏ یا در RabbitMQ 4.3 و بالاتر retry با تأخیر خود quorum queue را فعال کنید که پیام برگشتی را پیش از تحویل دوباره نگه می‌دارد.

delivery-limit در quorum queue چه می‌کند؟

از RabbitMQ 4.0 مقدار پیش‌فرض آن ۲۰ است و پیامی که بیش از این تعداد ناموفق تحویل شود،‏ اگر DLX تعریف شده باشد dead-letter و در غیر این صورت حذف می‌شود. از نسخهٔ 4.3 فقط شکست‌های واقعی،‏ مثل basic.reject یا قطع اتصال،‏ شمرده می‌شوند و basic.nack نه. برای همین مستندات توصیه می‌کند هر quorum queue پیکربندی dead-letter داشته باشد.

الگوی Outbox چه مشکلی را حل می‌کند؟

وقتی باید هم داده را در پایگاه‌داده ذخیره کنید و هم پیامی منتشر کنید،‏ Outbox پیام را در همان تراکنش در یک جدول می‌نویسد و فرایندی جداگانه آن را منتشر می‌کند. پیام فقط در صورت commit منتشر می‌شود،‏ اما ممکن است تکراری برسد؛‏ پس مصرف‌کننده باید idempotent باشد.

در همین بخش

اشتراک‌گذاری
راهنمای بک‌اند