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

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

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

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

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

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

установите s_SUB_recv = aZmqCONTEXT.socket( zmq.SUB );
и s_PUB_send = aZmqCONTEXT.socket( zmq.PUB );

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

map 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()


Аналогично ,
Экземпляр client может развернуть обращенный назад тандем из PUB-конечной точки и SUB-конечной точки, готовый .connect() к действию nown transport-target-URL.

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

установить c_SUB_recv = aZmqCONTEXT.socket( zmq.SUB );
и c_PUB_send = aZmqCONTEXT.socket( zmq.PUB );

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

set c_PUB_send.connect( "tcp://server:8888" ); // reverse .bind on [SUB]
+
set c_SUB_recv.setsockopt( ZMQ_SUBSCRIBE, "" ); // modified on-the-fly
set c_SUB_recv.setsockopt( ZMQ_MAXMSGSIZE, 32000 ); // protective ceiling
set c_SUB_recv.setsockopt( ZMQ_CONFLATE, True ); // take just last< бр> се t 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" );