This is a viewer only at the moment see the article on how this works.
To update the preview hit Ctrl-Alt-R (or ⌘-Alt-R on Mac) or Enter to refresh. The Save icon lets you save the markdown file to disk
This is a preview from the server running through my markdig pipeline
Sunday, 23 November 2025
La contropressione è l'eroe unsung dei sistemi distribuiti. È ciò che impedisce alle vostre code di scoppiare alle cuciture quando i produttori stanno sparando messaggi più velocemente di quanto i consumatori possono masticare attraverso di loro. In parole povere: è il sistema che dice "aspettate un momento" quando le cose diventano troppo occupato.
Il fatto e' questo: le tecniche di questo articolo si applicano praticamente a ogni la coda dei messaggi e il bus di servizio.RabbitMQ, Kafka, Azure Service Bus, AWS SQS, NATS, lo nomini. Le specifiche variano, ma i principi sono universali. Utilizzerò RabbitMQ per la maggior parte degli esempi perché è quello che conosco meglio, ma vi mostrerò come questi modelli si traducono attraverso le piattaforme.
Una confessione: Anche la maggior parte degli sviluppatori senior non implementare la corretta gestione della contropressione. Costruiscono sistemi happy-path che funzionano bene in dev e messa in scena, quindi domandarsi perché la produzione cade durante Black Friday. La gestione della contropressione è una di quelle tecniche che separa "funziona" da "it scale." Se non ci stai pensando, si sta costruendo un sistema che alla fine fallirà sotto carico.
Al suo centro, la contropressione è un ciclo di feedback che rallenta i produttori quando i consumatori sono indietro. Pensate ad esso come semafori su una strada di scorrimento non si può semplicemente accumulare in autostrada ogni volta che si vuole. Le luci controllano il flusso, lasciando le auto si fondono in modo sicuro senza causare un pile-up.
Senza contropressione, un produttore veloce sopraffarà un consumatore lento. I messaggi si accumulano nelle code, la memoria si esaurisce, e alla fine il sistema cade. La contropressione dice "costante" prima che il disastro colpisca.
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 bellezza della contropressione è che è un conversazione tra produttore e consumatore. Il consumatore segnala "Sono pieno, dammi un minuto" e il produttore risponde "Nessun problema, aspettero'." E' educato, collaborativo, e impedisce a tutti di cadere.
RabbitMQ ha diversi meccanismi integrati per gestire la contropressione, e comprenderli è fondamentale se si stanno costruendo sistemi che devono rimanere in posizione verticale sotto carico. Documentazione RabbitMQ è eccellente il collegamento di pagine specifiche mentre andiamo.
Quando l'utilizzo della memoria o la profondità della coda di RabbitMQ supera le soglie configurate, attiva controllo del flusso. Questo blocca temporaneamente le connessioni editore I publishers non possono inviare nuovi messaggi fino a quando il broker ha eliminato abbastanza ritardo. Vedere anche i documenti su allarmi di memoria e allarmi di dischi che attivano il controllo del flusso.
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
L'intuizione chiave qui è che RabbitMQ non rilascia solo i messaggi quando sotto pressione si rallenta la sorgente. Si tratta di un approccio molto più civilizzato rispetto allo scarto silenzioso dei dati.
I consumatori controllano il ritmo attraverso riconoscimenti (ACKs e NACKs) Un messaggio non viene rimosso dalla coda finché il consumatore non lo riconosce esplicitamente. Se un consumatore non invia messaggi ACK abbastanza velocemente, la coda cresce [54]che alla fine attiva il controllo del flusso a monte.
Puoi anche usare Limiti di prefetch (QoS) controllare quanti messaggi sconosciuti un consumatore può avere in volo contemporaneamente. Ciò impedisce ad un singolo consumatore lento di accumulare messaggi.
// Set prefetch count to limit unacknowledged messages
channel.BasicQos(prefetchSize: 0, prefetchCount: 10, global: false);
Questo dice a RabbitMQ: "Inviami solo 10 messaggi alla volta. Una volta che IO ACK alcuni, si può inviare di più." E 'il consumatore esplicitamente dicendo quanta pressione può gestire. Documentazione del client .NET copre in dettaglio le API.
I modelli sono universali, ma le implementazioni differiscono. Ecco come alcuni altri sistemi di messaggistica popolare affrontano lo stesso problema.
Kafka adotta un approccio fondamentalmente diverso [47]consumatori tira Questo rende implicita la contropressione: se un consumatore non fa sondaggi, non riceve messaggi. Il broker non si preoccupa; mantiene i messaggi in giro fino a quando il consumatore non è pronto.
// 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
}
La punta intelligente: i gruppi di consumatori di Kafka ribilanciano automaticamente le partizioni. Se un consumatore rimane indietro, è possibile aggiungere più consumatori al gruppo e le partizioni vengono ridistribuite. La contropressione diventa una decisione di scala.
// 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 utilizza un MaxConcurrentCalls l'impostazione che è splendidamente semplice che controlla quanti messaggi il processore gestisce contemporaneamente. La contropressione è automatica.
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 supporta anche sessioni per il trattamento ordinato e code con lettere morte per i messaggi che falliscono ripetutamente sia importante per la gestione della pressione quando le cose vanno male.
SQS utilizza i timeout di visibilità come meccanismo di contropressione. Quando si riceve un messaggio, diventa invisibile agli altri consumatori. Se non lo si elimina in tempo, riappare per qualcun altro da provare.
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);
}
}
Il trucco intelligente SQS: usare ApproximateNumberOfMessages per monitorare la profondità della coda e i consumatori su scala automatica:
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 ha un controllo esplicito del flusso con riconoscimenti dei consumatori e limiti massimi di messaggio pendenti:
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());
Notate cosa hanno in comune tutti questi sistemi:
La sintassi differisce, ma la danza è la stessa: "Ecco quanto posso gestire. Dimmi quando me ne sono occupato. Se non te lo dico in tempo, supponiamo che abbia fallito."
Bene, entriamo nel codice. Ecco esempi pratici di implementazione e risposta alla contropressione nelle vostre applicazioni C#.
Le prime cose prima non si può gestire ciò che non si può misurare. Ecco come controllare quanti messaggi sono in attesa in una coda:
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");
}
Questo pezzetto di codice controlla quanti messaggi sono in attesa. Se il conteggio si arrampica, questo è il tuo segnale per i produttori di gas o scalare i consumatori. Non preoccuparti di controllare questo troppo frequentemente è di solito sufficiente un controllo periodico dello stato di salute.
Publisher conferma di farti sapere quando RabbitMQ ha ricevuto ed elaborato con successo il tuo messaggio. Se i riconoscimenti rallentano, questo è un chiaro segnale di contropressione:
// 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");
}
Se RabbitMQ è in difficoltà, le conferme richiedono più tempo o più tempo fuori del tutto. Il produttore può utilizzare questo segnale per indietreggiare piuttosto che accumulare più pressione.
Per scenari ad alto rendimento, vorrai conferma asincrona:
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
};
Quando si rileva la contropressione, la cosa peggiore che si può fare è immediatamente riprovare a tutta velocità. E 'come rispondere ad un ingorgo del traffico premendo l'acceleratore più difficile. Invece, implementare backoff esponenziale:
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);
}
}
}
Questo imita lo schema 429 (Troppo Many Requests) dell'HTTP. Invece di martellare il broker, ci fermiamo prima di riprovare, dando al sistema il tempo di recuperare.
Se state costruendo una conduttura interna (produttore → processore → consumatore tutto all'interno della vostra applicazione), .NET's Channel<T> fornisce un elegante supporto alla contropressione:
// 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)
);
Il canale limitato applica automaticamente la contropressione ai blocchi del produttore quando il canale è pieno, rallentando naturalmente il ritmo del consumatore.
Ecco un esempio più completo che riunisce monitoraggio, conferme e backoff:
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)
{
var factory = new ConnectionFactory { HostName = hostName };
_connection = factory.CreateConnection();
_channel = _connection.CreateModel();
_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();
}
}
Non fatevi prendere dal panico quando crescono le code. Un po' di profondità è normale e salutare, significa che il vostro sistema sta assorbendo i picchi di carico con grazia. L'obiettivo non è una coda vuota; è un stabile la coda che non cresce senza limiti.
Monitorare la profondità della coda nel tempo. Cercare le tendenze, non le istantanee. Una coda che è costantemente a 100 messaggi va bene. Una coda che è cresciuta da 100 a 10.000 nell'ultima ora ha bisogno di attenzione.
Applica schemi pragmaticamente. Non tutti i messaggi hanno bisogno di conferma dell'editore. Non tutte le code hanno bisogno di una gestione sofisticata della contropressione. Una coda che elabora 10 messaggi all'ora probabilmente non ha bisogno della stessa ingegnerizzazione della resilienza di un processo 10.000 al secondo.
Chiediti: "Qual è il costo effettivo se questo messaggio viene perso o ritardato?" Se la risposta è "non molto," non over-engineer. Se la risposta è "impatto significativo di integrità finanziaria o dati," investire nella corretta gestione della contropressione.
Quando le code sono di backup, la risposta non è sempre "aggiungere più produttori." E' come cercare di riparare un ingorgo aggiungendo più auto.
Considera:
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
Impostare avvisi per:
Vuoi sapere della contropressione? prima Diventa una crisi, non quando il tuo sistema e' gia' caduto.
La contropressione non è solo limitare il tasso è una tattica di sopravvivenza. Trattandolo come una conversazione tra produttore e consumatore, si costruiscono sistemi che rimangono resilienti sotto pressione.
Le principali intuizioni:
Quando il vostro sistema dice "Sono pieno, datemi un minuto," la risposta corretta è "Nessun problema, aspetterò." Questa è l'essenza di sistemi distribuiti ben educati, politi, collaborativi e resilienti.
Rimanete calmi quando le code crescono un po', e ricordate: un sistema sotto una graziosa contropressione è infinitamente meglio di uno che è caduto completamente sopra.
© 2026 Scott Galloway — Unlicense — All content and source code on this site is free to use, copy, modify, and sell.