ZeroMQ Публикация и подписка одновременно

ZeroMQ : способен хорошо справиться с этой задачей в масштабах, указанных выше
nanomsg : способен также выполнять эту задачу, необходимость перепроверить порты / привязки для клиентов

Обзор дизайна:

• клиент случаи не являются постоянными, могут свободно появляться самостоятельно, могут свободно исчезать сами по себе или по ошибке
• клиент Экземпляр решает сам, что он собирается PUB-Lish как сообщение полезной нагрузки
• клиент Экземпляр решает сам, что он собирается SUB-подписаться как актуальный входящий поток сообщений TOPIC-фильтр
• клиент экземпляр обменивается (отправляет), сам по себе, простой, не состоящий из нескольких частей, JSON-форматированные сообщения, которые он подготовил / произвел
• клиент Экземпляр собирает (получает) сообщения, для которых предполагается, что они находятся в том же, не многочастном, JSON-форматированная форма и для которой попытка получить их локально обрабатывается будет иметь место после завершения приема
• максимальное количество клиент-экземпляры не превышают небольшое количество сотен
• максимальный размер любого JSON-formatted полезная нагрузка меньше чем 32 kB, Об 4 kB в среднем
• максимальная задержка, приемлемая для межпроцессной доставки E2E через общую область коллизий LAN, ниже 5,000 [usec]
• сервер Экземпляр является центральной ролью и постоянной сущностью
• сервер экземпляр обеспечивает известный транспортный класс URLцель для всех просроченных столяры.connect()-s

Предложение:

сервер может использовать несколько вариантов поведения для достижения поставленных целей, используя как PUB а также SUB поведения, и обеспечивает код, управляемый, быстрый, SUB-блокированная, неблокирующая петля событий .poll() с выровненной повторной передачей любого из его SUB-боковая сторона .recv()полезные нагрузки к нему PUB-сторона, в настоящее время .connect()аудитория (в прямом эфире клиент экземпляры):

задавать s_SUB_recv = aZmqCONTEXT.socket( zmq.SUB );
а также s_PUB_send = aZmqCONTEXT.socket( zmq.PUB );

по причинам производительности, которые здесь не так сложны, можно также разделить обработку потоков рабочих нагрузок, сопоставив каждый из них с несвязанными подмножествами нескольких созданных потоков ввода-вывода:

карта s_SUB_recv.setsockopt( ZMQ_AFFINITY, 0 );
а также s_PUB_send.setsockopt( ZMQ_AFFINITY, 1 );

задавать s_PUB_send.bind( "tcp://localhost:8899" );
+
задавать s_SUB_recv.setsockopt( ZMQ_SUBSCRIBE, "" ); // forever *every*-TOPIC
задавать s_SUB_recv.setsockopt( ZMQ_MAXMSGSIZE, 32000 ); // protective ceiling
задавать s_SUB_recv.setsockopt( ZMQ_CONFLATE, True ); // retain just the last msg
задавать s_SUB_recv.setsockopt( ZMQ_LINGER, 0 ); // avoid blocking
задавать s_SUB_recv.setsockopt( ZMQ_TOS, anAppToS_NETWORK_PRIO_CODE );

а также s_SUB_recv.bind( "tcp://localhost:8888" ); // [PUB]s .connect()

Так же,
клиент Экземпляр может развернуть обратный тандем как PUBконечная точка и SUB-конечный, готов к .connect() известной транспортной цели-URL,

клиент конкретная подписка локально решает, что фильтровать из входящего потока сообщений (до ZeroMQ v.3.1 API множество сообщений будет доставлено каждому клиент экземпляр над классом транспорта, однако, так как API v.3.1+, тема-фильтр работает на PUBсторона, которая в желаемом modus-operandi устраняет потерянные объемы данных по сети, но в то же время это увеличивает PUBнакладные расходы на обработку ( исх .: Замечания по принципу увеличенного отображения / увеличения потоков мульти-ввода-вывода выше)

задавать c_SUB_recv = aZmqCONTEXT.socket( zmq.SUB );
а также c_PUB_send = aZmqCONTEXT.socket( zmq.PUB );

до тех пор, пока накладные расходы на сборку / обработку полезной нагрузки не приблизятся к допустимому пороговому значению сквозной задержки, не будет необходимости разделять / отделять ZeroMQ потоки ввода / вывода низкого уровня здесь:
карта c_SUB_recv.setsockopt( ZMQ_AFFINITY, 0 );
а также c_PUB_send.setsockopt( ZMQ_AFFINITY, 1 );

задавать c_PUB_send.connect( "tcp://server:8888" ); // reverse .bind on [SUB]
+
задавать c_SUB_recv.setsockopt( ZMQ_SUBSCRIBE, "" ); // modified on-the-fly
задавать c_SUB_recv.setsockopt( ZMQ_MAXMSGSIZE, 32000 ); // protective ceiling
задавать c_SUB_recv.setsockopt( ZMQ_CONFLATE, True ); // take just last
задавать c_SUB_recv.setsockopt( ZMQ_LINGER, 0 ); // avoid blocking
задавать c_SUB_recv.setsockopt( ZMQ_TOS, anAppToS_NETWORK_PRIO_CODE );
а также c_SUB_recv.connect( "tcp://server:8899" );

Обсуждение:

Для хобби-проектов больше нет необходимости в инфраструктуре обмена сообщениями, тем не менее для более серьезных доменов существуют дополнительные сервер а также клиент экземпляры должны иметь некоторые дополнительные формы поведения формальной коммуникации.
— r/KBD для удаленной клавиатуры, с CLIподобные специальные инспекционные утилиты
— KEEP_ALIVE транспондеры для общесистемного мониторинга состояния / перфоманса
— SIG_EXIT обработчики для разрешения всей системы / конкретного экземпляра SIG_EXITs
— distributed syslog сервис, позволяющий безопасно собирать / хранить неблокирующую реплику записей журнала (будь то на этапе отладки или на этапе настройки производительности или сбора доказательств производственного уровня)

— Identity Management инструменты для аудита-следы и др

— WhiteList/BlackList для повышения надежности инфраструктуры, чтобы сделать ее более защищенной от DoS-атак / отравления ошибочных пакетов трафика NIC и др.

— Adaptive Node re-Discovery для более разумного / специального проектирования инфраструктуры и мониторинга состояния или когда на сцену выходят сценарии с активным горячим резервированием и передачей ролей с активным горячим резервированием с много ролевой / (N + M) и т. д.

Резюме

A1: Да, полностью в рамках возможностей ZeroMQ
A2: Да, примеры кода C ++ в книге ZeroMQ / Руководства доступны
A3: Ref .: A1, плюс может понравиться в посте Мартина СУСТРИКА на тему «Различия между nanomsg а также ZeroMQ«

Надеюсь, вам понравятся возможности распределенной обработки, будь то при поддержке ZeroMQ или же nanomsg или оба.

Только собственное воображение — это предел.

Если интересно Более подробная информация, кому-то может понравиться книга, упомянутая в Лучший следующий шаг раздел этого поста