c++ - 如何最好地中断 zeroMQ 轮询方法以进行清理和终止

Question

用 C++ 编写 我有一个线程使用 zmq poll 方法来发现何时有新事件要处理，它工作正常。不过，我想要的是，当没有更多事件发生时，该线程退出，同时很好地清理。

我可以在其中放置一个条件，而不是无限 while 循环，但它需要 REQUEST_TIMEOUT_MS 才能到达那里。所以我的问题是，中断程序退出轮询的最佳方法是什么？

void * Requester::recieve_thread(void *arg) {

    zmq::socket_t * soc = (zmq::socket_t *) arg;

    zmq::pollitem_t items[] = { { *soc, 0, ZMQ_POLLIN, 0 } };

    while (1) {
        zmq::poll(&items[0], 1, REQUEST_TIMEOUT_MS);

        if (items[0].revents & ZMQ_POLLIN) {

            // process the event
        }
    }

    // clean up
}

Answer 1

人们经常提到，您可以只销毁 zmq 上下文，任何共享该上下文的东西都会退出，但这会造成一场噩梦，因为这会删除套接字对象，并且您退出的代码必须尽最大努力避免死雷区指针。

尝试关闭套接字也不起作用，因为它们不是线程安全的，最终会导致崩溃。

回答:最好的方法是按照 ZeroMQ 指南的建议，通过多线程进行任何使用；使用 zmq 套接字而不是线程互斥/锁/等。

Requester::Requester(zmq::context_t* context)
{
    m_context = context;

    // Create a socket that you'll use as the interrupt-event receiver
    // I'm using a random address and an inproc socket (inprocs need to share a context)
    snprintf(m_signalStopAddr, sizeof(m_signalStopAddr) / sizeof(*m_signalStopAddr), "inproc://%lx%x", (unsigned long)this, rand());
    m_signalStop = new zmq::socket_t(m_context, ZMQ_PAIR);
    m_signalStop->bind(m_signalStopAddr);
}

// Your thread-safe poll interrupter
Requester::interrupt()
{
    char dummy;
    zmq::socket_t doSignal(m_context, ZMQ_PAIR);
    doSignal.connect(m_signalStopAddr);
    doSignal.send(&dummy, sizeof(dummy));
}

void * Requester::recieve_thread(void *arg)
{
    zmq::socket_t * soc = (zmq::socket_t *) arg;
    zmq::pollitem_t items[] =
    {
        { *soc, 0, ZMQ_POLLIN, 0 },
        { *m_signalStop, 0, ZMQ_POLLIN, 0 }
    };

    while (1)
    {
        zmq::poll(items, 2, REQUEST_TIMEOUT_MS);
        if (items[1].revents & ZMQ_POLLIN)
        {
            break; // exit
        }

        if (items[0].revents & ZMQ_POLLIN)
        {
            // process the event
        }
    }

    // Cleanup
}

zmq::context_t* m_context;
zmq::socket_t* m_signalStop; // Don't forget to delete this!
char m_signalStopAddr[100];