Что такое backpressure и как его реализовать?

Система из нескольких звеньев работает со скоростью самого медленного. Если быстрое звено продолжает толкать данные, излишек должен где-то лежать — в очереди. Неограниченная очередь:

• растёт в памяти, пока процесс не упадёт;
• увеличивает задержку: каждый новый элемент ждёт всех предыдущих, и к моменту обработки результат может быть уже не нужен (клиент ушёл по таймауту — работа впустую);
• прячет проблему до момента, когда она становится аварией.

Способы сказать «помедленнее»:

• ограниченный буфер с блокировкой — put в полную очередь ждёт (BlockingQueue, буферизованный канал Go, CallerRunsPolicy в пуле потоков). Замедление распространяется вверх по цепочке;
• pull вместо push — потребитель сам запрашивает следующую порцию: Kafka consumer делает poll, Reactive Streams — request(n), итераторы и генераторы. Источник физически не может отдать больше, чем просили;
• кредиты или окно — получатель выдаёт разрешение на N элементов (окно TCP, flow control HTTP/2 и gRPC, prefetch в RabbitMQ);
• явный отказ — 429/503, nack, и источник повторяет позже с отсрочкой;
• сброс лишнего — для данных, где потеря допустима: метрики, позиции на карте, кадры видео — оставить только последнее значение.

Где ломается. Backpressure работает, только если все звенья его поддерживают. Типичные разрывы:

• асинхронный HTTP-клиент без ограничения числа одновременных запросов;
• LinkedBlockingQueue без ёмкости в пуле потоков;
• Flux.create или subscribe с неограниченным буфером, onBackpressureBuffer без размера;
• WebSocket-рассылка медленному клиенту — буфер на сервере растёт;
• Kafka как буфер: лаг потребителей растёт, это не авария, но нужен алерт и масштабирование потребителей.

Внешний мир (пользователи) замедлить нельзя — для него на входе ставят rate limiting и load shedding, а внутри системы — backpressure.

// Пул с ограниченной очередью: при заполнении задачу выполняет вызывающий поток,
// и источник естественно замедляется.
var pool = new ThreadPoolExecutor(8, 8, 0, TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(1000), new ThreadPoolExecutor.CallerRunsPolicy());

// Reactor: обрабатываем не больше 16 элементов одновременно, остальные запрашиваются по мере готовности
flux.flatMap(this::callService, 16)