c++ - ZeroMQ 是否有数据到达时的通知/回调事件/消息？

Question

我正在尝试将 ZMQ 集成到现有的严重依赖 MFC 套接字 (CASyncSocket) 的 Windows 应用程序中。

我有一个 CWinThread 派生的 UI 线程(没有 GUI)，它使用 CAsyncSocket 与服务器异步通信。我想添加一个 ZMQ inproc 通信线路来处理将从服务器接收到的数据(基于 REQ/REP)传送到应用程序中的其他线程。

使用 CAsyncSocket，只要套接字上有新数据可供接收，MFC 框架就会调用 OnReceive 方法(对于那里的核心 MFC 专家来说，这可能是一种过度简化)。

ZMQ有没有这样的机制？或者我是否必须添加一个额外的专用 WorkerThread，UI 线程启动它来处理我与应用程序其余部分的 ZMQ 通信？两条管道上的流量都很小，所以我真的不想创建 2 个单独的线程(如果我可以创建 1 个线程的话)。

请注意，我已经掌握了基础知识，只是在同步方面遇到了问题。如果我使用 ZMQ 的阻塞 recv/send，它会耗尽我的 CAsycSocket，因为 Windows 消息永远不会被线程处理，导致有时永远不会从 ZMQ 应该传递的服务器获取数据。但是如果我使用非阻塞 ZMQ 调用，那么线程经常会闲置，因为它不知道要读取 ZMQ 套接字。

Answer 1

最终，答案是否。当数据到达您可以链接到的 ZeroMQ 时，当前没有回调/通知。我也找不到任何添加此功能的分支。

在单个线程中使用 MFC 套接字框架提供的传统 OnReceive 调用并添加第二个线程专用于 ZMQ 时，我无法让 ZMQ 工作，这破坏了使用它的全部目的(它用于线程同步).

我的工作实现最终放弃了 MFC 套接字，并为我的 inproc 服务器(用于与其他线程通信)以及我的 TCP(非 ZMQ)服务器连接使用 ZMQ，并使用阻塞轮询调用(zmq_poll())在 OnIdle() 方法中(每次返回 1 以创建一个繁忙的循环)。阻塞轮询

BOOL CMyThreaClass::OnIdle(LONG lCount)
{
    UNREFERENCED_PARAMETER(lCount);

    zmq_pollitem_t items [] = {
    { m_pZMQInprocServer, 0, ZMQ_POLLIN, 0 },
    { m_pZMQTCPSocket, 0, ZMQ_POLLIN, 0 }
    };
    const int iZMQInfiniteTimeout(-1);
    iResult = zmq_poll(&items[0], sizeof(items) / sizeof(items[0]), iZMQInfiniteTimeout);
    TRACE("zmq_poll result: %d\n", iResult);

    if (items[0].revents & ZMQ_POLLIN)
    {
        sMyStruct sMessage;
        iResult = zmq_recv(m_pZMQInprocServer, &sMessage, sizeof(sMessage), ZMQ_DONTWAIT); // don't block (the zmq_poll blocks for us)
        TRACE("inproc recv result: %d\n", iResult);
        //  Process inproc messages
        iResult = zmq_send(pZMQInprocServer, &sMessage, sizeof(sMessage), ZMQ_NULL); // block
        TRACE("inproc send result: %d\n", iResult);
    }
    if (items[1].revents & ZMQ_POLLIN)
    {
        // there will be an ZMQ_IDENTITY identifier on the beginning of the socket buffer, read it off first
        uint8_t id [256];
        size_t id_size = 256;
        iResult = zmq_getsockopt(m_pZMQTCPSocket, ZMQ_IDENTITY, id, &id_size);
        TRACE("getsockopt poll result %d:id %d\n", iResult, id);
        iResult = zmq_recv(m_pZMQTCPSocket, &id, id_Size, ZMQ_DONTWAIT); // don't block
        // now get our actual data
        char szBuffer[1024];
        int iBytesReceived = zmq_recv(m_pZMQSocket, szBuffer, sizeof(szBuffer), ZMQ_DONTWAIT);
        if (iBytesReceived > 0)
        {
            // process TCP data
        }
    }
}

注意:此答案需要使用 ZMQ 4 或更高版本，因为早期版本的 ZMQ 不会与常规 TCP 套接字连接进行通信。