windows – 为什么ZeroMQ PGM组播接收卡在中间并且没有进一步接

这不是一个好习惯.我敢于指责图书馆维护人员在发布图书馆之前没有彻底测试过PGM / EPGM,或者在应用程序设计得到充分理解,设计稳健,诊断良好之前随时在开发中做不好的工作.性能/延迟测试在实际部署生态系统的现实检查,通常由{localhost |家庭子网|远程网络| remote-host(s)}.

[PUB] – 发送部分需要得到应有的注意：

如果不出意外,文档的这一部分就是警告并响起所有的钟声和响声.吹嘘所有口哨,如果在一些模拟SLOC中进行不充分的资源管理,而对于非阻塞,超快速循环的残酷尝试确实存在应有的谨慎：

?MQ does not guarantee that the socket will accept as many as ZMQ_SNDHWM messages,and the actual limit may be as much as 60-70% lower depending on the flow of messages on the socket.

所以,可能是对的,你的[PUB] -sender丢失了丢失的消息,然后再将这些消息发送到线路上.

下一个警告来自O / S权限：

The pgm transport implementation requires access to raw IP sockets. Additional privileges may be required on some operating systems for this operation. Applications not requiring direct interoperability with other PGM implementations are encouraged to use the epgm transport instead which does not require any special privileges.

接下来是[SUB] -receiver：

一些更多的调整将有助于嗅探[PUB] -sender,类似于下面提出的[SUB] -receiver的内联状态/跟踪工具：

------------------------------------------------
// ZMQCPP-PGM-receive.cpp : Defines the entry point for the console application.
//                          MODs: https://stackoverflow.com/q/44526517/3666197

#include "stdafx.h"
#include <stdio.h>
#include "zmq.hpp"

#include <chrono>                                                       // since C++ 11
typedef std::chrono::high_resolution_clock              nanoCLK;

#define ZMQ_IO_THREAD_POOL_SIZE                         8

#define ZMQ_AFINITY_PLAIN_ROUNDROBIN_UNMANAGED_RISKY    0
#define ZMQ_AFINITY_LO_PRIO_POOL                        0 | 1
#define ZMQ_AFINITY_HI_PRIO_POOL                        0 | 0 | 2
#define ZMQ_AFINITY_MC_EPGM_POOL                        0 | 0 | 0 | 4 | 8 | 0 | 0 | 64 | 128


int main( int argc,char* argv[] ) {

    auto RECV_start = nanoCLK::now();
    auto RECV_ret   = nanoCLK::now();
    auto RECV_last  = nanoCLK::now();
    auto TEST_start = nanoCLK::now();

    try {
           zmq::context_t context( ZMQ_IO_THREAD_POOL_SIZE );           printf ( "Connecting to server..." );
           int            major,minor,patch;
           zmq::version( &major,&minor,&patch );                      printf ( "Using ZeroMQ( %d.%d.%d )",major,patch );

           zmq::socket_t *s1 = new zmq::socket_t( context,ZMQ_SUB );   // Socket to talk to server

           int zmqLinger   =       0,// [  ms]
               zmqAffinity =       0,// [   #]  mapper bitmap-onto-IO-thread-Pool (ref. #define-s above )

               recvBuffer  =       2 * 123456,// [   B]
               recvMaxSize =    9876,// [   B]
               recvHwMark  =  123456,// [   #]  max number of MSGs allowed to be Queued per connected Peer

               recvRate    =   80000 * 10,// [kbps]
               recvTimeout =    3000,// [  ms]  before ret EAGAIN { 0: NO_BLOCK | -1: INF | N: wait [ms] }
               recoverMSEC =      60 * 60      // [  ms]
               ;

           s1->setsockopt ( ZMQ_AFFINITY,&zmqAffinity,sizeof(int) );
           s1->setsockopt ( ZMQ_LINGER,&zmqLinger,sizeof(int) );
           s1->setsockopt ( ZMQ_MAXMSGSIZE,&recvMaxSize,sizeof(int) );
           s1->setsockopt ( ZMQ_RCVBUF,&recvBuffer,sizeof(int) );
           s1->setsockopt ( ZMQ_RCVHWM,&recvHwMark,sizeof(int) );
           s1->setsockopt ( ZMQ_RCVTIMEO,sizeof(int) );
           s1->setsockopt ( ZMQ_RATE,sizeof(int) );
     //    s1->setsockopt ( ZMQ_RECOVERY_IVL,&recoverMSEC,sizeof(int) );

           s1->connect ( "pgm://local_IP;239.255.0.20:30001" );
           s1->setsockopt ( ZMQ_SUBSCRIBE,0 );                   printf ( "done. n" );

           int seq = 0;
           while( true ) {
                  zmq::message_t         msgbuff;                  RECV_start = nanoCLK::now(); RECV_last = RECV_ret;
                  int   ret = s1->recv( &msgbuff,0 );             RECV_ret   = nanoCLK::now();
                  if ( !ret )                                           printf ( "[T0+ %14d [ns]]: [SUB] did not receive any message within set timeout(%d). RC == %d LOOP_ovhd == %6d [ns] RECV_wait == %10d [ns]n",std::chrono::duration_cast<std::chrono::nanoseconds>( RECV_ret - TEST_start ).count(),recvTimeout,ret,std::chrono::duration_cast<std::chrono::nanoseconds>( RECV_ret - RECV_last ).count(),std::chrono::duration_cast<std::chrono::nanoseconds>( RECV_ret - RECV_start ).count() );
                  else                                                  printf ( "[T0+ %14d [ns]]: [SUB] did now receive   a message SEQ#(%6d.) DATA[%6d] B. RC == %d LOOP_ovhd == %6d [ns] RECV_wait == %10d [ns]n",++seq,msgbuff.size(),std::chrono::duration_cast<std::chrono::nanoseconds>( RECV_ret - RECV_start ).count() );
           }
    }
    catch( zmq::error_t   &e ) {                                        printf ( "[T0+ %14d [ns]]: [EXC.ZMQ] An error occurred: %snWill RET(1)",e.what() );
           return 1;
    }
    catch( std::exception &e ) {                                        printf ( "[T0+ %14d [ns]]: [EXC.std] An error occurred: %snWill RET(1)",e.what() );
           return 1;
    }
    return 0;
}