为什么 ZeroMQ PUSH/PULL 可以工作，而 PUB/SUB 不行？

Question

环境：在他们的 Jetson 开发板上使用 NVIDIA 风格的 Ubuntu 18.01 和他们的 TX2i 处理器。 ZMQ 4.3.2，利用 ZMQ 的 cppzmq C++ 包装器。

我有大量使用带有 ZeroMQ 的 google 协议缓冲区运行的代码，而且都是 PUSH/PULL，它工作正常，除了我有一个不是点对点的情况，而是 1： 3.此处正确的解决方案是执行 PUB/SUB，但我无法将消息传递给我的订阅者。

我将代码精简为这个简单的示例。如果我取消注释 #define 语句，订阅者将一无所获。注释（编译为 PUSH/PULL 而不是 PUB/SUB），然后订阅者按预期获得消息。由于sleep_for() 次数过多，我希望订阅者在发布者执行发送之前有足够的时间进行注册。

编辑：

为什么要对订阅者进行尝试/捕获？我很早就遇到了例外，并认为这是因为出版商还没有准备好。这似乎不再是这种情况，所以不是我想的那样。

// Publisher
#include "/usr/local/include/zmq.hpp"
#include "protobuf_namespace.pb.h"
#include <chrono>
#include <thread>


#define PUB_SUB

int main( void )
{
  zmq::context_t* m_pContext = new zmq::context_t( 1 );

#ifdef PUB_SUB
  zmq::socket_t*  m_pSocket  = new zmq::socket_t( *m_pContext, ZMQ_PUB );
#else
  zmq::socket_t*  m_pSocket  = new zmq::socket_t( *m_pContext, ZMQ_PUSH );
#endif

  std::this_thread::sleep_for( std::chrono::seconds( 1 ) );
  //m_pSocket->bind( "tcp://*:53001" );       // using '*' or specific IP doesn't change result
  m_pSocket->bind( "tcp://127.0.0.1:53001" );
  std::this_thread::sleep_for( std::chrono::seconds( 1 ) );

  // Send the parameters
  protobuf_namespace::Params params;
  params.set_calibrationdata( protobuf_namespace::CalDataType::CAL_REQUESTED ); // init one value to non-zero
  std::string        params_str = params.SerializeAsString();
  zmq::message_t     zmsg( params_str.size() );

  memcpy( zmsg.data(), params_str.c_str(), params_str.size() );
  m_pSocket->send( zmsg, zmq::send_flags::none );

  std::this_thread::sleep_for( std::chrono::seconds( 1 ) );
  m_pSocket->close();
  zmq_ctx_destroy( m_pContext );
}

// Subscriber - start me first!
#include "/usr/local/include/zmq.hpp"
#include "protobuf_namespace.pb.h"
#include <chrono>
#include <thread>
#include <stdio.h>

#define PUB_SUB


int main( void )
{
  zmq::context_t* m_pContext = new zmq::context_t( 1 );

#ifdef PUB_SUB
  zmq::socket_t*  m_pSocket  = new zmq::socket_t( *m_pContext, ZMQ_SUB );
  m_pSocket->connect( "tcp://127.0.0.1:53001" );

  int linger = 0;
  zmq_setsockopt( m_pSocket, ZMQ_LINGER, &linger, sizeof( linger ) );
  zmq_setsockopt( m_pSocket, ZMQ_SUBSCRIBE, "", 0 );
#else
  zmq::socket_t*  m_pSocket  = new zmq::socket_t( *m_pContext, ZMQ_PULL );
  m_pSocket->connect( "tcp://127.0.0.1:53001" );
#endif

  protobuf_namespace::Params params;
  zmq::message_t zmsg;
  bool retry = true;

  do {
    try {
      m_pSocket->recv( zmsg, zmq::recv_flags::none );
      retry = false;
      std::this_thread::sleep_for( std::chrono::seconds( 1 ) );
    } catch( ... ) { 
      printf("caught\n");
    }
    std::this_thread::sleep_for( std::chrono::seconds( 1 ) );
  } while( retry );

  std::string param_str( static_cast<char*>( zmsg.data() ), zmsg.size() );
  params.ParseFromString( param_str );

  if( params.calibrationdata() == protobuf_namespace::CalDataType::CAL_REQUESTED )
    printf( "CAL_REQUESTED\n" );
  else
    printf( "bad data\n" );


  std::this_thread::sleep_for( std::chrono::seconds( 1 ) );
  m_pSocket->close();
  zmq_ctx_destroy( m_pContext );
}

Answer 1

_{如果有人从未使用过 ZeroMQ，
在这里可以先看看“ZeroMQ Principles in less than Five Seconds”
在深入了解更多细节之前}

Q：为什么对订阅者进行尝试/捕获？

因为 :
a ) // Subscriber - start me first! 并且在执行 tcp:// 传输类之前让 PUB 侧几乎“永远”处于休眠状态路径设置以接受.bind() 中的任何第一个.connect()，这里之前是一个巨大的小睡...
std::this_thread::sleep_for( std::chrono::seconds( 1 ) ); 和

b) try'd m_pSocket->recv( zmsg, zmq::recv_flags::none ); 必须根据定义抛出异常，因为到目前为止还没有 tcp:// 传输类路径设置（如 PUB-side还没从睡眠中回来）

Q：为什么 ZeroMQ PUSH/PULL 可以工作，但 PUB/SUB 不行？

好吧，如果设计得当，两者都会尊重已发布的 API。

只需删除任何阻塞sleep()-s，阻止SUB-s 加入，使.connect()-s 能够尽快成功。另外，可能会进入.recv()-ops 的非*阻塞形式（重构try/catch），这在distributed-computing 中很常见，以更好地反映预防性设计的基于.poll() 或反应性.recv(..., ZMQ_NOBLOCK ) 的性质事件处理。

---

最后，但同样重要的是：

ZeroMQ v4+（相对于 v2+ 和 pre-v3.? API ）被切换为使用 PUB-side 消息过滤，因此适当考虑订阅管理（定时/错误-处理/弹性）也必须发生。

如有疑问，可以集成使用 ZeroMQ 内置 socket_monitor 工具，扩展 Context()-instance，并跟踪/检查每个事件，内部到Context()-instance，在已发布的 API 事件之下，一直到最低的详细级别。

不要犹豫read and ask more