Netty的编码器和解码器使用KRYO系列化

Question

我是新来的Netty 4.0.29，并努力使自定义对象编码器，并使用KRYO作为序列化库对象解码器，我无法使它工作。因为我是Netty的新手，也许这就是我搞乱某些东西的原因。我使用MessageToMessageEncoder和MessageToMessageDecoder

这里是我的编码器 -

public class KryoEncoder extends MessageToMessageEncoder<Object>{ 

    private Kryo kryo; 
    private Output output; 

    public KryoEncoder(Kryo kryo, int bufSize, int maxBufSize) { 
     this.kryo = kryo; 
     output = new Output(bufSize, maxBufSize); 
    } 

    @Override 
    protected void encode(ChannelHandlerContext ctx, Object msg, 
      List<Object> out) throws Exception { 
     output.clear(); 
     kryo.writeClassAndObject(output, msg); 
     int total = output.position(); 
     out.add(Unpooled.wrappedBuffer(output.getBuffer(), 0 , total)); 
    } 

} 

这里是解码器 -

public class KryoDecoder extends MessageToMessageDecoder<Object>{ 
    private Kryo kryo; 
    private Input input; 

    public KryoDecoder (Kryo kryo) { 
     this.kryo = kryo; 
     input = new Input(); 
    } 

    @Override 
    protected void decode(ChannelHandlerContext ctx, Object msg, List<Object> out) throws Exception { 
     ByteBuf buffer = (ByteBuf)msg; 
     byte[] ar = new byte[buffer.readableBytes()]; 
     buffer.readBytes(ar); 
     input.read(ar, buffer.readerIndex(), buffer.readableBytes()); 
     Object object = kryo.readClassAndObject(input); 
     buffer.readerIndex(input.position()); 
     out.add(object); 
    } 
} 

这里是异常的短栈抛出，我知道它是一个空指针异常，在解码器的字节数组中有某些地方出错，但不知道是什么。

Caused by: java.lang.NullPointerException 
    at java.lang.System.arraycopy(Native Method) 
    at com.esotericsoftware.kryo.io.Input.read(Input.java:254) 
    at TestWithKryo.KryoDecoder.decode(KryoDecoder.java:31) 
    at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:89) 
    ... 14 more 

例外解码和我检查了编码并发送编码的对象时检查的ChannelFuture，并没有抛出任何异常时，会发生。再次

，因为我真正的新网状我可能做的东西真的错了，所以请让我知道。

Answer 1

我觉得这个更好的将是ByteToMessageDecoder和MessageToByteEncoder。不要忘记发送第一个数据长度，否则你不会识别一个数据包开始和另一个结束的位置。还有就是我的简单的例子：

public class KryoDecoder extends ByteToMessageDecoder { 

    private final Kryo kryo; 

    public KryoDecoder(Kryo kryo) { 
     this.kryo = kryo; 
    } 

    @Override 
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { 

     if (in.readableBytes() < 2) 
      return; 

     in.markReaderIndex(); 

     int len = in.readUnsignedShort(); 

     if (in.readableBytes() < len) { 
      in.resetReaderIndex(); 
      return; 
     } 

     byte[] buf = new byte[len]; 
     in.readBytes(buf); 
     Input input = new Input(buf); 
     Object object = kryo.readClassAndObject(input); 
     out.add(object); 

    } 
} 

和编码器

public class KryoEncoder extends MessageToByteEncoder<Object> { 

    private final Kryo kryo; 

    public KryoEncoder(Kryo kryo) { 
     this.kryo = kryo; 
    } 

    @Override 
    protected void encode(ChannelHandlerContext ctx, Object in, ByteBuf out) throws Exception { 
     ByteArrayOutputStream outStream = new ByteArrayOutputStream(); 
     Output output = new Output(outStream, 4096); 

     kryo.writeClassAndObject(output, in); 
     output.flush(); 

     byte[] outArray = outStream.toByteArray(); 
     out.writeShort(outArray.length); 
     out.writeBytes(outArray); 
    } 

}