java - Java读取protobuf消息时出现异常

Question

我现在使用 protobuf 已经有几个星期了，但是在用 Java 解析 protobuf 消息时我仍然不断遇到异常。

我使用 C++ 创建我的 protobuf 消息，并使用 boost sockets 将它们发送到 Java 客户端监听的服务器套接字。发送消息的C++代码是这样的:

boost::asio::streambuf b;
std::ostream os(&b);

ZeroCopyOutputStream *raw_output = new OstreamOutputStream(&os);
CodedOutputStream *coded_output = new CodedOutputStream(raw_output);

coded_output->WriteVarint32(agentMessage.ByteSize());
agentMessage.SerializeToCodedStream(coded_output);

delete coded_output;
delete raw_output;

boost::system::error_code ignored_error;

boost::asio::async_write(socket, b.data(), boost::bind(
        &MessageService::handle_write, this,
        boost::asio::placeholders::error));

如您所见，我使用 WriteVarint32 编写了消息的长度，因此 Java 端应该通过使用 parseDelimitedFrom 知道它应该读取多远:

AgentMessage agentMessage = AgentMessageProtos.AgentMessage    
                                .parseDelimitedFrom(socket.getInputStream());

但这没有帮助，我不断收到这些异常:

Protocol message contained an invalid tag (zero).
Message missing required fields: ...
Protocol message tag had invalid wire type.
Protocol message end-group tag did not match expected tag.
While parsing a protocol message, the input ended unexpectedly in the middle of a field.  This could mean either than the input has been truncated or that an embedded message misreported its own length.

重要要知道，这些异常不会在每条消息上抛出。这只是我收到最多的消息中的一小部分，效果很好 - 我仍然想解决这个问题，因为我不想遗漏这些消息。

如果有人能帮助我或花费他的想法，我将非常感激。

---

另一个有趣的事实是我收到的消息数量。 2 秒内总共 1.000 条消息对于我的程序来说通常是正常的。在 20 秒内大约 100.000 等等。我人为减少了发送的消息，当只发送6-8条消息时，完全没有错误。那么这可能是 Java 客户端套接字端的缓冲问题吗？

假设有 60.000 条消息，其中平均有 5 条已损坏。

Answer 1

[我不是真正的 TCP 专家，这可能有点偏]

问题是，[Java] TCP Socket 的 read(byte[] buffer) 将在读取到 TCP 帧末尾后返回。如果这恰好是中间消息(我的意思是，protobuf 消息)，解析器将阻塞并抛出一个 InvalidProtocolBufferException。

任何 protobuf 解析调用在内部使用 CodedInputStream ( src here )，如果源是 InputStream，它依赖于 read() -- 因此，受 TCP 套接字问题的影响。

因此，当您通过套接字填充大量数据时，一些消息必然会分成两个帧——这就是它们被破坏的地方。

我猜，当您降低消息传输速率(如您所说的每秒 6-8 条消息)时，每个帧都会在下一个数据片段放入流之前发送，因此每条消息总是有自己的TCP 帧，即没有被分割也没有错误。 (或者可能只是错误很少见，低错误率只是意味着您需要更多时间才能看到它们)

至于解决方案，最好的办法是自己处理缓冲区，即从套接字读取 byte[](可能使用 readFully() 而不是read() 因为前者会阻塞，直到有足够的数据填充缓冲区 [或遇到 EOF]，所以它有点抵抗中间消息帧结束的事情)，确保它有足够的数据被解析成一个完整的消息，然后将缓冲区提供给解析器。

另外，在 this Google Groups topic 中有一些关于这个主题的好读物-- 这就是我得到 readFully() 部分的地方。