Обратное давление

Подписка читает свой сокет не более чем на read_ahead сообщений впереди обработчика; дальше она перестаёт читать, и ожидание доходит до отправки узла.

Схема текстом

1. Узел отправка PUSH или PUB → Буферы сокетов ядра
2. Буферы сокетов ядра → Задача сокета читает одно сообщение
3. Задача сокета читает одно сообщение → Канал глубиной read_ahead
4. Канал глубиной read_ahead → Обработчик или воркеры workers(n)
5. Канал глубиной read_ahead → Обработчик отстаёт: канал заполнен
6. Обработчик отстаёт: канал заполнен → Задача ждёт и перестаёт читать
7. Задача ждёт и перестаёт читать → Буферы заполняются, окно TCP закрывается
8. Буферы заполняются, окно TCP закрывается → Отправка узла ждёт места
9. PUB ждёт самого медленного подходящего подписчика; сообщение без совпавшего фильтра отбрасывается

Задача подписки читает из сокета по одному сообщению и кладёт его в канал. Канал вмещает read_ahead сообщений, по умолчанию 1000: столько же libzmq задаёт сокету как предельный размер очереди приёма. Настройку хранит брокер, поэтому каждая подписка этого брокера читает вперёд на столько же сообщений.

Когда обработчик отстаёт, канал заполняется, и задача ждёт свободного места. Ожидающая задача ничего не читает, поэтому кадры остаются в буферах ядра, пока те не заполнятся. Тогда TCP останавливает отправителя, и отправка узла ждёт, пока обработчик не догонит. Медленный обработчик замедляет отправителя, а память этого процесса остаётся ограниченной.

В PUB/SUB ждёт отправитель - сокет PUB. Он пишет каждому подписчику с совпавшим фильтром по очереди, поэтому один медленный подписчик задерживает сообщения для всех. Сообщение, которому не подошёл ни один фильтр подписчиков, по-прежнему отбрасывается, как объясняет страница Очереди.

Обработчик берёт по одному сообщению, если точка монтирования не просит больше. workers(n) запускает n обработчиков параллельно, и свободный воркер берёт из канала следующее сообщение. У сообщений ZeroMQ нет ключа, поэтому workers(n, by_key) нечего хешировать, и сообщения достаются воркерам по кругу. Пакет закрывается при размере, заданном в точке монтирования, или через 20 миллисекунд после первого сообщения.

Публикация ждёт так же, когда узел читает медленно. Публикацию в очередь можно отменить, пока она ждёт, и следующая публикация завершит её отправку. Широковещательную публикацию, ответ и запрос отменить нельзя, поэтому ограничивайте их таймаутом.