如何用ZeroMQ发送含变长字符串的自定义C结构体？

直接拆分多次发送确实不够高效，还容易出现部分消息丢失、日志不完整的问题。正确的做法是把结构体的所有数据（包括变长字符串的实际内容）序列化成一段连续的字节流，一次性发送，接收端再按相同格式反序列化重建结构体。下面是具体实现思路和代码示例：

核心思路 #

你的结构体里的char*是内存指针，跨进程/机器完全无效，必须把字符串的实际内容打包进字节流。我们需要：

• 先计算所需的总缓冲区大小：每个字符串的长度（建议带\0终止符，方便接收端直接使用） + 所有数值类型（unsigned int）的字节数
• 将数值、字符串长度、字符串内容依次拷贝到连续缓冲区
• 用ZeroMQ发送这个缓冲区，接收端按相同顺序解析数据

代码示例（简化版） #

假设你的结构体简化为：

struct msg {
    char *item1;
    unsigned int amount1;
    char *name1;
    unsigned int cost1;
};

1. 打包（序列化）函数 #

// 计算所需缓冲区总大小
size_t calc_msg_size(const struct msg *m) {
    size_t size = 0;
    // 每个字符串先存长度（unsigned int），再存带'\0'的内容
    size += sizeof(unsigned int) + strlen(m->item1) + 1;
    size += sizeof(unsigned int); // amount1的大小
    size += sizeof(unsigned int) + strlen(m->name1) + 1;
    size += sizeof(unsigned int); // cost1的大小
    return size;
}

// 把结构体打包成连续字节流
char* pack_msg(const struct msg *m, size_t *out_size) {
    *out_size = calc_msg_size(m);
    char *buf = malloc(*out_size);
    if (!buf) return NULL;

    char *ptr = buf;
    unsigned int len;

    // 打包item1：先存长度，再存内容
    len = strlen(m->item1) + 1;
    memcpy(ptr, &len, sizeof(len));
    ptr += sizeof(len);
    memcpy(ptr, m->item1, len);
    ptr += len;

    // 打包amount1（跨平台需转网络字节序）
    len = htonl(m->amount1);
    memcpy(ptr, &len, sizeof(len));
    ptr += sizeof(len);

    // 打包name1：先存长度，再存内容
    len = strlen(m->name1) + 1;
    memcpy(ptr, &len, sizeof(len));
    ptr += sizeof(len);
    memcpy(ptr, m->name1, len);
    ptr += len;

    // 打包cost1（跨平台转网络字节序）
    len = htonl(m->cost1);
    memcpy(ptr, &len, sizeof(len));
    ptr += sizeof(len);

    return buf;
}

2. 发送代码 #

// 假设已创建并完成绑定/连接的ZeroMQ套接字（如ZMQ_PUB、ZMQ_REQ）
struct msg log_msg = {
    .item1 = "apple",
    .amount1 = 5,
    .name1 = "storeA",
    .cost1 = 10
};

size_t buf_size;
char *buf = pack_msg(&log_msg, &buf_size);
if (!buf) {
    perror("pack_msg failed");
    return -1;
}

// 发送字节流
int rc = zmq_send(socket, buf, buf_size, 0);
if (rc == -1) {
    perror("zmq_send failed");
}

free(buf); // 发送完成后释放缓冲区

3. 接收端解包（反序列化）函数 #

struct msg* unpack_msg(const char *buf, size_t buf_size) {
    struct msg *m = malloc(sizeof(struct msg));
    if (!m) return NULL;

    const char *ptr = buf;
    unsigned int len;

    // 解析item1
    memcpy(&len, ptr, sizeof(len));
    ptr += sizeof(len);
    m->item1 = malloc(len);
    memcpy(m->item1, ptr, len);
    ptr += len;

    // 解析amount1（转回主机字节序）
    memcpy(&len, ptr, sizeof(len));
    m->amount1 = ntohl(len);
    ptr += sizeof(len);

    // 解析name1
    memcpy(&len, ptr, sizeof(len));
    ptr += sizeof(len);
    m->name1 = malloc(len);
    memcpy(m->name1, ptr, len);
    ptr += len;

    // 解析cost1（转回主机字节序）
    memcpy(&len, ptr, sizeof(len));
    m->cost1 = ntohl(len);
    ptr += sizeof(len);

    return m;
}

// 接收代码示例
char recv_buf[4096]; // 大日志建议用zmq_msg_t动态分配缓冲区
int rc = zmq_recv(socket, recv_buf, sizeof(recv_buf), 0);
if (rc == -1) {
    perror("zmq_recv failed");
    return -1;
}

struct msg *received_msg = unpack_msg(recv_buf, rc);
// 使用received_msg处理日志...

// 用完记得释放内存
free(received_msg->item1);
free(received_msg->name1);
free(received_msg);

关键注意事项 #

• 字节序兼容：如果发送和接收端架构不同（如x86和ARM），必须用htonl/ntohl转换unsigned int的字节序，否则数值会出现错误。
• 内存管理：打包时分配的缓冲区、解包时分配的字符串都要手动释放，避免内存泄漏。
• 动态缓冲区：如果日志内容可能很大，建议用zmq_msg_t（zmq_msg_init_size）动态管理缓冲区，替代固定大小数组。
• 错误检查：实际项目中要检查每个malloc、memcpy、ZeroMQ调用的返回值，避免崩溃。

为什么比拆分发送好 #

• 原子性：单个消息发送是原子操作，不会出现部分日志发送成功的情况，保证日志完整性。
• 性能：减少ZeroMQ系统调用次数，降低通信开销。
• 可维护性：序列化逻辑集中，后续修改结构体只需调整打包/解包函数，不用改动发送接收逻辑。

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