linux - 使用 ZeroMQ 流连接到设备失败，但 Linux 套接字工作

Question

我有一个串行设备正在通过串行服务器转换为 TCP/IP。这似乎可以正常工作，因为其他应用程序可以通过这种方式连接。

在我的应用程序中，我尝试使用 ZeroMQ ZMQ_STREAM 向设备发送信息请求，并等待回复。

// from it's own thread: threadTx
zmq::socket_t soc_tx(zmq_ctx, ZMQ_STREAM);
soc_tx.connect("tcp://10.10.10.100:4003");

std::string msg("hello");
zmq::message_t zmsg(msg.c_str(), msg.size());
soc_tx.send(zmsg);

然后从另一个线程:

// from another thread: threadRx
zmq::socket_t soc_rx(zmq_ctx, ZMQ_STREAM);
soc_rx.connect("tcp://10.10.10.100:4003");

zmq::message_t zrecv;
soc_rx.recv(&zrecv);

send 成功返回，我确实收到了一条 0 长度的消息，根据文档表明连接成功(或断开连接)。

但是，我从未收到对原始发送请求的回复。 soc_rx.recv() 将无限等待。 Wireshark 指示正在发送回复，只是 ZeroMQ 主机从未收到回复。

我也尝试在我打开发送的同一个套接字上接收，但没有成功。

// from it's own thread: threadTx
zmq::socket_t soc(zmq_ctx, ZMQ_STREAM);
soc.connect("tcp://10.10.10.100:4003");

std::string msg("hello");
zmq::message_t zmsg(msg.c_str(), msg.size());
soc.send(zmsg);

while (true)
{
  zmq:message_t zrecv;
  soc.recv(&zrecv);
}

但我仍然只收到 0 长度的连接消息，之后没有其他消息。

我认为这可能是配置问题，但同样，其他应用程序也能正常工作。我还能够使用 native Linux 套接字成功连接并从设备发送和接收消息。

uint16_t const PORT = 4003;
struct sockaddr_in address;
int sock = 0, valread;
struct sockaddr_in serv_addr;
char buffer[1024] = {0};
if ((sock = socket(AF_INET, SOCK_STREAM, 0)) < 0)
{
  return -1;
}

memset(&serv_addr, '0', sizeof(serv_addr));

serv_addr.sin_family = AF_INET;
serv_addr.sin_port = htons(PORT);

// Convert IPv4 and IPv6 address from text to binary form
if(inet_pton(AF_INET, "10.10.10.100", &serv_addr.sin_addr) <= 0)
{
  std::cout << "> invalid address. address not supported" << std::endl;
  return nullptr;
}

if (connect(sock, (struct sockaddr *)&serv_addr, sizeof(serv_addr)) < 0)
{
  std::cout << "> connection failed" << std::endl;
}

send(sock, out_buf, out_buf_size, 0);

std::this_thread::sleep_for(std::chrono::milliseconds(1000));

valread = recv(sock, buffer, 1024, 0);

这很好用。

为了与我系统中的其他应用程序保持一致，我喜欢将 ZeroMQ 用于所有套接字通信，但是是否存在阻止这种情况的限制？

Answer 1

使用 ZMQ_STREAM 有点棘手，但您应该能够使用单个套接字与外部 TCP 套接字通信。

一个免责声明，我将展示使用 zmqpp 库的示例，但关键点不是要调用的确切函数，而是如何正确应用流，因此我希望您能够将其转换为纯 ZeroMQ。

看看zmq api docs , 原生模式段落。

发起与远程的连接，连接后需要获取ZMQ_IDENTITY。

To open a connection to a server, use the zmq_connect call, and then fetch the socket identity using the ZMQ_IDENTITY zmq_getsockopt call.

zmqpp::socket tcp{context, zmqpp::socket_type::stream};
tcp.connect("tcp://127.0.0.1:12345");
std::string identity = tcp.get<std::string>(zmqpp::socket_option::identity);

什么是身份？二进制数据。来自 docs :

Identity should be at least one byte and at most 255 bytes long. Identities starting with binary zero are reserved for use by ØMQ infrastructure.

获取身份后，可以在socket上接收到获取零长度消息——表示连接已经建立。

When a connection is made, a zero-length message will be received by the application.

我发现，当 ZMQ_STREAM api 文档提到零长度消息时，它实际上是指具有 2 个帧的消息，第一个身份，第二个零长度。

现在，发送。您的代码中缺少的是将身份作为消息的第一帧传递。 ZeroMQ 引擎将删除它并仅发送第二帧(您不能向消息添加更多帧)。

You must send one identity frame followed by one data frame. The ZMQ_SNDMORE flag is required for identity frames but is ignored on data frames.

zmqpp::message msg;
msg << identity;
msg << "Hello world";
tcp.send(msg);

现在您可以等待来自远程的响应，您将收到正确的回复(消息由 2 帧组成，第一个身份，第二个，远程回答了什么)或断开连接指示(零长度消息)。

When receiving TCP data, a ZMQ_STREAM socket shall prepend a message part containing the identity of the originating peer to the message before passing it to the application.

Similarly, when the peer disconnects (or the connection is lost), a zero-length message will be received by the application.

zmqpp::message msg;
tcp.receive(msg);

std::string r_ident;
msg >> r_ident;
std::string body;
msg >> body;