后压是分布式系统不响的英雄@. @ it'}让您的队列不会在裁缝中爆裂, 当制片人发信息的速度快于消费者能够通过它们咀嚼的速度时#._简略地说:}(itMS K4})这个系统表示“"}等一下<"}当事情变得太忙的时候 @.}

在这里,'的东西 _: 本条中的手法适用于 每个 您所命名的队列与服务总线@— @RabbitM,}(卡夫卡),}Azure Service Bus* ,} AWSSQSMS K4}NATSMPK5}您所命名的具体细节有差异

一种忏悔 即使大多数高级开发商都实施适当的后压处理@.}他们也建立快乐的-path系统, 在 dev 和 streading ,}然后想知道为什么在黑色星期五期间生产会下降?

后压是

当消费者落后于“.”时, 反压是一个反馈回路, 它会让生产者慢下来。 想象一下它就像滑行路上的交通灯光: “—”你就可以在“'”上堆到高速公路上, 当你喜欢“MS K4”的时候,

快速制片人若没有回压@, @ a fast professor 将会淹没在队列中堆积的慢速消费信件@ . @ ,} 记忆力将耗尽 *, 最终你的系统会跌落到 <.# Backpressure MS K5{speaty on @"} 在灾难袭击前 @MSSK7}

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

后压的美丽之处在于 对话框 消费者信号 @"I'm full_,给我一分钟#"和制片人回应@"没有问题$,I'}等待礼貌的,协作}, 让每个人都不要跌倒 在.

RabbitMQ 处理后压如何

兔子兔MQ 是一个执行此命令的 Open+- 源代码消息中介 高级信件排队协议@(AMQP)“.”和“I(”是公司从新创办企业到Bloomberg ,Pivotal,和VMware@MSK5等企业使用的信息最广泛的信息经纪人之一。

兔子超赞的

• 复杂路线 -Q Exchanges @,Q 绑定@,Q 和路由密钥让您建立精密的信息路由:%.Q 想要在信头_?}(直接针对特定消费者)的基础上向多个队列发送信息 交换模式的交流、交换和 (pirect,fanoutMS K2 主题,headers)}你是否覆盖过.
• 可靠性担保 - 出版商确认, 消费方承认,和 持久性信息 表示您可以建立无法接受信息丢失的系统@. @%
• 灵活部署 - 运行在开发的单一节点上@,或部署 a 组群 与 法定人数队列 生产中高供应量
• 礼宾支持 @-超越AMQP,兔子支持 MQTT MQT TT # (#IOT#I),# 斯托姆 (simple 文本协议@),}并有一个 溪流 Kafka- 的功能类似附件@ - @ 仅logs @ .}

何时考虑替代品@: @% 如果您需要使用更简单的路由“( 百万条消息”来进行大规模输送, 请用您的管理服务 “(Azure Service Bus@,} AWSSQS) 降低运行的顶部“MS K12 ”

RabbitMQ 拥有若干已建的“-in”处理后压机制, rabbitMQ 文件 当我们去“.”时,它会链接到特定的页面。

流动控制

当 RabitM' 内存用量或队列深度超过配置的阈值@ , 它激活 流量控制此临时封隔出版商连接@— @publishers can'}在经纪人清理完足够积压的邮件前, 内存提醒 和 磁盘警报器 触发电流控制

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 在压力下只发送消息 =—} 它会减慢源@.}这是比静悄悄地丢弃数据更文明的方法 .

消费者致谢

消费者控制速度 承认的确认 'QT ACK K5 队列增速=— 最终触发上游流量控制=MSC7}

您也可以使用 附加权限值 以控制消费者在飞行中能够同时收到多少未确认的信息@.}这样可以防止一个慢速的消费者存储信息 @.

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

这告诉RabbitM:+ " 一次只给我发@10 信息$只要我加了点",你就可以发送更多"MSKO5}'}消费者明确表示它能承受多少压力{.} .NET 客户文件 详细覆盖 API+.

其它系统如何处理后压

这些模式是通用的 @,, 但执行方式却不同 @MS K1 这里#' 其它流行信息系统如何处理同样的问题?

Kafka:消费者 □-控制投票

阿帕契卡卡夫卡 是一个分布式事件流平台,设计用于高@- @topput_,}错误 @-#容恕性讯息.}和传统信息经纪人不同{,}卡夫卡在附件>-}仅日志中存储消息并让消费者跟踪自己的位置\ (offsetsMS K7}

卡夫卡采取了完全不同的方法 消费者 拉拉 而不是让它们推动信件@ . @ 这样使回压隐含 @ : @ 如果一个消费者不这样做的话 '} t polks@ I, 它不接受消息_ MS K4} 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
}

如果一个消费者落后于“,” ,您可以在该组中增加更多的消费者,而分区获得再分配=.}后压则成为缩放决定_.

// 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: 同步信件控制

服务客车 是微软' 完全管理的企业信息经纪人,配有队列并出版*- 注册题材 . ItMS K3 设计用于在云层中解开应用程序和服务

服务客车使用a MaxConcurrentCalls 设置“' @%s beautifully simple”\—}控制您处理器同时处理多少信件@.}后压是自动 @.} 信息处理文件 包含详细性能调控的功能@. @%

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

服务客车也支援 届会各届 被命令处理和 $-}字母队列 对于多次失败的信件 —}两者都很重要 当事情出错时

AWS SQS: 可见度超时舞蹈

Amazon 简单队列服务 @(@SQS) @% 是一个来自 AWS @.}的完全管理的信件排队服务, 它提供两种队列类型@:标准 @ (Best-#effort 命令{,}在-lent-once delietion_)和FIFO MS K8exactly-ONce programing},}严格命令<).}

SQS 用途 可见度超时 当您接收到一条消息 , n它成为其他消费者的看不见@.}如果你在时间中做\ I'\ t删除它..."," 它会重新出现,让其他人尝试\ O. >

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 trick: 使用 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 : 流动控制装置

海军 是一个轻量级的 , 高 MS K1 性能云 -native信息系统. 喷气流 是 NATS's 建造的MS K1#in 持久性层,增加耐久性流 ,} 确切地说-once sufferation,和{—}对于我们来说很重要 ~—直径流量控制.

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

共同线索

注意所有这些系统的共同点 @:@

1. 明度批量/货币限额 - 你控制你多愿意处理
2. 致谢=-基流 -}在您确认处理前, 信件一直可用 。
3. 超时@-基于恢复 - 如果您失败了 @, 信件返回重试
4. 深度监测 - 你总是可以问" "I?""的备份吗?

语法不同 @,}但舞步相同#: . 告诉我 I''"已经处理过它的时候 如果我不做'"

C# 代码示例

在您的 CQ# 应用程序中执行和应对后压的实际实例@. @%

队列深度监测

首先 — 您可以控制您能管理什么 您可以% K2 @ t量度@ . @ 在这里 _'}这里如何检查有多少信件在等待 队列:

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

此片段检查有多少消息等待@ . @ 如果计数在攀爬 @ MS K1, 那么% '{ 提示您加速生产商或扩大消费者范围@ I._ Doñ'} 不要担心过频繁的检查@ MPK5a 定期健康检查通常就足够了 @.}

发布器确认

出版商确认 当 RabbitMQ 成功接收和处理您的信件时, 请告知您@ . @ 如果确认速度放慢了 @ MS K1} _ ' k4 a clear sign of backpressure_ :}

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

如果 RabitMQ 正在挣扎 @,确认需要更长的时间或完全超时@. 您的制片人可以使用此信号后退, 而不是用更多压力堆积 #.#

对于高“-”的通量假想 “, ”您想要非同步确认“MS K3 ”

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

使用指数后退重试

当您检测到回压@ , @ 最坏的操作是立即全速重试@ I.} 这就像按下加速器来应对交通阻塞吗?

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 ' @%s @ MS K1 @ @ @ I( @ Too many request @ MPK3}模式@ ._ 而不是敲打经纪人@,# 我们暂停, 然后重试,# 让系统有时间恢复 #. @

使用通道用于 Inç- process 后压

如果您正在建造一个内部管道, 请在您的应用程序中将您所有的消费方 ), @ .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)
);

连接的频道自动应用回压@ — @ 生产区块, 当该频道是全Q,} 自然减速以匹配消费者'} 步调.} 不要求手动抽动\ MS K4}

完整后压@ - @Aware 发布器

这里的'* 是一个更完整的例子, 持久性信息 来确保信件在经纪人身份中有效重现@: @%

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

最佳做法

有关使用RabitM,建立可靠系统的全面概况,请见官方网站 可靠性指南.

保持冷静

当队列增长时, Don't 恐慌@ .# 有一点点深度是正常的和健康的 @ MS K2\it 表示您的系统正在以优雅的方式吸收负载加注Q.}目标为 @ MPK4t 一个空的队列@; it'}a 稳定 队列中无约束的增减@ ' @ title

监视队列的深度@. 寻找趋势 @ , 而非快照 @ MPK2 一个在 @ I100 信件中始终一致的队列\ MS K3 是很好的@ . 一个在过去一个小时里从 @ MCK7 增长到 @10,000 需要注意 @ OMK9

实用化

实际应用模式@. @% 不是每个信件需要的出版商都确认 .}不是每一个队列都需要复杂的后压处理QZMK2}一个每小时处理 &10}消息的队列 可能不需要和每秒处理一次的#10,000}一样的恢复力工程

询问您自己@: @ @ " @ 什么'} 如果此信件丢失或被延迟是否实际成本@ MS K3 @ 如果答案是 @ I" @ non much @ MPK5 @ don' @ tover @- @ engineerQ. @ 如果回答是“ " @ 重要财务或数据完整性影响” @ OMK10_ 投资适当的回压处理 @ MIK11 @

规模智能

当队列支持“,”时, 答案总是“ '” 。 “ MS K2” 不是总能添加更多的制片人@ MSC3 ” 这“ I'” 就像试图通过增加汽车来修补交通阻塞 @ . ”

考虑一下:

• 以消费者为首 - @% 您可以增加更多工人来处理积压的工作吗? @?@%
• 检查瓶颈 - 是一个缓慢的下游依赖关系 导致备份?
• 尽可能批次 @- 消费者可以同时处理多个消息@?

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

监测和警报

使用 RabitMQ 管理插件 或 HTTP API 设置用于 :的警报

• 队列深度超过阈值
• 消费者的时滞增加
• 出版商确认时出时间
• 连接屏蔽事件

你想知道后压的情况 之前 而不是当你的系统... ...'...

结论结论

后压是'}你所建立的系统 在压力下保持抗御力的系统=□.}*

关键洞察力:

1. 后压是反馈 *- 生产者和消费者合作寻找可持续的投入
2. 监视队列深度 - 您可以控制您能管理的东西 @ MS K2 t量度
3. 使用出版商确认 {\fn黑体\fs22\bord1\shad0\3aHBE\4aH00\fscx67\fscy66\2cHFFFFFF\3cH808080}我知道经纪人什么时候在挣扎
4. 执行指数反转 @ -_ don'_ t敲一个已经承受压力的系统@'%s
5. 不仅是生产商, # 修复瓶颈 # # M-# # 不是症状 #

当您的系统说“"I'm full,给我一分钟”时, 正确的回答是:“"没有问题吗?,I'll 等待'."#那"'有良好分配的系统的精髓-}(MSPK10polite},协作=,和具有弹性的.

当队列增长有点,时保持冷静 记住:}在优雅的后压下,一个系统比这个系统要好得多