Kafka Consumer 中反序列化 protostuff 对象的问题答案

【问题标题】：Issue in deserialize protostuff object in Kafka ConsumerKafka Consumer 中反序列化 protostuff 对象的问题
【发布时间】：2016-11-11 21:11:15
【问题描述】：

在 Kafka Consumer 中将 byte[] 反序列化为 protostuff 对象时出现以下异常

java.lang.NegativeArraySizeException
at com.dyuproject.protostuff.GraphIOUtil.mergeDelimitedFrom(GraphIOUtil.java:209)
at com.gme.protocols.protostuff.GmeTrade.readExternal(GmeTrade.java:2772)
at java.io.ObjectInputStream.readExternalData(Unknown Source)
at java.io.ObjectInputStream.readOrdinaryObject(Unknown Source)
at java.io.ObjectInputStream.readObject0(Unknown Source)
at java.io.ObjectInputStream.readObject(Unknown Source)

使用以下代码将 protostuff 对象转换为 byte[]。

public static byte[] toBytes(Object o)
{
    try
    {
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        ObjectOutputStream oos = new ObjectOutputStream(baos);
        oos.writeObject(o);
        oos.close();
        byte[] b = baos.toByteArray();
        return b;
    }
    catch (IOException e)
    {
        return new byte[0];
    }
}

使用主题为“XX”的 Kafka 生产者发送该字节 []，其中字节 [] 长度仅为 240。使用 Kafka 消费者接收到该记录。 record.value().length (byte[]) 长度与我从生产者端发送的相同 240。

使用以下代码将该字节 [] 反序列化为对象。

public static Object fromBytes(byte[] bytes)
{
    try
    {
        return new ObjectInputStream(new ByteArrayInputStream(bytes)).readObject();
    }
    catch (Exception e)
    {
        e.printStackTrace();
        return null;
    }
}

得到上述异常。我做错了什么？使用 kafka_2.11-0.9.0.0 供您参考。还有其他需要的吗？

【问题讨论】：

标签： java apache-kafka kafka-consumer-api kafka-producer-api protostuff

【解决方案1】：

我找到了解决方案。 toBytes 和 fromBytes 中存在问题。需要使用 ProtostuffIOUtil.toByteArray 方法将其转换为 byte[]。

序列化

public static byte[] toBytes(Foo o)
{
  LinkedBuffer BUFFER = LinkedBuffer.allocate(1024*1024);
  Schema<Foo> SCHEMA = Foo.getSchema();
  return ProtostuffIOUtil.toByteArray(o, SCHEMA, BUFFER);
}

再次需要使用 ProtostuffIOUtil.mergeFrom 方法将 byte[] 转换为对象。

反序列化

public static Foo fromBytes(byte[] bytes)
{
  Foo tmp = Foo.getSchema().newMessage();
  ProtostuffIOUtil.mergeFrom(bytes, tmp, Foo.getSchema());
  return tmp;
}

注意：使用 ObjectOutputStream/ObjectInputStream 的序列化/反序列化不适用于 protostuff 对象。

【讨论】：