Что такое Backpressure?

#Проблема

Очередь сама по себе не решает проблему — она лишь откладывает её. Если средняя скорость производства выше средней скорости потребления, любая конечная очередь переполнится, а бесконечная — съест память.

#Стратегии

#Channels: bounded buffer

var channel = Channel.CreateBounded<ImageJob>(new BoundedChannelOptions(capacity: 100)
{
    FullMode = BoundedChannelFullMode.Wait,      // Wait | DropNewest | DropOldest | DropWrite
    SingleReader = false,
    SingleWriter = false
});

// Производитель: WriteAsync «ждёт», пока в канале не освободится место — это и есть backpressure
async Task ProduceAsync(IAsyncEnumerable<ImageJob> source, CancellationToken ct)
{
    await foreach (var job in source.WithCancellation(ct))
        await channel.Writer.WriteAsync(job, ct);
    channel.Writer.Complete();
}

// Потребители: 4 параллельных обработчика
var consumers = Enumerable.Range(0, 4).Select(_ => Task.Run(async () =>
{
    await foreach (var job in channel.Reader.ReadAllAsync(ct))
        await ResizeAsync(job, ct);
}, ct));

#Брокеры сообщений

• RabbitMQ: prefetchCount — сколько неподтверждённых сообщений брокер отдаёт потребителю; больше не пришлёт, пока нет ack. Без ограничения брокер «вывалит» в потребителя всю очередь. Также flow control на уровне соединения при нехватке памяти/диска у брокера (публикация блокируется).
• Kafka: pull-модель — потребитель сам запрашивает данные (max.poll.records), темп задаёт потребитель; брокер хранит журнал на диске.
• Azure Service Bus: MaxConcurrentCalls, PrefetchCount.
• Длина очереди — ключевая метрика: рост очереди = потребители не успевают. Масштабирование по ней (KEDA в Kubernetes).

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata: { name: image-worker }
spec:
  scaleTargetRef: { name: image-worker }
  minReplicaCount: 1
  maxReplicaCount: 20
  triggers:
    - type: rabbitmq
      metadata: { queueName: images, mode: QueueLength, value: "50" }   # ~50 сообщений на под

#HTTP и ASP.NET Core

builder.Services.AddRateLimiter(o =>
{
    o.GlobalLimiter = PartitionedRateLimiter.Create<HttpContext, string>(_ =>
        RateLimitPartition.GetConcurrencyLimiter("global", _ => new ConcurrencyLimiterOptions
        {
            PermitLimit = 200,           // одновременно в обработке
            QueueLimit = 100,            // ещё столько ждут
            QueueProcessingOrder = QueueProcessingOrder.OldestFirst
        }));
    o.RejectionStatusCode = StatusCodes.Status503ServiceUnavailable;   // остальным — быстрый отказ
});

Лучше быстро отказать части запросов, чем принять все и обслужить всех за 60 секунд (когда клиенты уже ушли по таймауту — «работа впустую»).

Другие уровни: TCP (окно получателя), HTTP/2 flow control, Kestrel (MaxRequestBodySize, лимиты соединений), System.IO.Pipelines (PauseWriterThreshold / ResumeWriterThreshold — Kestrel перестаёт читать из сокета, когда приложение не успевает обрабатывать тело).

#Сквозной backpressure

Хорошая система пропускает сигнал перегрузки назад к источнику, а не накапливает работу в промежуточных буферах.

#Вопросы на засыпку

#Итог

Backpressure — обратная связь от медленного потребителя к быстрому производителю. Без неё очереди растут до отказа. Используйте ограниченные буферы (bounded channels, prefetch), pull-модели, ограничение параллельности и быстрые отказы (429/503), следите за длиной очередей и масштабируйте потребителей по ней.