5

我尽力遵循ZMQ 终止白皮书中的说明，但到目前为止我失败得很惨。我有一个父类，它产生一个侦听器线程（使用win32-pthreads）。

根据白皮书，终止时，我应该设置_stopped标志，删除上下文，这反过来会调用zmq_term()并释放阻塞recv()。相反，我得到的是：

  • 调用delete _zmqContext使应用程序崩溃（可能存在分段错误）
  • 将删除替换为zmq_term(_zmqContext)不会释放阻塞recv()

我正在添加一个部分代码示例，它很长，因为我不确定哪个部分可能很重要。

AsyncZmqListener.hpp：

class AsyncZmqListener
{
public:
    AsyncZmqListener(const std::string uri);
    ~AsyncZmqListener();

    bool Start();
    void Stop();

private:
    static void* _threadEntryFunc(void* _this);     
    void _messageLoop();

private:
    bool _stopped;
    pthread_t _thread;
    zmq::context_t* _zmqContext;
};

AsyncZmqListener.cpp：

AsyncZmqListener::AsyncZmqListener(const std::string uri) : _uri(uri)
{
    _zmqContext = new zmq::context_t(1);
    _stopped = false;
}

void AsyncZmqListener::Start()
{
    int status = pthread_create(&_thread, NULL, _threadEntryFunc, this);
}

void AsyncZmqListener::Stop()
{
    _stopped = true;
    delete _zmqContext;             // <-- Crashes the application. Changing to 'zmq_term(_zmqContext)' does not terminate recv()
    pthread_join(_thread, NULL);    // <-- This waits forever
}

void AsyncZmqListener::_messageLoop()
{        
    zmq::socket_t listener(*_zmqContext, ZMQ_PULL);
    listener.bind(_uri.c_str());

    zmq::message_t message;    
    while(!_stopped)
    {
        listener.recv(&message);    // <-- blocks forever
        process(message);
    }
}

附言

我知道这个相关问题，但没有一个答案与白皮书中描述的干净退出流程完全匹配。如果必须，我将决定投票...

4

1 回答 1

6

ZMQ recv()在其相关上下文终止后解除阻塞

我不知道发生这种情况时会recv()引发ETERM异常。修改后的代码有效：

void AsyncZmqListener::_messageLoop()
{        
    zmq::socket_t listener(*_zmqContext, ZMQ_PULL);
    listener.bind(_uri.c_str());

    zmq::message_t message;    
    while(!_stopped)
    {
        try
        {
            listener.recv(&message);
            process(message);
        }
        catch(const zmq::error_t& ex)
        {
            // recv() throws ETERM when the zmq context is destroyed,
            //  as when AsyncZmqListener::Stop() is called
            if(ex.num() != ETERM)
                throw;
        }
    }
}
于 2013-09-24T09:31:48.393 回答