是否有C ++当量为Protocol Buffers的分隔符在Java中的I / O功能？是否有C

2019-05-09 04:04发布

站内文章 / 前端开发

9 0

聊天终结者

女 | 书童

私信

我想读/从文件中写入多个Protocol Buffers的消息，在C ++和Java。谷歌暗示邮件之前书写长度前缀，但有没有办法做到这一点在默认情况下（即我可以看到）。

然而，在2.1.0版本的Java API获得了一套“分隔” I / O功能，这显然做的工作：

parseDelimitedFrom
mergeDelimitedFrom
writeDelimitedTo

是否有C ++等价物？如果不是，什么是对的大小电线格式前缀了Java API附加，这样我就可以在解析C ++这些消息？

更新：

现在，这些存在于google/protobuf/util/delimited_message_util.h为3.3.0的。

Answer 1:

我有点晚了这里的聚会，但是下面的实现包括一些优化从其他的答案失踪和输入的64MB后不会失败（虽然它仍然强制执行的64MB限制的各个消息，只是没有对整个流）。

（我是C ++和Java的protobuf库的作者，但我不再为谷歌工作。对不起，该代码从未进入正式lib中。这是它会是什么样子，如果它有）。

bool writeDelimitedTo(
    const google::protobuf::MessageLite& message,
    google::protobuf::io::ZeroCopyOutputStream* rawOutput) {
  // We create a new coded stream for each message.  Don't worry, this is fast.
  google::protobuf::io::CodedOutputStream output(rawOutput);

  // Write the size.
  const int size = message.ByteSize();
  output.WriteVarint32(size);

  uint8_t* buffer = output.GetDirectBufferForNBytesAndAdvance(size);
  if (buffer != NULL) {
    // Optimization:  The message fits in one buffer, so use the faster
    // direct-to-array serialization path.
    message.SerializeWithCachedSizesToArray(buffer);
  } else {
    // Slightly-slower path when the message is multiple buffers.
    message.SerializeWithCachedSizes(&output);
    if (output.HadError()) return false;
  }

  return true;
}

bool readDelimitedFrom(
    google::protobuf::io::ZeroCopyInputStream* rawInput,
    google::protobuf::MessageLite* message) {
  // We create a new coded stream for each message.  Don't worry, this is fast,
  // and it makes sure the 64MB total size limit is imposed per-message rather
  // than on the whole stream.  (See the CodedInputStream interface for more
  // info on this limit.)
  google::protobuf::io::CodedInputStream input(rawInput);

  // Read the size.
  uint32_t size;
  if (!input.ReadVarint32(&size)) return false;

  // Tell the stream not to read beyond that size.
  google::protobuf::io::CodedInputStream::Limit limit =
      input.PushLimit(size);

  // Parse the message.
  if (!message->MergeFromCodedStream(&input)) return false;
  if (!input.ConsumedEntireMessage()) return false;

  // Release the limit.
  input.PopLimit(limit);

  return true;
}

Answer 2:

好了，所以我一直没能找到顶级的C ++函数实现我所需要的，但有些是通过Java API参考洞穴探险止跌回升下面，里面MessageLite接口：

void writeDelimitedTo(OutputStream output)
/*  Like writeTo(OutputStream), but writes the size of 
    the message as a varint before writing the data.   */

因此Java大小前缀是一个（Protocol Buffers的）varint！

有了这些信息，我所经历的C ++ API挖掘和发现CodedStream头，它有以下：

bool CodedInputStream::ReadVarint32(uint32 * value)
void CodedOutputStream::WriteVarint32(uint32 value)

使用这些，我应该能够推出自己的C ++函数来完成这项工作。

他们真的应该加入这主要消息API虽然; 它缺少的功能考虑Java有它，也是如此马克Gravell的优秀protobuf网C＃端口（通过SerializeWithLengthPrefix和DeserializeWithLengthPrefix）。

Answer 3:

我解决了使用CodedOutputStream / ArrayOutputStream写消息（具有大小）和CodedInputStream / ArrayInputStream读取消息（具有大小）相同的问题。

例如，下面的伪代码中写入消息大小由该消息执行以下操作：

const unsigned bufLength = 256;
unsigned char buffer[bufLength];
Message protoMessage;

google::protobuf::io::ArrayOutputStream arrayOutput(buffer, bufLength);
google::protobuf::io::CodedOutputStream codedOutput(&arrayOutput);

codedOutput.WriteLittleEndian32(protoMessage.ByteSize());
protoMessage.SerializeToCodedStream(&codedOutput);

书写时，您也应该检查你的缓冲区足够大，以适应信息（包括大小）。和阅读时，你应该检查你的缓冲区包含整个消息（包括大小）。

它如果他们更加方便的方法来C ++ API类似于由Java API提供的那些肯定会得心应手。

Answer 4:

干得好：

#include <google/protobuf/io/zero_copy_stream_impl.h>
#include <google/protobuf/io/coded_stream.h>

using namespace google::protobuf::io;

class FASWriter 
{
    std::ofstream mFs;
    OstreamOutputStream *_OstreamOutputStream;
    CodedOutputStream *_CodedOutputStream;
public:
    FASWriter(const std::string &file) : mFs(file,std::ios::out | std::ios::binary)
    {
        assert(mFs.good());

        _OstreamOutputStream = new OstreamOutputStream(&mFs);
        _CodedOutputStream = new CodedOutputStream(_OstreamOutputStream);
    }

    inline void operator()(const ::google::protobuf::Message &msg)
    {
        _CodedOutputStream->WriteVarint32(msg.ByteSize());

        if ( !msg.SerializeToCodedStream(_CodedOutputStream) )
            std::cout << "SerializeToCodedStream error " << std::endl;
    }

    ~FASWriter()
    {
        delete _CodedOutputStream;
        delete _OstreamOutputStream;
        mFs.close();
    }
};

class FASReader
{
    std::ifstream mFs;

    IstreamInputStream *_IstreamInputStream;
    CodedInputStream *_CodedInputStream;
public:
    FASReader(const std::string &file), mFs(file,std::ios::in | std::ios::binary)
    {
        assert(mFs.good());

        _IstreamInputStream = new IstreamInputStream(&mFs);
        _CodedInputStream = new CodedInputStream(_IstreamInputStream);      
    }

    template<class T>
    bool ReadNext()
    {
        T msg;
        unsigned __int32 size;

        bool ret;
        if ( ret = _CodedInputStream->ReadVarint32(&size) )
        {   
            CodedInputStream::Limit msgLimit = _CodedInputStream->PushLimit(size);
            if ( ret = msg.ParseFromCodedStream(_CodedInputStream) )
            {
                _CodedInputStream->PopLimit(msgLimit);      
                std::cout << mFeed << " FASReader ReadNext: " << msg.DebugString() << std::endl;
            }
        }

        return ret;
    }

    ~FASReader()
    {
        delete _CodedInputStream;
        delete _IstreamInputStream;
        mFs.close();
    }
};

Answer 5:

IsteamInputStream是正交函数和其他错误很脆弱，很容易用的std :: istream的一起使用时发生。在此之后，protobuf的流permamently损坏，而且已经使用缓冲区中的数据被破坏。有在protobuf的从传统的流阅读适当的支持。

实现google::protobuf::io::CopyingInputStream ，并与一起使用CopyingInputStreamAdapter 。做输出变量相同。

在实践中，分析在通话结束了google::protobuf::io::CopyingInputStream::Read(void* buffer, int size) ，其中一个缓冲区中给出。剩下要做的唯一事情是莫名其妙读进去。

下面是用于与短耳同步流一起使用的实例（ SyncReadStream / SyncWriteStream ）：

#include <google/protobuf/io/zero_copy_stream_impl_lite.h>

using namespace google::protobuf::io;


template <typename SyncReadStream>
class AsioInputStream : public CopyingInputStream {
    public:
        AsioInputStream(SyncReadStream& sock);
        int Read(void* buffer, int size);
    private:
        SyncReadStream& m_Socket;
};


template <typename SyncReadStream>
AsioInputStream<SyncReadStream>::AsioInputStream(SyncReadStream& sock) :
    m_Socket(sock) {}


template <typename SyncReadStream>
int
AsioInputStream<SyncReadStream>::Read(void* buffer, int size)
{
    std::size_t bytes_read;
    boost::system::error_code ec;
    bytes_read = m_Socket.read_some(boost::asio::buffer(buffer, size), ec);

    if(!ec) {
        return bytes_read;
    } else if (ec == boost::asio::error::eof) {
        return 0;
    } else {
        return -1;
    }
}


template <typename SyncWriteStream>
class AsioOutputStream : public CopyingOutputStream {
    public:
        AsioOutputStream(SyncWriteStream& sock);
        bool Write(const void* buffer, int size);
    private:
        SyncWriteStream& m_Socket;
};


template <typename SyncWriteStream>
AsioOutputStream<SyncWriteStream>::AsioOutputStream(SyncWriteStream& sock) :
    m_Socket(sock) {}


template <typename SyncWriteStream>
bool
AsioOutputStream<SyncWriteStream>::Write(const void* buffer, int size)
{   
    boost::system::error_code ec;
    m_Socket.write_some(boost::asio::buffer(buffer, size), ec);
    return !ec;
}

用法：

AsioInputStream<boost::asio::ip::tcp::socket> ais(m_Socket); // Where m_Socket is a instance of boost::asio::ip::tcp::socket
CopyingInputStreamAdaptor cis_adp(&ais);
CodedInputStream cis(&cis_adp);

Message protoMessage;
uint32_t msg_size;

/* Read message size */
if(!cis.ReadVarint32(&msg_size)) {
    // Handle error
 }

/* Make sure not to read beyond limit of message */
CodedInputStream::Limit msg_limit = cis.PushLimit(msg_size);
if(!msg.ParseFromCodedStream(&cis)) {
    // Handle error
}

/* Remove limit */
cis.PopLimit(msg_limit);

Answer 6:

我跑进C ++和Python的同样的问题。

对于C ++版本，我使用的代码肯顿·瓦尔达张贴在这个线程，并从他送往protobuf的球队（因为在此发布的版本不处理EOF而一个他送到github上确实拉入请求的代码的混合）。

#include <google/protobuf/message_lite.h>
#include <google/protobuf/io/zero_copy_stream.h>
#include <google/protobuf/io/coded_stream.h>


bool writeDelimitedTo(const google::protobuf::MessageLite& message,
    google::protobuf::io::ZeroCopyOutputStream* rawOutput)
{
    // We create a new coded stream for each message.  Don't worry, this is fast.
    google::protobuf::io::CodedOutputStream output(rawOutput);

    // Write the size.
    const int size = message.ByteSize();
    output.WriteVarint32(size);

    uint8_t* buffer = output.GetDirectBufferForNBytesAndAdvance(size);
    if (buffer != NULL)
    {
        // Optimization:  The message fits in one buffer, so use the faster
        // direct-to-array serialization path.
        message.SerializeWithCachedSizesToArray(buffer);
    }

    else
    {
        // Slightly-slower path when the message is multiple buffers.
        message.SerializeWithCachedSizes(&output);
        if (output.HadError())
            return false;
    }

    return true;
}

bool readDelimitedFrom(google::protobuf::io::ZeroCopyInputStream* rawInput, google::protobuf::MessageLite* message, bool* clean_eof)
{
    // We create a new coded stream for each message.  Don't worry, this is fast,
    // and it makes sure the 64MB total size limit is imposed per-message rather
    // than on the whole stream.  (See the CodedInputStream interface for more
    // info on this limit.)
    google::protobuf::io::CodedInputStream input(rawInput);
    const int start = input.CurrentPosition();
    if (clean_eof)
        *clean_eof = false;


    // Read the size.
    uint32_t size;
    if (!input.ReadVarint32(&size))
    {
        if (clean_eof)
            *clean_eof = input.CurrentPosition() == start;
        return false;
    }
    // Tell the stream not to read beyond that size.
    google::protobuf::io::CodedInputStream::Limit limit = input.PushLimit(size);

    // Parse the message.
    if (!message->MergeFromCodedStream(&input)) return false;
    if (!input.ConsumedEntireMessage()) return false;

    // Release the limit.
    input.PopLimit(limit);

    return true;
}

这里是我的python2实现：

from google.protobuf.internal import encoder
from google.protobuf.internal import decoder

#I had to implement this because the tools in google.protobuf.internal.decoder
#read from a buffer, not from a file-like objcet
def readRawVarint32(stream):
    mask = 0x80 # (1 << 7)
    raw_varint32 = []
    while 1:
        b = stream.read(1)
        #eof
        if b == "":
            break
        raw_varint32.append(b)
        if not (ord(b) & mask):
            #we found a byte starting with a 0, which means it's the last byte of this varint
            break
    return raw_varint32

def writeDelimitedTo(message, stream):
    message_str = message.SerializeToString()
    delimiter = encoder._VarintBytes(len(message_str))
    stream.write(delimiter + message_str)

def readDelimitedFrom(MessageType, stream):
    raw_varint32 = readRawVarint32(stream)
    message = None

    if raw_varint32:
        size, _ = decoder._DecodeVarint32(raw_varint32, 0)

        data = stream.read(size)
        if len(data) < size:
            raise Exception("Unexpected end of file")

        message = MessageType()
        message.ParseFromString(data)

    return message

#In place version that takes an already built protobuf object
#In my tests, this is around 20% faster than the other version 
#of readDelimitedFrom()
def readDelimitedFrom_inplace(message, stream):
    raw_varint32 = readRawVarint32(stream)

    if raw_varint32:
        size, _ = decoder._DecodeVarint32(raw_varint32, 0)

        data = stream.read(size)
        if len(data) < size:
            raise Exception("Unexpected end of file")

        message.ParseFromString(data)

        return message
    else:
        return None

它可能不是最好看的代码，我敢肯定，它可以进行重构公平一点，但至少应该表现出你做这件事。

现在的大问题：它的速度慢 。

即使是使用C ++实现的python-的protobuf的时候，它的幅度比在纯C ++慢一个数量级。我有一个基准，其中我从文件中读取的每〜30个字节10M protobuf的消息。这需要在〜C ++ 0.9S，和35S的蟒蛇。

让它快一点将重新实现varint解码器，使其从一个旅途中的文件和解码读取，而不是从文件中读取并解码，然后作为这个代码目前确实，一个办法。（分析表明，时间显著量在varint编码器/解码器略去）。但是，不用说，是不够关闭Python版本和C ++版本之间的差距。

任何想法，使其更快是非常欢迎的。

Answer 7:

也一直在寻找一个解决方案。下面是我们解决方案的核心，承担一些Java代码写了很多MyRecord消息以writeDelimitedTo到文件中。打开文件，并循环，这样做：

if(someCodedInputStream->ReadVarint32(&bytes)) {
  CodedInputStream::Limit msgLimit = someCodedInputStream->PushLimit(bytes);
  if(myRecord->ParseFromCodedStream(someCodedInputStream)) {
    //do your stuff with the parsed MyRecord instance
  } else {
    //handle parse error
  }
  someCodedInputStream->PopLimit(msgLimit);
} else {
  //maybe end of file
}

希望能帮助到你。

Answer 8:

与协议缓冲区的一个Objective-C版本的工作，我就遇到了这个确切的问题。在从iOS客户端使用parseDelimitedFrom认为，这一长度作为第一个字节一个基于Java的服务器发送，我需要首先调用writeRawByte到CodedOutputStream。张贴在这里hopegully帮助其他人遇到这个问题。同时，通过这个问题的工作，人们会认为谷歌原BUFS会拿出一个简单的标志，该标志为您完成此...

    Request* request = [rBuild build];

    [self sendMessage:request];
} 


- (void) sendMessage:(Request *) request {

    //** get length
    NSData* n = [request data];
    uint8_t len = [n length];

    PBCodedOutputStream* os = [PBCodedOutputStream streamWithOutputStream:outputStream];
    //** prepend it to message, such that Request.parseDelimitedFrom(in) can parse it properly
    [os writeRawByte:len];
    [request writeToCodedOutputStream:os];
    [os flush];
}

Answer 9:

因为我不能写这对与前述肯顿·瓦尔达的回答评论; 我相信这是在他发布的代码中的错误（以及在已经提供了其他的答案）。下面的代码：

...
google::protobuf::io::CodedInputStream input(rawInput);

// Read the size.
uint32_t size;
if (!input.ReadVarint32(&size)) return false;

// Tell the stream not to read beyond that size.
google::protobuf::io::CodedInputStream::Limit limit =
    input.PushLimit(size);
...

设置不正确的限制，因为它没有考虑到它已经从输入读取varint32的大小。这会导致数据丢失/损坏作为附加字节从它可以是下一个消息的一部分的流中读取。正确处理这种通常的方法是删除用于读取的大小并创建用于读取的有效载荷一个新的CodedInputStream：

...
uint32_t size;
{
  google::protobuf::io::CodedInputStream input(rawInput);

  // Read the size.
  if (!input.ReadVarint32(&size)) return false;
}

google::protobuf::io::CodedInputStream input(rawInput);

// Tell the stream not to read beyond that size.
google::protobuf::io::CodedInputStream::Limit limit =
    input.PushLimit(size);
...

Answer 10:

您可以使用函数getline用于从流中读取一个字符串，使用指定的分隔符：

istream& getline ( istream& is, string& str, char delim );

（在报头中定义的）

文章来源: Are there C++ equivalents for the Protocol Buffers delimited I/O functions in Java?

标签： java c++ serialization protocol-buffers

聊天终结者

女 | 书童

私信

收藏的人(0)

Ta的文章更多文章

0条评论

还没有人评论过~