c++ - ZMQ 导致主线程卡住(或类似的东西..？)

Question

我有以下一段 C++ 代码，它打开 ZMQ 订阅者套接字并在无限循环中接收消息。

listener.cc:(代码应该可以工作，编译:g++ -lzmq listener.cc)

#include <iostream>
#include <zmq.hpp>

class Listener {
public:
    Listener() {
        std::cout << "constructor call" << std::endl;

        // Not working:
//        zmq::context_t context(1);
//        sck = new zmq::socket_t(context, ZMQ_SUB);
//        sck->connect("tcp://127.0.0.1:9999");
//        sck->setsockopt( ZMQ_SUBSCRIBE, "", 0);

        std::cout << "constructor end" << std::endl;
    }

    void run() {
        // Ok:
        zmq::context_t context(1);
        sck = new zmq::socket_t(context, ZMQ_SUB);
        sck->connect("tcp://127.0.0.1:9999");
        sck->setsockopt(ZMQ_SUBSCRIBE, "", 0);

        while (1) { // Receive messages, not interesting:
            std::cout << "listening..." << std::endl;
            zmq::message_t message;
            sck->recv(&message);
            std::cout << "received something" << std::endl;
        }
    }
    zmq::socket_t *sck;
};

int main(int argc, char *argv[]) {
    Listener listener;
    std::cout << "foo" << std::endl;
    listener.run();
    return 0;
}

到目前为止，代码按预期工作:

$ g++ -lzmq listener.cc
$ ./a.out 
constructor call
constructor end
foo
listening...

不过，我想把zmq-context/socket的初始化移到类的构造函数中(被注释掉的部分)。但是随后代码根本不会从构造函数调用中返回，构造函数中的所有语句都被执行但是 main 的第二行没有被执行并且程序卡住了。输出是:

$ g++ -lzmq listener.cc
$ ./a.out 
constructor call
constructor end

我唯一想到的就是主线程因为某种原因停止执行。谁能解释一下并提供解决方案？

干杯

Answer 1

谁能解释一下并提供解决方案？是的...

ZeroMQ 使用每个 Context( nIOthreads = 1 ) 实例作为引擎盖下的一个非常强大的引擎，必须小心谨慎，这样资源管理过程就不会让人感到意外(因为阻塞/卡住就是这样一种情况)。

如果有一些活跃使用的套接字实例(在 Context() 实例的幕后管理)，可能会出现这样一种情况，即在进入析构函数处理阶段之前并非所有传输都已完成，或者如果有人手动执行类似的步骤，试图尝试 .close() 这样的套接字实例和/或 .term() 上下文实例。

无论有意或无意，人们都会多次陷入这种困境。

ZeroMQ native API 文档在这个主题上非常清楚，并警告了一个风险，即尚未完成的低级事务可能会让代码无限等待永远不会出现的外部(远程代理操作)事件。这种无意识的代码看起来像是卡住/挂起的故障，但只是由于没有意识到这种风险并且没有采取适当的预防措施而导致这种确定性的情况。

虽然较新的 API 版本更改了一些默认设置，但我建议所有用户明确设置安全配置，即使较新的默认设置可能避免手动执行此操作的需要。然而，这种做法有助于提高人们对应该及时考虑哪些类型的碰撞的认识 distributed-system设计实践。

---

解决方案？始终 .setsockopt( ZMQ_LINGER, 0 );

zmq_term() shall block until the following conditions are satisfied:

All sockets open within context have been closed with zmq_close().
For each socket within context, all messages sent by the application with zmq_send() have either been physically transferred to a network peer, or the socket's linger period set with the ZMQ_LINGER socket option has expired.

如上所述，这是每个套接字实例化的经验法则。

class Listener {
   // zmq::context_t aClassLocalCONTEXT;              // MAY GET SET LOCAL CTX BY VALUE
   // zmq::socket_t  aClassLocalSOCKET;               // MAY GET SET LOCAL SCK BY VALUE EITHER
      zmq::socket_t  *sck;

public:
    Listener() {
        std::cout << "constructor call" << std::endl;

     // zmq::context_t context(1);                    // not a best practice here
     // ---------------------------------------------------
     // sck = new zmq::socket_t( aClassLocalCONTEXT, ZMQ_SUB );
        sck = new zmq::socket_t( context, ZMQ_SUB );
        sck->setsockopt( ZMQ_LINGER, 0 );             // ALWAYS, best before .bind()/.connect()
        sck->connect(   "tcp://127.0.0.1:9999" );
        sck->setsockopt( ZMQ_SUBSCRIBE, "", 0 );
     // ----------------------------------------------// IF SETUP BY AN INSTANTIATION CALL INTERFACE
     // aClassLocalSOCKET->setsockopt( ZMQ_LINGER, 0 );
     // aClassLocalSOCKET->connect(    ... );
     // aClassLocalSOCKET->setsockopt( ZMQ_SUBSCRIBE, "", 0 );
     // ---------------------------------------------------
        std::cout << "constructor end" << std::endl;
    }
   ~Listener() {
        sck->close();                                 // A GOOD PRACTICE
     // ----------------------------------------------// IF SETUP BY AN INSTANTIATION CALL INTERFACE
     // aClassLocalSOCKET->close();
    }
    void run() {            
        while (1) {                                   // recv()-messages, not interesting:
            std::cout << "listening..." << std::endl;
            zmq::message_t message;
            sck->recv(&message);
            std::cout << "received something" << std::endl;
            zmq::zmq_msg_close(&message);             // A GOOD PLACE TO DISCARD A NEVER MORE USED RESOURCE
        }
    }
};


int main(int argc, char *argv[]) {
    zmq::context_t context(1);                        // GLOBAL CTX
    Listener listener;
    std::cout << "foo" << std::endl;
    listener.run();
    return 0;
}

高效的资源降低间接成本

资源的智能处理很重要，因为每次实例化和销毁都会在 [TIME] 域和 [SPACE] 域中承担成本(内存分配/删除分配成本，再次及时)而且这些都不便宜。

此外，应遵循 ZeroMQ Zen of Zero - 不要共享任何东西(好吧，有时共享 Context()-实例是一种方式，但是......如果你认真对待 distributed-system 设计，最好阅读一本很棒的书Pieter HINTJENS 的书“Code Connected: Volume 1”，绝对值得花时间和精力)。