如何解决C++中ZeroMQ发送含数组结构体时服务器仅接收一次即终止问题

Let's break down the issues and fix your code so the server receives all ten structs with arrays properly.

Key Issues & Fixes #

1. Correct Socket Pattern Usage #

ZeroMQ enforces strict sequence rules for certain socket types. If you were using REQ/REP, your client must wait for a reply after each send, and the server must send a reply after each receive. For one-way data transfer (client → server without acknowledgments), PUSH/PULL is simpler and avoids sequence errors.

2. Server Loop Structure #

Your server was likely exiting after one receive instead of looping to handle all ten messages. We'll add a loop around the receive/process logic.

3. Message Size Validation #

Ensure the received message size matches your struct's size to prevent memory corruption, which can cause unexpected exceptions.

4. Proper Exception Handling #

Catch ZeroMQ errors and print detailed messages to debug issues instead of letting the program crash abruptly.

Corrected Code Examples #

structure.h (Fixed-Size Array Struct) #

#ifndef STRUCTURE_H
#define STRUCTURE_H

#include <cstdint>

struct SensorData {
    uint32_t message_id;
    float readings[10]; // Fixed-size array (critical for raw memcpy)
};

#endif

Client Code (Using PUSH Socket) #

#include <zmq.hpp>
#include <iostream>
#include <chrono>
#include <thread>
#include "structure.h"

int main() {
    try {
        zmq::context_t context(1);
        zmq::socket_t socket(context, ZMQ_PUSH);
        socket.connect("tcp://localhost:5555");

        // Send 10 structs
        for (int i = 0; i < 10; ++i) {
            SensorData data;
            data.message_id = i + 1;
            // Fill array with sample data
            for (int j = 0; j < 10; ++j) {
                data.readings[j] = static_cast<float>(i * 10 + j);
            }

            // Package struct into ZeroMQ message
            zmq::message_t msg(sizeof(SensorData));
            memcpy(msg.data(), &data, sizeof(SensorData));

            std::cout << "Sending message " << data.message_id << std::endl;
            socket.send(msg);
        }

        // Give server time to process all messages before exiting
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    } catch (const zmq::error_t& e) {
        std::cerr << "Client ZeroMQ Error: " << e.what() << " (Code: " << e.num() << ")" << std::endl;
        return 1;
    }

    return 0;
}

Server Code (Using PULL Socket) #

#include <zmq.hpp>
#include <iostream>
#include "structure.h"

int main() {
    try {
        zmq::context_t context(1);
        zmq::socket_t socket(context, ZMQ_PULL);
        socket.bind("tcp://*:5555");

        std::cout << "Server waiting for messages..." << std::endl;

        // Receive 10 structs
        for (int i = 0; i < 10; ++i) {
            zmq::message_t msg;
            socket.recv(&msg); // Blocks until message arrives

            // Validate message size to avoid corruption
            if (msg.size() != sizeof(SensorData)) {
                std::cerr << "Warning: Received invalid message size (" << msg.size() 
                          << ") expected " << sizeof(SensorData) << std::endl;
                continue;
            }

            // Convert message data to struct
            SensorData data;
            memcpy(&data, msg.data(), sizeof(SensorData));

            // Print received data
            std::cout << "\nReceived Message " << data.message_id << ":" << std::endl;
            std::cout << "Message ID: " << data.message_id << std::endl;
            std::cout << "Readings: ";
            for (float reading : data.readings) {
                std::cout << reading << " ";
            }
            std::cout << std::endl;
        }

        std::cout << "\nReceived all 10 messages successfully!" << std::endl;
    } catch (const zmq::error_t& e) {
        std::cerr << "\nServer ZeroMQ Error: " << e.what() << " (Code: " << e.num() << ")" << std::endl;
        return 1;
    }

    return 0;
}

Alternative: REQ/REP Pattern (With Acknowledgments) #

If you need to confirm each message was received, use REQ/REP instead:

Client (REQ) #

Add a reply receive after each send:

// After sending the message:
zmq::message_t reply;
socket.recv(&reply);
std::string ack(static_cast<char*>(reply.data()), reply.size());
std::cout << "Received acknowledgment: " << ack << std::endl;

Server (REP) #

Send an acknowledgment after processing each message:

// After printing the struct:
zmq::message_t reply("OK", 2);
socket.send(reply);

Additional Notes #

• Dynamic Arrays: If you need dynamic arrays, avoid raw memcpy. Use serialization libraries like Protocol Buffers or FlatBuffers to handle variable-length data safely.
• Padding: sizeof(SensorData) includes any compiler-generated padding, so raw memcpy works for fixed-size structs.
• Error Codes: Use the error number from zmq::error_t::num() to look up specific ZeroMQ errors (e.g., ETERM means the context was terminated prematurely).

内容的提问来源于stack exchange，提问作者Rakshith