दबाव के तहत: कैसे कतार प्रणाली C में उदाहरणों के साथ बैकप्रेस को संभालें (हिन्दी (Hindi))

दबाव के तहत: कैसे कतार प्रणाली C में उदाहरणों के साथ बैकप्रेस को संभालें

Sunday, 23 November 2025

//

17 minute read

Comments Raw Edit
Azure Service Bus C# Distributed Systems Kafka Messaging RabbitMQ

पीछला पेज होम पेज अगला पेज पीछली-पीछे दबाव वितरित प्रणालियों का असंगूढ़ नायक है।

यह बात है इस लेख में तकनीकों के लिए आभासी रूप से लागू हर संदेश कतार और सेवा बस

एक अभिवचन अधिकतर वरिष्ठ विकासकर्ता भी सही बैकप्रेसर प्रबंधन को लागू नहीं करते हैं। वे प्रसन्नतापूर्वक पथ प्रणालियां बनाते हैं जो डेव और स्टेजिंग में अच्छी तरह काम करती हैं।

बैकप्रिसर क्या है

इसके मूल में-,-backpressure एक फीडबैक लूप है जो उत्पादकों को धीमी करता है जब उपभोक्ता पीछे रह जाते हैं।

बिना बैकप्रैसर्स के, एक तेजी से उत्पादक एक धीमी उपभोक्ता को छीन लेगा।

flowchart LR
    P[Producer] --> Q[Queue]
    Q --> C[Consumer]
    C -. "Slow down!" .-> P

    style P stroke:#f59e0b,stroke-width:2px
    style Q stroke:#0ea5e9,stroke-width:2px
    style C stroke:#10b981,stroke-width:2px

पीछला पेज होम पेज अगला पेज दबाव की सुंदरता यह है कि यह वार्तालाप उत्पादक और उपभोक्ता के बीच।

कैसे रब्बीएमक्यू बैकप्रेशर संभालता है

खरगोशMQ एक open-source संदेश ब्रोकर है जो उन्नत संदेश कतार प्रोटोकॉलयह सबसे व्यापक रूप से प्रयुक्त संदेश ब्रोकरों में से एक है. इसे स्टार्टअप से लेकर Bloomberg जैसे उद्यमों तक कंपनियों द्वारा उपयोग किया जाता है.

क्या RabbitMQ में श्रेष्ठता है

विकल्पों पर कब विचार करें यदि आपको अधिक सरल रूटिंग के साथ बड़े पैमाने पर स्प्रेडपुट की जरूरत है तो (millions of messages/secondM SK2KafkaMSC3s logMST4based architecture may be more appropriateMSSK5 If youMS're allMSL7in a cloud providerMSP8 their managed services |(Azure Service Bus | , | AWS SQS | ) | reduce operational overhead |.

RabbitMQ के पास बैकप्रेस को संभालने के लिए कई built-in mechanisms हैं, और उन्हें समझना महत्वपूर्ण है अगर आप building systems that need to stay upright under load रब्बिटएमक्यू दस्तावेज़ीकरण बहुत अच्छा है

प्रवाह नियंत्रण

जब RabbitMQ' का स्मृति उपयोग या कतार गहराई कॉन्फ़िगर किए गए थ्रेसहोल्डों से अधिक है तो यह सक्रिय करता है प्रवाह नियंत्रण. यह अस्थायी रूप से प्रकाशक कनेक्शनों को अवरोधित करता है स्मृति अलार्म और डिस्क अलार्म जो प्रवाह नियंत्रण ट्रिगर करता है

flowchart TD
    subgraph "RabbitMQ Flow Control"
        A[Publisher Sends Message] --> B{Memory/Queue<br/>Threshold OK?}
        B -->|Yes| C[Message Accepted]
        C --> D[Add to Queue]
        B -->|No| E[Connection Blocked]
        E --> F[Publisher Waits]
        F --> G{Threshold<br/>Cleared?}
        G -->|No| F
        G -->|Yes| H[Connection Unblocked]
        H --> A
    end

    style E stroke:#ef4444,stroke-width:3px
    style H stroke:#10b981,stroke-width:2px

यहाँ मुख्य अंतर्दृष्टि यह है कि RabbitMQ न केवल दबाव के तहत संदेश छोड़ता है, जब नहीं।

उपभोक्ता अभिस्वीकृति

उपभोक्ताओं द्वारा गति नियंत्रित करता है मानदंड ACKs और NACKs

आप भी उपयोग कर सकते हैं प्रीफ़ैच सीमाएँ नियंत्रण करने के लिए कितने अज्ञात संदेश एक उपभोक्ता एक बार उड़ान में हो सकता है

// Set prefetch count to limit unacknowledged messages
channel.BasicQos(prefetchSize: 0, prefetchCount: 10, global: false);

यह Rabbit को बताता हैMQ: "मुझे केवल एक ही समय पर संदेश भेजें। .NET क्लाइंट मिसिलीकरण API को विस्तृत रूप से शामिल करता है

कैसे अन्य प्रणालियां बैकप्रेसर को संभालें

पैटर्न वैश्विक हैं, लेकिन क्रियान्वयन भिन्न है

काफका:consumer-controlled polling

अपाचे काफका एक वितरित घटना स्ट्रीमिंग प्लेटफार्म है जो उच्च के लिए डिजाइन किया गया है।

काफ्का एक मूलतः भिन्न दृष्टिकोण लेता है खींचना संदेशों को दबाने के बजाय उन्हें दबाना। काफ़्का उपभोक्ता दस्तावेज़ मतदान मॉडल के बारे में जानकारी के लिए

// Kafka consumer with explicit backpressure control
using var consumer = new ConsumerBuilder<string, string>(config).Build();
consumer.Subscribe("orders");

while (!cancellationToken.IsCancellationRequested)
{
    // Only fetch what you can handle - this IS your backpressure
    var result = consumer.Consume(timeout: TimeSpan.FromSeconds(1));

    if (result != null)
    {
        await ProcessMessageAsync(result.Message.Value);

        // Manual commit = explicit acknowledgement
        consumer.Commit(result);
    }

    // If processing is slow, you simply poll less frequently
    // Kafka doesn't push more messages at you
}

चुने हुए बिट: काफका' उपभोक्ता समूहों को स्वतः विभाजनों के संतुलन में पुनर्विभाजित करता है। . यदि एक उपभोक्ता पीछे रह जाता है तो आप समूह में अधिक उपभोक्ता जोड़ सकते हैं और विभाजन पुनर्वितरित किए जाते हैं।

// Control batch size to manage memory pressure
var config = new ConsumerConfig
{
    BootstrapServers = "localhost:9092",
    GroupId = "order-processors",
    AutoOffsetReset = AutoOffsetReset.Earliest,
    MaxPollIntervalMs = 300000,      // 5 mins max between polls
    MaxPartitionFetchBytes = 1048576, // 1MB max per partition fetch
    FetchMaxBytes = 52428800          // 50MB max total fetch
};

Azure सेवा बस: समवर्ती संदेश नियंत्रण

Azure सेवा बस यह क्लाउड में अनुप्रयोगों और सेवाओं को अलग करने के लिए डिजाइन किया गया है

Azure सेवा बस का उपयोग एक MaxConcurrentCalls यह सेट करना कि' सुंदर रूप से सरल है—यह नियंत्रण करता है कि आपके प्रोसेसर कितनी संदेशों को एक साथ लेता है संदेश प्रक्रमण दस्तावेज़ विस्तृत विवरण में निष्पादन अनुकुलन शामिल करता है

var processor = client.CreateProcessor("orders-queue", new ServiceBusProcessorOptions
{
    // This IS your backpressure - only process 10 at a time
    MaxConcurrentCalls = 10,
    AutoCompleteMessages = false,
    PrefetchCount = 20  // Buffer 20 messages locally
});

processor.ProcessMessageAsync += async args =>
{
    try
    {
        await ProcessOrderAsync(args.Message.Body.ToString());
        await args.CompleteMessageAsync(args.Message);
    }
    catch (Exception ex)
    {
        // Abandon returns message to queue for retry
        await args.AbandonMessageAsync(args.Message);
    }
};

processor.ProcessErrorAsync += args =>
{
    Console.WriteLine($"Error: {args.Exception.Message}");
    return Task.CompletedTask;
};

await processor.StartProcessingAsync();

Azure सेवा बस भी समर्थन करता है सत्र आदेशित प्रसंस्करण और मरे हुए पत्र कतार उन संदेशों के लिए जो बार-बार असफल होते हैं

AWS SQS: दृश्यता समय समाप्ति नृत्य

अमाजोन सरल लाम सेवा (SQS) AWS से पूरी तरह से प्रबंधित संदेश कतार सेवा है।

एसक्यूएस उपयोग करता है दृश्यता समय समाप्ति जब आप एक संदेश प्राप्त करते हैं तो यह अन्य उपभोक्ताओं के लिए अदृश्य हो जाता है

var sqsClient = new AmazonSQSClient();

// Receive with explicit backpressure control
var response = await sqsClient.ReceiveMessageAsync(new ReceiveMessageRequest
{
    QueueUrl = queueUrl,
    MaxNumberOfMessages = 10,           // Batch size = backpressure control
    WaitTimeSeconds = 20,               // Long polling
    VisibilityTimeout = 300             // 5 mins to process before retry
});

foreach (var message in response.Messages)
{
    try
    {
        await ProcessAsync(message.Body);

        // Only delete after successful processing
        await sqsClient.DeleteMessageAsync(queueUrl, message.ReceiptHandle);
    }
    catch
    {
        // Don't delete - message will become visible again after timeout
        // Optionally, change visibility timeout to retry sooner
        await sqsClient.ChangeMessageVisibilityAsync(queueUrl,
            message.ReceiptHandle, visibilityTimeout: 0);
    }
}

चुने हुए एसक्यूएस ट्रिक का उपयोग ApproximateNumberOfMessages कतार गहराई और स्वतः मॉनीटर करने के लिए

var attributes = await sqsClient.GetQueueAttributesAsync(new GetQueueAttributesRequest
{
    QueueUrl = queueUrl,
    AttributeNames = new List<string> { "ApproximateNumberOfMessages" }
});

var depth = int.Parse(attributes.Attributes["ApproximateNumberOfMessages"]);

if (depth > 1000)
{
    // Signal to scale up consumers
    await TriggerAutoScalingAsync();
}

नाट्स जेटस्ट्रीम

NATS एक हल्का भार है जेटस्ट्रीम NATS है। ' का निर्माण किया गया है।- एक स्थायी परत जो स्थायी धाराओं को जोड़ता है।स्पष्ट प्रवाह नियंत्रण.

एनएटीएस जेटस्ट्रीम में उपभोक्ता स्वीकृति और अधिकतम लंबित संदेश सीमाओं के साथ स्पष्ट प्रवाह नियंत्रण है

var js = connection.CreateJetStreamContext();

var subscription = js.PushSubscribeAsync("orders.>", (sender, args) =>
{
    try
    {
        ProcessMessage(args.Message.Data);
        args.Message.Ack();
    }
    catch
    {
        args.Message.Nak();  // Negative ack - redeliver
    }
}, new PushSubscribeOptions.Builder()
    .WithConfiguration(new ConsumerConfiguration.Builder()
        .WithMaxAckPending(100)     // Max unacked messages - THIS is backpressure
        .WithAckWait(30000)         // 30 seconds to ack
        .Build())
    .Build());

सामान्य थ्रेड

ध्यान दें कि इन सभी प्रणालियों में क्या समानता है

  1. स्पष्ट बैच - आप नियंत्रित कर सकते हैं जितनी आप को संभालने के लिए तैयार है
  2. स्वीकृति - संदेश तब तक उपलब्ध रहेंगे जब तक आप प्रक्रिया की पुष्टि न करें
  3. समय समाप्ति---आधारित पुनर्प्राप्ति - यदि आप विफल रहे हैं, संदेश पुनः प्रयास के लिए लौटाएँ
  4. गहराई निगरानी आप हमेशा पूछ सकते हैं

वाक्य संरचना भिन्न है, लेकिन नृत्य एक ही है मुझे बताओ जब मैंने इसे संभाला है।

C# कोड उदाहरण

दायाँ>,>let>'>s को कोड में ले जाएँ।

कतार गहराई मॉनीटरिंग

पहले चीजें First things first—you can't manage what you can 't measureM SK3 HereMSC4s how to check how many messages are waiting in a कतार:

var queue = channel.QueueDeclare(
    queue: "tasks",
    durable: true,
    exclusive: false,
    autoDelete: false);

Console.WriteLine($"Messages ready: {queue.MessageCount}");

// React to queue depth
if (queue.MessageCount > 1000)
{
    Console.WriteLine("Queue backing up - consider throttling producers");
}

यह स्निपेट जाँच करता है कि कितने संदेश प्रतीक्षा कर रहे हैं।

प्रकाशक पुष्टि करता है

प्रकाशक पुष्टि करता है आपको यह बता दें कि जब RabbitMQ ने आपका संदेश सफलतापूर्वक प्राप्त और प्रक्रमित किया है।

// Enable publisher confirms
channel.ConfirmSelect();

var body = Encoding.UTF8.GetBytes("Hello, Queue!");

channel.BasicPublish(
    exchange: "",
    routingKey: "tasks",
    basicProperties: null,
    body: body);

// Wait for confirmation - timeout indicates backpressure
bool confirmed = channel.WaitForConfirms(TimeSpan.FromSeconds(5));

if (!confirmed)
{
    Console.WriteLine("Message not confirmed - broker may be under pressure");
}

यदि RabbitMQ संघर्ष कर रहा है, पुष्टिकरणों में अधिक समय या पूरी तरह से समय समाप्त हो जाता है. आपका उत्पादक अधिक दबाव पर जमा करने के बजाय पीछे हटने के लिए इस संकेत का उपयोग कर सकता है

उच्च के लिए -throughput परिदृश्यों के लिए

channel.ConfirmSelect();

var outstandingConfirms = new ConcurrentDictionary<ulong, string>();

channel.BasicAcks += (sender, ea) =>
{
    if (ea.Multiple)
    {
        var confirmed = outstandingConfirms.Where(k => k.Key <= ea.DeliveryTag);
        foreach (var entry in confirmed)
        {
            outstandingConfirms.TryRemove(entry.Key, out _);
        }
    }
    else
    {
        outstandingConfirms.TryRemove(ea.DeliveryTag, out _);
    }
};

channel.BasicNacks += (sender, ea) =>
{
    // Message was rejected - implement retry logic
    Console.WriteLine($"Message {ea.DeliveryTag} was nacked - broker under pressure");
    // Back off before retrying
};

एक्सपोनेंशल बैकऑफ़ के साथ पुनः प्रयास करें

जब आप बैकप्रेस पता लगाते हैं, तो सबसे बुरी बात जो आप कर सकते है यह है कि आप तुरंत पूर्ण गति पर पुनः प्रयास करें।

public async Task PublishWithBackpressureAsync(
    IModel channel,
    byte[] body,
    int maxRetries = 5)
{
    int attempt = 0;

    while (attempt < maxRetries)
    {
        try
        {
            channel.ConfirmSelect();
            channel.BasicPublish(
                exchange: "",
                routingKey: "tasks",
                basicProperties: null,
                body: body);

            if (channel.WaitForConfirms(TimeSpan.FromSeconds(5)))
            {
                return; // Success
            }

            throw new Exception("Publish not confirmed");
        }
        catch (Exception ex)
        {
            attempt++;

            if (attempt >= maxRetries)
            {
                throw new Exception($"Failed to publish after {maxRetries} attempts", ex);
            }

            // Exponential backoff: 1s, 2s, 4s, 8s, 16s
            var delay = TimeSpan.FromSeconds(Math.Pow(2, attempt - 1));
            Console.WriteLine($"Backpressure detected - retry {attempt} after {delay}");

            await Task.Delay(delay);
        }
    }
}

यह एचटीटीपी अनुकरण करता है

In-प्रक्रिया बैकप्रेस के लिए चैनल का उपयोग

यदि आप एक आंतरिक पाइपलाइन निर्मित कर रहे हैं Channel<T> बैकप्रेस समर्थन प्रदान करता है

// Create a bounded channel - backpressure is automatic
var channel = Channel.CreateBounded<WorkItem>(new BoundedChannelOptions(100)
{
    FullMode = BoundedChannelFullMode.Wait // Block producer when full
});

// Producer - will automatically wait when channel is full
async Task ProduceAsync(ChannelWriter<WorkItem> writer)
{
    for (int i = 0; i < 10000; i++)
    {
        var item = new WorkItem { Id = i };

        // This awaits if the channel is at capacity
        await writer.WriteAsync(item);

        Console.WriteLine($"Produced item {i}");
    }

    writer.Complete();
}

// Consumer - processes at its own pace
async Task ConsumeAsync(ChannelReader<WorkItem> reader)
{
    await foreach (var item in reader.ReadAllAsync())
    {
        // Simulate slow processing
        await Task.Delay(100);
        Console.WriteLine($"Processed item {item.Id}");
    }
}

// Run both concurrently
await Task.WhenAll(
    ProduceAsync(channel.Writer),
    ConsumeAsync(channel.Reader)
);

बार्डन चैनल स्वचालित रूप से बैकप्रेस लागू करता है।

एक पूर्ण बैकप्रेस

यहाँ' एक अधिक पूर्ण उदाहरण है, जो एक साथ मॉनिटरिंग लाता है। निरंतर संदेश संदेश को बचाने के लिए ब्रोकर पुनः आरंभ करता है:

public class BackpressureAwarePublisher : IDisposable
{
    private readonly IConnection _connection;
    private readonly IModel _channel;
    private readonly string _queueName;
    private readonly int _queueDepthThreshold;

    public BackpressureAwarePublisher(
        string hostName,
        string queueName,
        int queueDepthThreshold = 1000)
    {
        // See https://www.rabbitmq.com/docs/connections for connection best practices
        var factory = new ConnectionFactory { HostName = hostName };
        _connection = factory.CreateConnection();
        _channel = _connection.CreateModel(); // Creates a channel - see https://www.rabbitmq.com/docs/channels
        _queueName = queueName;
        _queueDepthThreshold = queueDepthThreshold;

        _channel.QueueDeclare(
            queue: queueName,
            durable: true,
            exclusive: false,
            autoDelete: false);

        _channel.ConfirmSelect();
    }

    public async Task<bool> PublishAsync(byte[] body, CancellationToken ct = default)
    {
        // Check queue depth first
        var queueInfo = _channel.QueueDeclarePassive(_queueName);

        if (queueInfo.MessageCount > _queueDepthThreshold)
        {
            Console.WriteLine($"Queue depth {queueInfo.MessageCount} exceeds threshold - applying backpressure");

            // Wait for queue to drain a bit
            while (queueInfo.MessageCount > _queueDepthThreshold * 0.8)
            {
                await Task.Delay(1000, ct);
                queueInfo = _channel.QueueDeclarePassive(_queueName);
            }
        }

        // Publish with retry
        for (int attempt = 1; attempt <= 3; attempt++)
        {
            try
            {
                var properties = _channel.CreateBasicProperties();
                properties.Persistent = true;

                _channel.BasicPublish(
                    exchange: "",
                    routingKey: _queueName,
                    basicProperties: properties,
                    body: body);

                if (_channel.WaitForConfirms(TimeSpan.FromSeconds(5)))
                {
                    return true;
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine($"Publish attempt {attempt} failed: {ex.Message}");
            }

            if (attempt < 3)
            {
                await Task.Delay(TimeSpan.FromSeconds(Math.Pow(2, attempt)), ct);
            }
        }

        return false;
    }

    public void Dispose()
    {
        _channel?.Dispose();
        _connection?.Dispose();
    }
}

सर्वोत्तम कार्यकलाप

RabbitMQ, के साथ विश्वसनीय प्रणालियों का निर्माण करने की व्यापक समीक्षा के लिए आधिकारिक देखें विश्वसनीयता निर्देशिका.

शांत रहें

जब कतारें बढ़ती हैं तो भयभीत मत होना। स्थिर कतार जो बिना किसी सीमा के नहीं बढ़ता है

समय के साथ कतार गहराई का मॉनीटर करें।

व्यावहारिक होना

पैटर्नों को व्यावहारिक रूप से लागू करें

अपने आप से पूछें-: "What'if this message is lost or delayed, whats the actual cost

स्मार्ट मापें

जब कतारें बैकअप कर रहे हैं, उत्तर यह है-' हमेशा नहीं होता।

विचार करें

  • सबसे पहले उपभोक्ताओं को आकार दें बैकलॉग को संसाधित करने के लिए आप अधिक कर्मचारी जोड़ सकते हैं
  • bottlenecks के लिए जाँचें एक धीमी नीचे प्रवाह निर्भरता है जो बैकअप को उत्पन्न करता है
  • जहां संभव, बैच करें उपभोक्ता एक ही समय में कई संदेशों को संसाधित कर सकते हैं
flowchart TD
    A[Queue Growing] --> B{Consumer<br/>Saturated?}
    B -->|Yes| C[Add Consumers]
    B -->|No| D{Downstream<br/>Bottleneck?}
    D -->|Yes| E[Fix/Scale Downstream]
    D -->|No| F{Can Batch<br/>Process?}
    F -->|Yes| G[Implement Batching]
    F -->|No| H[Accept Higher Latency<br/>or Reduce Load]

    style C stroke:#10b981,stroke-width:2px
    style E stroke:#f59e0b,stroke-width:2px
    style G stroke:#0ea5e9,stroke-width:2px

मॉनिटर और चेतावनी

उपयोग करें रब्बिटएमक्यू प्रबंधन प्लगइनName या एचटीटीपी API व्यापक निगरानी के लिए

  • गहराई की कतार सीमाओं से अधिक है
  • उपभोक्ता रुकावट बढ़ रही है
  • प्रकाशक समय समाप्ति की पुष्टि करता है
  • कनेक्शन अवरोधित घटनाएँ

आप बैकप्रेसर के बारे में जानना चाहते हैं पहले यह एक संकट बन जाता है

निष्कर्ष

पीछला पेज होम पेज अगला पेज पीछली-पीछे दबाव केवल दर सीमित नहीं है, यह एक जीवित रहने की रणनीति है. इसे उत्पादक और उपभोक्ता के बीच वार्तालाप के रूप में देखते हुए, आप ऐसी प्रणालियां बनाते हैं जो दबाव के अधीन लचीले रहते हैं.

प्रमुख अंतर्दृष्टि

  1. बैकप्रिसर फीडबैक है - उत्पादकों और उपभोक्ताओं को निरंतर उत्पादन उपलब्ध कराने के लिए सहयोग करना
  2. मॉनिटर कतार गहराई - आप कर सकते हैं
  3. उपयोग प्रकाशक पुष्टि करता है - पता है जब दलाल झगड़ा कर रहा है
  4. एक्सपोनेन्शल बैकऑफ़ लागू करें - DonM SK1thammer a system that's already under pressure
  5. पैमाने पर उपभोक्ता-, न केवल उत्पादक - bottleneck ठीक करें

जब आपका तंत्र कहता है, ‘"’I‘'’m full’,’ give me a minute“,"” the correct response is “"”No problem”,” I“'”ll wait“MSC7” That“M SK8”s the essence of well”-”behaved distributed systems“MST10”polite”MST11”cooperative“Mst12”and resilient”

जब कतारें थोड़ा बढ़ती हैं तो शांत रहें।

Finding related posts...
logo

© 2026 Scott Galloway — Unlicense — All content and source code on this site is free to use, copy, modify, and sell.