使用 kryo 序列化 protobuf 的问题

Question

我已经用 Java 实现了我自己的序列化程序。让我们称之为 abcSerializer。我要序列化的对象是 abc，它是一个 Google Protocol Buffer 类。

我正在尝试使用 kryo 框架来序列化这个对象。经过谷歌的一些研究和阅读后，我决定继续使用 kryo 序列化程序。我本身没有指定任何序列化程序，所以我假设 kryo 选择了一个默认的序列化程序。

public class abcSerializer implements AttributeSerializer <abc> {

  public abcSerializer() {
      kryo = new Kryo();
  }

  public static Kryo getKryo() {
      return kryo;
  }

  @Override
  public abc read(byte[] buffer) {

    abc xyz = null;

    ByteArrayInputStream abcsStream = new ByteArrayInputStream(buffer);
    Input abcsStreamInput = new Input(abcsStream);
    xyz = getKryo().readObject(abcsStreamInput, abc.class);
    return xyz;
}

@Override
public void write(byte[] buffer, abc abc) {

   ByteArrayOutputStream abcStream = new ByteArrayOutputStream();
   Output abcOutput = new Output(abcStream);

   getKryo().writeObject(abcOutput, abc);

   abcOutput.toBytes()        
}

}

当我执行 writeObject 时，一切都很好。但是，当我执行 readObject 时问题就来了。 Kyro 抛出以下异常。

com.esotericsoftware.kryo.KryoException: Class cannot be created (missing no-arg constructor): java.util.Collections$Unmodifiabljava.lang.IllegalStateException: Encountered error in deserializer [null value returned]. Check serializer compatibility.eRandomAccessList

上面的异常几乎是不言自明的。

kryo 文档如下。 ""特定类型的序列化程序使用 Java 代码创建该类型的新实例。诸如 FieldSerializer 之类的序列化器是通用的，必须处理创建任何类的新实例。默认情况下，如果一个类具有零参数构造函数，则通过 ReflectASM 或反射调用它，否则将引发异常。如果零参数构造函数是私有的，则尝试使用 setAccessible 通过反射访问它。如果这是可以接受的，私有零​​参数构造函数是允许 Kryo 创建类的实例而不影响公共 API 的好方法。""

现在，我有两个问题。

1) google 协议缓冲区生成的类确实有一个无参数构造函数。然而，这似乎是一个私人的。这是一个问题和上述 kryo 异常的根本原因吗？

2) 如果是这样，如何处理上述问题？我的意思是，我如何编写自己的序列化程序并仍然用于序列化 google 协议缓冲区对象数据？

Answer 1

这对你有什么好处？

/**
 * This lets Kryo serialize protobufs more efficiently.
 */
public class ProtobufKryo<P extends GeneratedMessage> extends Serializer<P> {
    protected final Method parser;

    public ProtobufKryo(Class<P> theClass) {
        try {
            parser = theClass.getDeclaredMethod("parseFrom", InputStream.class);
            parser.setAccessible(true);
        } catch (NoSuchMethodException e) {
            throw new IllegalArgumentException(theClass.toString() + " doesn't have parser");
        }
    }

    @Override
    public void write(Kryo kryo, Output output, P generatedMessage) {
        try {
            generatedMessage.writeTo(output);
        } catch (IOException e) {
            // This isn't supposed to happen with a Kryo output.
            throw new RuntimeException(e);
        }
    }

    @Override
    public P read(Kryo kryo, Input input, Class<P> gmClass) {
        try {
            return (P)parser.invoke(null, input);
        } catch (InvocationTargetException | IllegalAccessException e) {
            // These really shouldn't happen
            throw new IllegalArgumentException(e);
        }
    }
}

好吧，解释一下……

当 Kryo 遇到它无法识别的类的对象时，它会回退到 Java 序列化。并非每一种都有效，有时甚至不起作用。

（好吧，我承认以上可能并不总是正确的。它可能是 Kryo 配置的一部分。在我工作的环境中是正确的。）

您可以告诉它为特定类使用自己的序列化，但有时您可以通过为特定类创建自定义序列化器来做得更好。

以上内容利用了 Kryo 中 protobufs 的现有序列化。基本上，它使用现有的 protobuf writeTo() 和 parseFrom() 来处理 Kryo 中的序列化。您将注册上述类以序列化每个 protobuf 类。 （Protobuf 类扩展 GeneratedMessage。）

写出对象只使用普通的protobuf writeTo() 方法。回读 protobuf 使用类 parseFrom() 方法，该方法是通过构造函数中的反射找到的。

因此，您可以使用以下内容配置序列化程序：

  Kryo k = new Kryo();
  k.addDefaultSerializer(MyProtobuf.class, ProtobufKryo.class);
  k.addDefaultSerializer(MyOtherProtobuf.class, ProtobufKryo.class);

等等

Answer 2

我发现 DaveWill's 之前的 answer 有两个小问题：

1. 使用InputStream 会在方法read 中引发异常。
2. 在使用带有许多嵌套类/消息的 Protobuf 生成的类时，有时编写类名很麻烦。

所以我在下面的代码中做了一些小的改进来解决这个问题：

• 将InputStream 替换为byte[]array。
• 添加了一个帮助器 registerMessagesFrom，用于在顶级 Protobuf 生成的类中查找所有嵌套类/消息。

package com.juarezr.serialization;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.Serializer;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.google.protobuf.AbstractMessage;

import java.io.Serializable;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;

public class ProtobufSerializer<T extends AbstractMessage> extends Serializer<T> implements Serializable {
    
    static final long serialVersionUID = 1667386898559074449L;
    protected final Method parser;

    public ProtobufSerializer(final Class<T> protoMessageClass) {
        try {
            this.parser = protoMessageClass.getDeclaredMethod("parseFrom", byte[].class);
            this.parser.setAccessible(true);
        } catch (SecurityException | NoSuchMethodException ex) {
            throw new IllegalArgumentException(protoMessageClass.toString() + " doesn't have a protobuf parser", ex);
        }
    }

    @Override
    public void write(final Kryo kryo, final Output output, final T protobufMessage) {
        if (protobufMessage == null) {
            output.writeByte(Kryo.NULL);
            output.flush();
            return;
        }
        final byte[] bytes = protobufMessage.toByteArray();
        output.writeInt(bytes.length + 1, true);
        output.writeBytes(bytes);
        output.flush();
    }

    @SuppressWarnings({"unchecked", "JavaReflectionInvocation"})
    @Override
    public T read(final Kryo kryo, final Input input, final Class<T> protoMessageClass) {
        final int length = input.readInt(true);
        if (length == Kryo.NULL) {
            return null;
        }
        final Object bytesRead = input.readBytes(length - 1);
        try {
            final Object parsed = this.parser.invoke(protoMessageClass, bytesRead);
            return (T) parsed;
        } catch (IllegalAccessException | InvocationTargetException e) {
            throw new RuntimeException("Unable to deserialize protobuf for class: " + protoMessageClass.getName(), e);
        }
    }

    @Override
    public boolean getAcceptsNull() {
        return true;
    }

    @SuppressWarnings("unchecked")
    public static <M extends AbstractMessage> void registerMessagesFrom(final M rootMessage, final Kryo kryo) {

        final Class<M> messageClass = (Class<M>) rootMessage.getClass();
        final ProtobufSerializer<M> serializer = new ProtobufSerializer<>(messageClass);
        kryo.register(messageClass, serializer);

        final Class<?>[] nestedClasses = messageClass.getDeclaredClasses();
        for (final Class<?> innerClass : nestedClasses) {
            if ((AbstractMessage.class).isAssignableFrom(innerClass)) {
                final Class<M> typedClass = (Class<M>) innerClass;
                final ProtobufSerializer<M> serializer2 = new ProtobufSerializer<>(typedClass);
                kryo.register(typedClass, serializer2);
            }
        }
    }
}

您可以使用以下内容配置序列化：

// ...
final Kryo kryo = new Kryo();

ProtobufSerializer.registerMessagesFrom(MyProtoEnclosingClass.MyProtoTopLevelClass.getDefaultInstance(), kryo);

// Add a registration for each generated file and top level class ...

Answer 3

https://github.com/twitter/chill by twitter 是 kryo 序列化器的集合，其中还包括一个用于 google 协议缓冲区的序列化器：ProtobufSerializer。