后压是分布式系统的无声英雄。 它使您的队列不至于在制片人发布信息的速度快于消费者能够咀嚼的速度时在接缝上爆裂。 简而言之, 系统是说“ 等一等 ” , 当事情太忙的时候。
事情是这样的: 本条中的手法适用于 每个 列队列和服务巴士- RabbitMQ, Kafka, Azure Service Bus, AWS SQS, NATS, 由您命名。 细节各有不同, 但原则是普遍性的。 我将使用 RabbitMQ 来举大多数例子, 因为这是我最了解的, 但我会向您展示这些模式是如何在平台上翻译的 。
供认: 即使是大多数高级开发商也不执行适当的后压处理。他们建立快乐病态系统,这些系统在进化和中转方面运作良好,然后问为什么生产在黑色星期五期间会停止。后压处理是把“它起作用”和“它的规模”区分开来的技术之一。如果你不考虑,你正在建立一个最终在载荷下失败的系统。
在其核心,后压是一个反馈循环,在消费者落后时会减缓生产者的速度。 想象一下它就像滑坡路上的交通灯光 — — 你不能随心所欲地在高速公路上堆积。 灯光控制着车流,让汽车安全地融合,而不会造成堆积。
没有后压,快速制片人会压倒一个缓慢的消费者。 信息堆积在队列中,记忆耗尽,最终你的系统会崩溃。 后压在灾难袭击前说“稳定下来 ” 。
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
后压的美丽之处在于 对话框 消费者的信号是"我饱了,给我一分钟" 制片人回答"没问题,我会等"
RabbitMQ有几种处理后压的内置机制,理解它们至关重要,如果你正在建立需要保持直立状态的系统。 RabbitMQ 文件 非常好——我将会与特定页面链接。
当 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不只是在压力下传递信息,它会减缓源头的速度。这比静悄悄地丢弃数据要文明得多。
消费者控制速度 承认的确认 在消费者明确承认之前,一个信息不会从队列中删除。如果消费者没有足够快的ACK信息,则队列会增加,最终会触发上游的流量控制。
您也可以使用 附加权限值 ( Qos) 来控制消费者可以同时在飞行中持有多少个未知信息。 这样可以防止单个慢速消费者存储信息 。
// Set prefetch count to limit unacknowledged messages
channel.BasicQos(prefetchSize: 0, prefetchCount: 10, global: false);
这告诉RabbitMQ: “每次只给我发送10条信息。 一旦我整理了一些信息, 你可以发送更多的信息。” 消费者明确表示它能承受多少压力。 NET 客户文件 详细覆盖 API 。
这些模式是通用的,但执行方式却不同。这是其他一些大众信息系统如何处理同样的问题。
Kafka采取了一种根本不同的方法——消费者 拉拉拉 这让回压隐含了:如果消费者不进行民意调查,它不接受信息。经纪人不在乎;它只是在消费者准备好之前保持信息。
// 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
}
聪明的一点: Kafka 的消费群体自动重新平衡分区。 如果一个消费者落后, 您可以在组内增加更多的消费者, 分区可以重新分配。 后压会变成一个缩放决定 。
// 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
};
服务客车使用 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();
服务车也支援AxAsure服务车 届会届会的届会和届会的届会 被命令处理和 死字母队列 对于屡次失败的信息来说 两者都很重要 当事情发生时 管理压力
SQS 使用可见度超时作为它的后压机制。 当您收到一条信息时, 它会成为其他消费者的隐形信息。 如果您不及时删除它, 它会重新出现, 让其他人尝试 。
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);
}
}
智能 SQS 骗术: 使用 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 JetStream具有明确的流量控制,包括消费者确认和最多待定电文限制:
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());
注意到所有这些系统的共同点如下:
语法不同,但舞蹈是一样的: "这是我能处理的,告诉我什么时候处理 如果我没有及时告诉你,假设我失败了"
对,让我们进入代码。这里是应用 C# 应用程序中执行和应对后压的实际例子。
首先,你无法管理无法测量的东西。这里是如何检查在队列中等待的讯息数量 :
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在挣扎,确认需要更长的时间或全部时间。 您的制片人可以使用这个信号后退, 而不是在更大的压力下堆积。
对于高通量情景, 你会想要 的非同步确认:
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);
}
}
}
这模仿了 HTTP 的 429 模式( 要求过多 ) 。 我们没有敲打经纪人,而是在重试之前暂停, 给系统恢复的时间 。
如果您正在建造一个内部管道(生产器 处理器 消费者 全部在您的申请中) , .NET's 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)
{
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();
}
}
当队列增长时, 不要惊慌。 稍微深一点是正常和健康的—— 这意味着您的系统正在以优雅的方式吸收负荷。 目标不是一个空队列; 它是一个 稳定 队列不会无限制地增长 。
监视队列的深度 。 寻找趋势, 而不是快照 。 一贯为 100 条消息的队列是好的 。 过去一个小时从 100 条增加到 10,000 条的队列需要注意 。
务实地应用模式。 并不是每个信息都需要出版商确认。 不是每个队列都需要复杂的后压处理。 每小时处理10条信息的队列可能不需要与每秒处理10,000条相同的复原力工程。
问自己:“如果这个信息丢失或延迟,实际成本是多少?”如果答案是“不多 ” , 不要过度设计。如果答案是“巨大的财务或数据完整性影响 ” , 投资适当的回压处理。
当队列备份时, 答案并不总是“ 增加更多的制片人 ” 。 这就像试图通过增加汽车来修补交通堵塞。
考虑:
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
设置下列警报:
你想知道后压的情况 之前 它变成了一场危机, 而不是当你的系统已经崩溃了。
后压不仅仅是限制利率 — — 这是一种生存策略。通过将它作为生产者和消费者之间的对话,你建立了在压力下保持弹性的系统。
关键见解:
当你的系统说"我饱了,给我一分钟"时, 正确的回答是“没问题,我会等的。"这就是行为良好的分布系统的实质——政治、合作和弹性。
当队列增长一点点时保持冷静, 并记住: 优雅的后压下的系统比完全倒塌的系统要好得多。
© 2026 Scott Galloway — Unlicense — All content and source code on this site is free to use, copy, modify, and sell.