La pression d’arrière est le héros inoubliable des systèmes distribués.

C’est là que le problème se pose. Les techniques présentées dans cet article s'appliquent à tous les message queue et bus de service—RabbitMQM SK1 Kafka, Azure Service BusMSC3 AWS SQSMska4 NATSMske5 vous le nommez . Les spécifications varientMka7 mais les principes sont universelsMKA8 J’utiliserai RabbitMQ pour la plupart des exemples car c’est ce que je connais le mieuxMkka11 mais j’utiliserais Mcka12 Je montrerai comment ces modèles se traduisent dans toutes les plates-formes

Une confession: Même la plupart des développeurs de haut niveau ne mettent pas en œuvre une bonne manipulation à l’arrière-pressure.

Qu'est-ce que la pression d'arrière?

À sa base, la pression arrière est un cycle de rétroaction qui ralentit les producteurs lorsque les consommateurs sont en retard.

Sans pression d'arrière, un producteur rapide surpeuplit un consommateur lent.1 Les messages s'accumulent dans les rangées.2 La mémoire est épuisée.3 Le système tombe à l'encontre de votre système.4 La pression d’arrière dit :

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

La beauté de la pression arrière est qu’elle Conversation entre le producteur et le consommateur.

Comment RabbitMQ manipule la pression arrière

RabbitMQ est un courtier de message source open- qui met en œuvre Protocole d'attente avancé de messages (AMQP). C’est l’un des courtiers de message les plus largement déployés, ', qui est utilisé par des entreprises allant des startups jusqu’à des entreprises comme Bloomberg.

Ce que RabbitMQ excelle à:

• Routage complexe - Échanges , Liens M SK2 et clés de routage vous permettent d'établir des topologies de routement de messages sophistiquées . Vous voulez que les messages s'écoulent à plusieurs rangées MSC4 Route basée sur les titres de l'en-tête Types d'échange (directM SK1 fanout , topic, headersMSC4 avez-vous couvert?
• Garanties de fiabilité - L'éditeur confirme, autorisations des consommateurs, et messages persistants signifie que vous pouvez construire des systèmes où la perte de message est inacceptable.
• Déploiement souple - Run it on a single node for development, or deploy a Groupe de travail avec Les couloirs de quorum pour une grande disponibilité dans la production.
• Soutien du protocole - Au-delà de AMQP, Appuie RabbitMQ MQTT (IoT), STOMP (protocole de texte simple), et a un États-Unis fonction pour Kafka-appendice similaire-logs seulementM SK2

Quand envisager des solutions de rechange: Si vous avez besoin d'un débit massif avec un routage plus simple (millions de messages/secondeM SK2 KafkaMSC3s logMST4l'architecture basée peut être plus appropriée MST5Si vousMst6 êtes tousMSST7dans un fournisseur de nuageMSS8 leurs services gérésMSR9Azure Service BusMSL10 AWS SQSMSV11 réduire les frais généraux opérationnelsMSP12

RabbitMQ possède plusieurs mécanismes de manipulation de la pression arrière, dont - et ,, et leur compréhension est cruciale si vous construisez des systèmes qui doivent rester en position droite sous les charges. Documentation RabbitMQ est excellent—IM SK1Nous ferons un lien vers des pages spécifiques à mesure que nous allons

Contrôle du flux

Lorsque l'utilisation de la mémoire de RabbitMQ' ou la profondeur de la queue dépasse les seuils configurés ,, elle est activée Contrôle du débit. Cela bloque temporairement les connexions de l’éditeurM SK1les éditeurs ne peuvent 'envoyer de nouveaux messages jusqu’à ce que le courtier ait éliminé suffisamment d’arriérés. Voir aussi les documents sur les alarmes de mémoire et Système d'alarme de disque qui déclenche le contrôle du débit.

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

Le point clé ici est que RabbitMQ n’émet pas de messages lorsqu’il se trouve sous pression, mais qu’il ralentit la source.

Remerciements des consommateurs

Les consommateurs contrôlent le rythme Remerciements (ACKs et NACKs ). Un message n’est pas retiré de la liste jusqu’à ce que le consommateur l’en reconnaisse explicitement. Si un consommateur ne le fait pas, les messages ACK4 ne sont pas suffisamment rapidesMSC5 la liste augmenteM SK6ce qui déclenche finalement le contrôle du flux en amontMNK7

Vous pouvez aussi utiliser Limites de préfetch (QoSM SK1 pour contrôler le nombre de messages non reconnus qu’un consommateur peut avoir en vol à la fois . Cela empêche un seul consommateur lent d’accumuler des messages

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

Cela dit RabbitMQ: "Envoyez-moi seulement des messages à la fois M SK2. Une fois que j’ai ACK une certaine message, , vous pouvez envoyer plus d’un message.." C’est le consommateur qui dit explicitement combien de pression il peut faire face. Documentation du client .NET couvre l’API en détail.

Comment d'autres systèmes manipulent la pression arrière

Les modèles sont universels, mais les mises en œuvres diffèrent . Voici' comment d’autres systèmes de messagerie populaires abordent le même problèmeM SK3

Kafka:Consommateur-Contrôle du scrutin

Apache Kafka est une plate-forme de diffusion d’événements distribuée conçue pour des messages hautement tolérants.

Kafka adopte une approche fondamentalement différente—consommateurs traction Le courtier ne prend pas en considération les messages qu’il transmet plutôt que de les faire pousser. Kafka Documentation pour les consommateurs pour plus de détails sur le modèle de vote.

// 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
}

Le bit clever: Kafka' les groupes de consommateurs rééquilibrent automatiquement les partitions . Si un consommateur est à l’arrièreM SK3 vous pouvez ajouter plus d’consommateurs au groupe et les parties sont redistribuées.

// 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 Service Bus: Contrôle des messages concurrents

Service d'Azure est Microsoft'le courtier des messages d’entreprise entièrement géré avec des queues et publie les sujets à souscrire.

Azure Service Bus utilise un MaxConcurrentCalls Settings that's beautifully simple —it controls how many messages your processor handles simultaneouslyM SK2 Backpressure is automatic. The Documentation de traitement des messages couvre le réglage du rendement en détail.

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 Service Bus appuie également Sessions pour le traitement des commandes et mort-carte des lettres pour les messages qui échouent à maintes reprises—ils sont tous deux importants pour gérer la pression lorsque des choses vont mal

AWS SQS: Visibilité Démarrage de la danse

Amazon Simple Queue Service (SQS) est un service de messagerie entièrement géré par AWS.

Utilisations de SQS Échéances de visibilité en tant que mécanisme de répression arrière. Lorsque vous recevez un message, il devient invisible aux autres consommateursM SK2 Si vous ne le supprimez pas dans les délais voulus ' il apparaît à quelqu’un d’autre pour qu’il l’essaie

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);
    }
}

Utilisation intelligente des SQS: ApproximateNumberOfMessages pour surveiller la profondeur de la queue et les consommateurs à grande échelle

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 JetStream: Contrôle du débit intégré

NATS est un système de messagerie natif . JetStream est NATS's construiteM SK1in couche de persistance qui ajoute des cours d’eau durables ,exactment-une fois livréesMST4 etMst5importante pour nous MST6Contrôle explicite du flux.

NATS JetStream a un contrôle explicite du débit avec les acknowledgements des consommateurs et des limites maximales de messages en attente:

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());

Le Thread commun

Notez ce que tous ces systèmes ont en commun:

1. Limites de concurrence explicites - vous contrôlez combien vous êtes disposés à manipuler
2. Flux basé sur la reconnaissance- - les messages restent disponibles jusqu'à ce que vous confirmez le traitement
3. Récupération basée sur l’arrêt de temps- - si vous échouez, les messages retournent pour essayer de nouveau
4. Surveillance de la profondeur - vous pouvez toujours demander " comment je suis-je à l'appui ?

La syntaxe diffère, mais la danse est la mêmeM SK1 "C'est là, 's combien je peux faire face à ce problème,. Dites-moi quand j'ai fait face à celui-ci.

Exemples de code C#

A droite, laissezM SK1s entrer dans le code. Voici des exemples pratiques de mise en œuvre et de réponse à la pression d'arrière dans vos applications C #

Surveillance de la profondeur des listes

Les premières choses sont les suivantes : first things—you can't manage what you can 't measureM SK3 HereMSC4s how to check how many messages are waiting in a à l'échelon:

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");
}

Cette partie vérifie combien de messages sont attendus.

L'éditeur confirme

L'éditeur confirme vous indiquer quand RabbitMQ a reçu et traité votre message avec succès. Si les réceptions ralentissentM SK1 cela ' est un signal clair de pression arrière:

// 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");
}

Si RabbitMQ éprouve des difficultés, les confirmations prennent plus de temps ou s’arrêtent complètementM SK1 Votre producteur peut utiliser ce signal pour se soustraire plutôt que d’augmenter la pression.

Pour les scénarios de haut débit-throughput, vous voudrez des confirmations asynchronesM SK3

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
};

Essayer de nouveau avec le retrait exponentiel

Lorsque vous détectez une pression d’arrière, le pire que vous pouvez faire est de réessayer immédiatement à la pleine vitesse.

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);
        }
    }
}

Ce protocole ressemble à HTTP's M SK1 (Too Many Requests) patternMSC4 Au lieu d’abattre le courtier, nous mettons un pause avant de l’essayer de nouveau.

Utilisation de canaux pour la pression arrière du processus In-

Si vous construisez un pipeline interne (producteur →processeur | →consommateur tout au sein de votre application Channel<T> fournit un support d'appui à la pression arrière élégante:

// 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)
);

Le canal limité applique automatiquement la pression de recul—le producteur bloque lorsque le canal est pleinM SK1 ralentit naturellement pour correspondre au rythme du consommateur ' pas nécessaire à l’interruption manuelle.

A Complete Backpressure-Aware Publisher

Voici un exemple plus complet qui réunit la surveillance, confirme , et l’arrêt de travail. Veuillez noter que messages persistants pour assurer que les messages survivent le relais du courtier:

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();
    }
}

Pratiques exemplaires

Pour un aperçu détaillé des systèmes fiables de construction avec RabbitMQ,, voir le document officiel Guide de fiabilité.

Restez calmes

Don' ne paniquez pas lorsque les rangées augmentent . Un peu de profondeur est normale et saine— cela signifie que votre système absorbe gracieusement les pics de chargeM SK3 L’objectif n’est pasMSC4t une liste vide Établissement stable queue qui n’augmente pas sans limite

Surveiller la profondeur de la queue au fil du temps. Chercher les tendancesM SK1 non des snapshots . Une queue qui ' est constamment à l’essai des messages |100 est en bon état | . Unequeue qui, depuis l’heure écoulée, a augmenté de ', passant de ♪100 à ♪ 10,000 mérite d’être prise en considération

Être pragmatique

Appliquer les modèles de façon pragmatique. Non tous les messages ont besoin d’un éditeur pour confirmer leur utilisation. Non toutes les cotes nécessitent une manipulation sophistiquée à l’arrière-pressureM SK2 Une cote qui traite des messages par heure 10 ne nécessite probablement pas la même résilience génie que le traitement d’une seule opération.

Demandez-vous ce qui suit :

Étalonner intelligentement

Lorsque les couloirs s’appuient, la réponse est : « ' n’est pas toujours le cas ».

Examiner:

• Élargir d'abord les consommateurs - pouvez-vous ajouter d’autres travailleurs pour traiter l’arriéré?
• Vérifier les lacunes - est une dépendance lente en aval qui cause le backup
• Partie lorsque possible - les consommateurs peuvent traiter plusieurs messages à la fois

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

Surveillance et alerte

Utilisez le Logiciel de gestion RabbitMQ ou HTTP API pour une surveillance globale. Établir des alertes pour:

• La profondeur de la liste excède les seuils
• Le retard des consommateurs augmente
• L'éditeur confirme le retrait du calendrier
• Événements bloqués de connexion

Vous voulez savoir plus sur la répression arrière précédemment Il devient une crise, pas lorsque votre système a déjà chuté.

Conclusion

La pression d’arrière n’est pas seulement une limite de vitesse, mais aussi une tactique de survie. En la traitant comme une conversation entre le producteur et le consommateur, vous bâtissez des systèmes qui demeurent résistants à la pression.

Les principaux points de vue:

1. La pression d'arrière est une rétroaction - Les producteurs et les consommateurs collaborent pour trouver un rendement durable
2. Surveiller la profondeur de la queue - vous pouvez ' ne gèrez pas ce que vous pouvez
3. Utiliser la confirmation de l'éditeur - sait quand le courtier est en difficulté
4. Mettre en œuvre un arrêt exponentiel - DonnM SK1thammer a system that's already under pressure
5. Les consommateurs de grande envergure, ,, non seulement les producteurs - corriger le crâne d’étranglement , non le symptôme

Lorsque votre système dit "IM SK1m plein,donnez-moi un minuteMSC3la réponse correcte est "Nous avons aucun problèmeMST5JeM ST6je vais attendreM st7 C’est ce qui constitue l’essence d’un système bien répartiMst9conduite par des systèmes distribuésMSt10politiqueM St11coopérativeMstr12 et résilienteMrt13

Gardez-vous calme lorsque les couloirs augmentent un peu.