ZeroMQ多线程实现请求应答模式报zmq::error_t无效参数错误

报错根因 #

你触发Invalid argument错误以及运行逻辑异常是以下多个问题共同导致的：

• connect方法地址拼接错误：现有拼接逻辑生成的地址格式非法，正确TCP连接地址格式为协议://地址:端口，你的代码多拼接了*:且端口前缺少分隔冒号，最终生成的错误地址为tcp://*:localhost5555，不符合ZeroMQ地址规范，这是报错的直接诱因。
• 套接字调用逻辑混乱：你封装的send、receive方法默认仅操作m_replySocket（REP类型，服务端应答套接字），客户端线程调用这两个方法时完全没有用到你单独定义的m_requestSocket（REQ类型，客户端请求套接字）。而且REQ/REP套接字有严格的收发顺序要求：
• REP套接字必须先接收请求，再发送应答，你客户端线程直接用REP套接字执行发送操作，违反顺序规则
• REQ套接字必须先发送请求，再接收应答，你代码中的REQ套接字连接后完全没有被使用
• ZeroMQ套接字不支持跨线程调用：zmq的套接字不是线程安全的，你在主线程初始化、服务端线程和客户端线程同时操作同一个m_replySocket，不符合ZeroMQ的使用规范。
• memcpy长度硬编码：memcpy (reply.data (), msg.c_str(), 5);写死复制5字节，仅对长度为5的消息有效，更换其他消息内容会出现截断或内存越界问题。
• 线程传参逻辑冗余：你已经用了std::shared_ptr管理实例，创建线程时不需要再套std::ref，直接传值即可。

修正后可运行示例代码 #

#include <zmq.hpp>
#include <iostream>
#include <unistd.h>
#include <thread>
#include <memory>

// 拆分服务端和客户端逻辑，避免套接字混用
class ZmqServer {
public:
    ZmqServer() : m_context(1), m_replySocket(m_context, ZMQ_REP) {
    }

    void Bind(std::string protocol ,int port){
        std::string bind_param = protocol + "://*:" + std::to_string(port);
        m_replySocket.bind(bind_param);
    }

    void run() {
        zmq::message_t request;
        // 等待客户端请求
        m_replySocket.recv(&request);
        std::cout << "服务端收到：" << std::string(static_cast<char*>(request.data()), request.size()) << std::endl;
        // 发送应答
        std::string reply_msg = "World";
        zmq::message_t reply(reply_msg.size());
        memcpy(reply.data(), reply_msg.c_str(), reply_msg.size());
        m_replySocket.send(reply);
    }

private:
    zmq::context_t m_context;
    zmq::socket_t m_replySocket;
};

class ZmqClient {
public:
    ZmqClient() : m_context(1), m_requestSocket(m_context, ZMQ_REQ) {
    }

    void connect(std::string protocol, int port) {
        std::string conn_param = protocol + "://localhost:" + std::to_string(port);
        m_requestSocket.connect(conn_param);
    }

    void run() {
        // 发送请求
        std::string req_msg = "Hello";
        zmq::message_t request(req_msg.size());
        memcpy(request.data(), req_msg.c_str(), req_msg.size());
        m_requestSocket.send(request);
        // 等待应答
        zmq::message_t reply;
        m_requestSocket.recv(&reply);
        std::cout << "客户端收到：" << std::string(static_cast<char*>(reply.data()), reply.size()) << std::endl;
    }
private:
    zmq::context_t m_context;
    zmq::socket_t m_requestSocket;
};

void serverTask()
{
    std::cout << "服务端启动，等待接收请求" << std::endl;
    ZmqServer server;
    server.Bind("tcp", 5555);
    server.run();
}

void clientTask()
{
    sleep(1); // 等服务端先启动完成
    std::cout << "客户端启动，连接5555端口" << std::endl;
    ZmqClient client;
    client.connect("tcp", 5555);
    client.run();
}

int main () {
    std::thread sThread(serverTask);
    std::thread cThread(clientTask);

    cThread.join();
    sThread.join();

    return 0;
}

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