c - 使用 ZeroMQ 和 ProtocolBuffers 发送结构

Question

我正在编写一个程序，该程序应该通过 ZeroMQ 发送 C 结构。因此，我使用 Google 的 ProtocolBuffers 来序列化结构。

我现在遇到的问题是我的订户端没有收到任何内容。发布者打印出“消息已成功发送”，因此我认为错误发生在订阅者端。

发布者:

int main (void)
{
    Message protomsg = MESSAGE__INIT;
        void *buf;
    unsigned len;

        void *context = zmq_ctx_new();
        void *subscriber = zmq_socket(context, ZMQ_PUB);
    zmq_bind(subscriber, "ipc://my.sock");

        //Initialising protomsg (not so important)

        //sending message

        len = message__get_packed_size(&protomsg);
    buf = malloc(len);
    message__pack(&protomsg, buf);

    zmq_msg_t output;
    zmq_msg_init_size(&output, len);
    zmq_msg_init_data(&output, buf, len, NULL, NULL);
    if(zmq_msg_send(&output, subscriber, 0) == -1)
        perror("Error sending message \n");
    else
        printf("Message successfully sent \n");
    zmq_msg_close(&output);
        free(buf);

        zmq_close (subscriber);
        zmq_ctx_destroy (context);
        return 0;
}

订阅者:

    int main (void){
        Message *protomsg;
            void *context = zmq_ctx_new ();
            void *publisher = zmq_socket (context, ZMQ_SUB);
            zmq_connect(publisher, "ipc://my.sock");
            zmq_setsockopt(publisher, ZMQ_SUBSCRIBE, "", 0);

            // Read packed message from ZMQ.
        zmq_msg_t msg;
        zmq_msg_init(&msg);
        if(zmq_msg_recv(&msg, publisher, 0) == -1)
            perror("Error receiving message \n");
        else
            printf("Message received");
        memcpy((void *)protomsg, zmq_msg_data(&msg), zmq_msg_size(&msg));

            // Unpack the message using protobuf-c.
        protomsg = message__unpack(NULL, zmq_msg_size(&msg), (void *)&data);   
        if (protomsg == NULL)
        {
            fprintf(stderr, "error unpacking incoming message\n");
            exit(1);
        }

            printf("Address: %u, Type: %u, Information[0]: %u, Information[1]: %u \n", protomsg->address-48, protomsg->frametype, protomsg->information[0], protomsg->information[1]);
        zmq_msg_close (&msg);

        // Free the unpacked message
        message__free_unpacked(protomsg, NULL);

            //close context,socket..
}

Answer 1

不知道是否有人仍然关心这个，但是这里...我同意@Steve-o的观点，这是一个时间问题，尽管我认为问题是你太早关闭了发布者套接字。

您的发布者代码发布消息，然后立即关闭套接字并终止上下文。因此，消息在发布者中存在几毫秒，然后就永远消失了。

如果您先运行发布者，它会执行该操作，然后退出并且消息消失。当您启动订阅者时，它会尝试连接到不再存在的 IPC 套接字。 ZeroMQ 允许这样做，并且订阅者将阻塞，直到有 IPC 套接字可以连接。

我还没有查看 ZeroMQ IPC 源代码，但我怀疑，在幕后，订阅者会定期尝试连接到发布者套接字。现在，如果您再次运行发布者，它可能会起作用，但您会遇到严重的竞争条件。如果您在 ZeroMQ 工作线程尝试重试的同时启动发布器，则连接可能会发生，您甚至可能在发布器销毁所有内容之前收到消息。

我很确定这个问题与结构和 protobuf 无关。从 ZeroMQ 的角度来看，您只是发送字节。没有区别。如果 ZeroMQ 字符串的测试用例与 ZeroMQ 结构的测试用例确实相同 - 那么代码更改可能会添加或删除几纳秒，从而能够以错误的方式打破竞争条件。

具体建议:

• 将发布者中的套接字重命名为“发布者”而不是订阅者(复制/粘贴错误)
• 在 zmq_close 之前添加 30 秒的 sleep (发布者)；
• 希望这能解决您的测试代码的问题
• 如果这不能解决问题，请考虑切换到 TCP 传输并使用 wireshark诊断到底发生了什么。