如何使用ZeroMQ接收多部分消息

How can I receive multipart messages with ZeroMQ?

本文关键字：多部 消息 何使用 ZeroMQ      更新时间：2023-10-16

我无法让ZeroMQ C++包装器接收多部分消息。使用C版本的相同代码运行得很好，但它导致了一个没有任何C++解释的异常。多部分处理代码如下：

int _tmain(int argc, _TCHAR* argv[])
{
    zmq::context_t context(1);
    zmq::socket_t socket(context, ZMQ_REP);
    socket.bind("tcp://*:5555");
    while(true) {
        // the following two lines lead to exception
        zmq::message_t request;
        socket.recv(&request);
        //zmq_msg_t message;
        //zmq_msg_init (&message);
        //zmq_recv (socket, &message, 0);   
    }
    return 0;
}

它非常简单；这个版本不起作用。但是，如果我注释掉while循环中的前两行，并取消注释当前注释的（C版本）代码，它就可以工作了。这是Windows XP sp3、Zeromq 2.1.1和Visual Studio 2010学习版。

如果我发送单部分消息，两个版本都可以正常工作。我做错了什么？

我也是ZMQ的新手，为了理解ZeroMQ中使用REP/REQ的多部分消息传递，我也经历了很多困难。为了理解这一点，我必须浏览多个资源和缝合数据。我认为这个答案将在不久的将来帮助许多寻求者，这就是为什么我在这里分享客户端和服务器代码。我已经测试了这个代码，它运行得非常好。然而，作为一个新手，我可能会错过一些重要的事情。请分享您的宝贵意见。

服务器代码

void
serverMultipartREPREQ()
{
    try
    {
        zmq::context_t context(1);
        zmq::socket_t socket(context, ZMQ_REP);
        socket.bind("tcp://*:5556");
        std::cout << "Listening at port 5556..." << std::endl;
        zmq::message_t reply;
        socket.recv(reply, zmq::recv_flags::none);
        auto rep = std::string(static_cast<char*> (reply.data()), reply.size());
        std::cout << "Received: " << rep << std::endl;
        
        while(1)
        {    
            if (input == "exit")
                break;
            for (int j = 0; j < 3; ++j)
            {
                std::string s("Message no - " + std::to_string(j));
                zmq::message_t message(s.length());
                memcpy(message.data(), s.c_str(), s.length());
                std::cout << "Sending: " << s << std::endl;
                if (j != 2)
                    socket.send(message, zmq::send_flags::sndmore);
                else
                    socket.send(message, zmq::send_flags::none); 
            }
        }
    }
    catch (const zmq::error_t& ze)
    {
        std::cout << "Exception: " << ze.what() << std::endl;
    }
    Sleep(5000);
}

客户端代码

void
clientMultipartREQREP()
{
    try
    {
        zmq::context_t context(1);
        std::cout << "Connecting to socket at 5556" << std::endl;
        zmq::socket_t socket(context, ZMQ_REQ);
        socket.connect("tcp://localhost:5556");
        std::cout << "Connected to socket at 5556" << std::endl;
        std::string msg("Hii this is client...");
        zmq::message_t message(msg.length());
        memcpy(message.data(), msg.c_str(), msg.length());
        socket.send(message, zmq::send_flags::none); // send to server (request message)
        while (true)
        {
            __int64 more = 1;
            if (more)
            {
                zmq::message_t message;
                socket.recv(message, zmq::recv_flags::none);
                auto rep = std::string(static_cast<char*> (message.data()), message.size());
                std::cout << "Reading from client: " << rep << std::endl;
                size_t size = sizeof(__int64);
                socket.getsockopt(ZMQ_RCVMORE, &more, &size); // if msg is not the last one then more = 1 else more = 0
            }
            else
            {
                std::cout << "Done..." << std::endl;
                break;
            }
        }
    }
    catch (const zmq::error_t& ze)
    {
        std::cout << "Exception: " << ze.what() << std::endl;
    }
    Sleep(5000);
}

可能C版本的代码也不起作用，但你没有检查zmq_recv的返回代码，所以你没有注意到它。此外，当接收miltipart消息时，你应该检查是否有更多的消息部分要通过套接字接收，比如：

int64_t more = 0;
size_t more_size = sizeof(more);
socket.getsockopt(ZMQ_RCVMORE, &more, &more_size);
if (more != 0)
{
  //has more parts
}

此外，还可以查看专门为发送和接收ZeroMQ多部分消息而设计的ZmqMessage C++库。

我决定使用C版本的代码。一般来说，所有的例子似乎都是用C表示的。

相关文章：