netty学习之四 编码解码和传输序列化

• netty发送或接收消息后，Netty必须将消息数据从一种形式转化为另外一种。接收消息后，须要将消息从字节码转成Java对象(由某种解码器解码)；发送消息前，须要将Java对象转成字节(由某些类型的编码器进行编码)。这种转换通常发生在网络程序中，由于网络上只能传输字节数据。

• 严格的说其余handlers能够作编码器和适配器，使用不一样的Adapter classes取决你想要作什么。若是是解码器则有一个ChannelInboundHandlerAdapter或ChannelInboundHandler，全部的解码器都继承或实现它们。“channelRead”方法/事件被覆盖，这个方法从入站(inbound)通道读取每一个消息。重写的channelRead方法将调用每一个解码器的“decode”方法并经过ChannelHandlerContext.fireChannelRead(Object msg)传递给ChannelPipeline中的下一个ChannelInboundHandler。
    相似入站消息，当你发送一个消息出去(出站)时，除编码器将消息转成字节码外还会转发到下一个ChannelOutboundHandler。

• 下面看实际状况开发中发送数据的encode方法

public class NettyEncoder extends MessageToByteEncoder<RemotingCommand> {
        @Override
        public void encode(ChannelHandlerContext ctx, RemotingCommand remotingCommand, ByteBuf out)
                throws Exception {
            if (remotingCommand == null) {
                LOGGER.error("Message is null");
                return;
            }
            try {
                ByteBuffer byteBuffer = codec.encode(remotingCommand);
                out.writeBytes(byteBuffer);
            } catch (Exception e) {
                Channel channel = new NettyChannel(ctx);
                LOGGER.error("encode exception, addr={}, remotingCommand={}", RemotingHelper.parseChannelRemoteAddr(channel), remotingCommand.toString(), e);
                RemotingHelper.closeChannel(channel);
            }
        }
    }

• 使用code.encode的做用是根据本身定义的序列化方法将须要的数据进行序列化，众所周知Java的序列化性能极低，在一些对性能要求较高的场景是不合适的，因此须要提供一些其它常见的序列化方式，好比hession,kryo,protobuf,jackson等序列化方式

public ByteBuffer encode(RemotingCommand remotingCommand) throws Exception {
       RemotingSerializable serializable =
               getRemotingSerializable(remotingCommand.getSid());
       // header length size
       int length = 4;
       // serializable id (int)
       length += 4;
       //  header data length
       byte[] headerData = serializable.serialize(remotingCommand);
       length += headerData.length;
       byte[] bodyData = null;
       byte[] bodyClass = null;
       RemotingCommandBody body = remotingCommand.getBody();
       if (body != null) {
           // body data
           bodyData = serializable.serialize(body);
           length += bodyData.length;
           bodyClass = body.getClass().getName().getBytes();
           length += bodyClass.length;
           length += 4;
       }
       ByteBuffer result = ByteBuffer.allocate(4 + length);
       // length
       result.putInt(length);
       // serializable Id
       result.putInt(serializable.getId());
       // header length
       result.putInt(headerData.length);
       // header data
       result.put(headerData);
       if (bodyData != null) {
           //  body length
           result.putInt(bodyData.length);
           //  body data
           result.put(bodyData);
           // body class
           result.put(bodyClass);
       }
       result.flip();
       return result;
   }

• 序列化接口：

public interface RemotingSerializable {

    int getId();

    byte[] serialize(final Object obj) throws Exception;

    <T> T deserialize(final byte[] data, Class<T> clazz) throws Exception;
}

经过实现该接口能够实现自定义的序列化方式，下面看下，当接收到网络回传的字节数据如何进行反序列化呢

public class NettyDecoder extends LengthFieldBasedFrameDecoder {
        public NettyDecoder() {
            super(appContext.getConfig().getParameter(ExtConfig.NETTY_FRAME_LENGTH_MAX, Constants.DEFAULT_BUFFER_SIZE), 0, 4, 0, 4);
        }
        @Override
        public Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
            try {
                ByteBuf frame = (ByteBuf) super.decode(ctx, in);
                if (frame == null) {
                    return null;
                }

                byte[] tmpBuf = new byte[frame.capacity()];
                frame.getBytes(0, tmpBuf);
                frame.release();

                ByteBuffer byteBuffer = ByteBuffer.wrap(tmpBuf);
                return codec.decode(byteBuffer);
            } catch (Exception e) {
                Channel channel = new NettyChannel(ctx);
                LOGGER.error("decode exception, {}", RemotingHelper.parseChannelRemoteAddr(channel), e);
                RemotingHelper.closeChannel(channel);
            }
            return null;
        }
    }

调用codec的反序列化方法，将byteBuffer转换成自定义的对象。

public RemotingCommand decode(ByteBuffer byteBuffer) throws Exception {
       int length = byteBuffer.limit();
       int serializableId = byteBuffer.getInt();
       RemotingSerializable serializable =
               getRemotingSerializable(serializableId);

       int headerLength = byteBuffer.getInt();
       byte[] headerData = new byte[headerLength];
       byteBuffer.get(headerData);

       RemotingCommand cmd = serializable.deserialize(headerData, RemotingCommand.class);

       int remaining = length - 4 - 4 - headerLength;

       if (remaining > 0) {

           int bodyLength = byteBuffer.getInt();
           int bodyClassLength = remaining - 4 - bodyLength;

           if (bodyLength > 0) {

               byte[] bodyData = new byte[bodyLength];
               byteBuffer.get(bodyData);

               byte[] bodyClassData = new byte[bodyClassLength];
               byteBuffer.get(bodyClassData);

               cmd.setBody((RemotingCommandBody) serializable.deserialize(bodyData, Class.forName(new String(bodyClassData))));
           }
       }
       return cmd;
   }

至此，整个数据传输中的序列化和反序列化过程结束，具体代码已经上传到github。

完整代码连接：https://github.com/winstonelei/Smt