Я быстро протестировал ZeroMQ PUB/SUB, и теперь у меня есть рабочий код. Тем не менее, я немного смущен концепцией высокой отметки, применяемой в zeromq.
Я установил HWM в своем коде издателя, который устанавливает длину очереди для каждого подписчика, подключенного к сокету.
Однако также возможно установить HWM на принимающем сокете абонента. Есть ли причина устанавливать HWM на стороне подписчика и чем это будет отличаться от установки HWM издателя?
В издателе мы почти всегда должны тщательно учитывать HWM, потому что существует множество причин сбоя (нехватки памяти), влияющих на систему в целом (поскольку издатель обслуживает всех подписчиков).
Также в подписчике бывают случаи, когда регулирование HWM может быть полезным, но это зависит главным образом от характера подписчика, того, что он делает с полученным сообщением, и насколько высока вероятность того, что он не мог вовремя обработать большое количество полученных сообщений; и ожидаемой средой выполнения (сколько памяти доступно, количество подписчиков и т.д.)
ZMQ использует концепцию HWM (высокая отметка) для определения пропускной способности внутренних труб. Каждое соединение из сокета или в сокет имеет свой канал и HWM для отправки и/или получения, в зависимости от типа сокета. Некоторые сокеты ( PUB, PUSH ) имеют только буферы отправки. Некоторые ( SUB, PULL, REQ, REP ) имеют только буферы приема. Некоторые ( DEALER, ROUTER, PAIR ) имеют буферы отправки и приема.
Доступные параметры сокета:
ZMQ_SNDHWM: установить верхний порог для исходящих сообщений (... на сокет издателя )ZMQ_RCVHWM: установить верхний порог для входящих сообщений (... на абонентский сокет )ZMQ 3.0+ накладывает ограничения по умолчанию на свои внутренние буферы (так называемые HWM), потому что HWM — это отличный способ уменьшить проблемы с переполнением памяти.
И ZMQ_PUB, и ZMQ_SUB имеют параметр ZMQ_HWM, для которого установлено значение «Отбросить», поэтому, когда лимиты увеличиваются, память подписчика или издателя должна перестать расти, по крайней мере, на то, что зависит от буферов ZMQ.
Обычно больше всего в защите от неразборчивого использования памяти (проблем нехватки памяти) нуждаются издатели:
Во внутрипроцессном транспорте отправитель и получатель совместно используют одни и те же буферы, поэтому реальный HWM представляет собой сумму HWM, установленных обеими сторонами.
Но если вы используете TCP и подписчик работает медленно, сообщения будут стоять в очереди на издателе.
Общие причины отказа PUB-SUB включают в себя:
из-за постановки сообщений в очередь на издателе у издателей заканчивается память и происходит сбой, особенно если есть много подписчиков и невозможно сбросить на диск из соображений производительности.
С точки зрения издателя, отличная стратегия, которую мы используем при правильной настройке ZWM, заключается в прекращении очереди новых сообщений через некоторое время, новые сообщения просто отклоняются или удаляются; это то, что делает ØMQ, когда издатель устанавливает HWM.
ZMQ также может ставить сообщения в очередь на подписчике.
Если у кого-то закончится память и произойдет сбой, то это будет подписчик, а не издатель, что справедливо. Это идеально подходит для «пиковых» потоков, когда подписчик какое-то время не успевает, но может наверстать упущенное, когда поток замедляется.
Примечание: HWM неточны; в то время как вы можете получить до 1000 сообщений по умолчанию, реальный размер буфера может быть намного меньше (всего половина) из-за того, как libzmq реализует свои очереди.
Основным источником этих предположений является книга Питера Хинтдженса Code Connected Volume 1, доступная онлайн в электронном формате; в нем есть глава, посвященная максимальным значениям и содержащая дополнительные пояснения по этой теме.
Было бы полезно установить ZMQ_SNDHWM на подписчике, если вы подписаны сразу более чем на 1000 тем (насколько мне известно, по умолчанию ZMQ_SNDHWM равно 1000). Итак, если вы подпишитесь следующим образом:
for(i = 0; i < 2000; i++)
{
subscriber.setsockopt( ZMQ_SUBSCRIBE, &i, sizeof(i));
}
Это не сработает из-за переполнения очереди вывода.
Но если вы установите перед подключением
subscriber.setsockopt( ZMQ_RCVHWM, &i, sizeof(i));
subscriber.setsockopt( ZMQ_SNDHWM, &i, sizeof(i));
Это работает хорошо.
