public class Header {
  private short magic; // 魔数
  private byte version; // 协议版本
  private byte extraInfo; // 附加信息
  private Long messageId; // 消息ID
  private Integer size; // 消息体长度
  ... // 省略getter/setter方法
}

public class Request implements Serializable {
  private String serviceName; // 请求的Service类名
  private String methodName; // 请求的方法名称
  private Class[] argTypes; // 请求方法的参数类型
  private Object[] args; // 请求方法的参数
  ... // 省略getter/setter方法
}
public class Response implements Serializable {
  private int code = 0; // 响应的错误码，正常响应为0，非0表示异常响应
  private String errMsg; // 异常信息
  private Object result; // 响应结果
  ... // 省略getter/setter方法
}

public interface Serialization {
  <T> byte[] serialize(T obj)throws IOException;
  <T> T deSerialize(byte[] data, Class<T> clz)throws IOException;
}

public class HessianSerialization implements Serialization {
  public <T> byte[] serialize(T obj) throws IOException {
      ByteArrayOutputStream os = new ByteArrayOutputStream();
      HessianOutput hessianOutput = new HessianOutput(os);
      hessianOutput.writeObject(obj);
      return os.toByteArray();
  }
  public <T> T deSerialize(byte[] data, Class<T> clazz) 
          throws IOException {
      ByteArrayInputStream is = new ByteArrayInputStream(data);
      HessianInput hessianInput = new HessianInput(is);
      return (T) hessianInput.readObject(clazz);
  }
}

public interface Compressor {
  byte[] compress(byte[] array) throws IOException;
  byte[] unCompress(byte[] array) throws IOException;
}

public class SnappyCompressor implements Compressor {
  public byte[] compress(byte[] array) throws IOException {
      if (array == null) { return null; }
      return Snappy.compress(array);
  }
  public byte[] unCompress(byte[] array) throws IOException {
      if (array == null) { return null; }
      return Snappy.uncompress(array);
  }
}

public class DemoRpcDecoder extends ByteToMessageDecoder {
  protected void decode(ChannelHandlerContext ctx,
         ByteBuf byteBuf, List<Object> out) throws Exception {
      if (byteBuf.readableBytes() < Constants.HEADER_SIZE) {
          return; // 不到16字节的话无法解析消息头，暂不读取
      }
      // 记录当前readIndex指针的位置，方便重置
      byteBuf.markReaderIndex();
      // 尝试读取消息头的魔数部分
      short magic = byteBuf.readShort();
      if (magic != Constants.MAGIC) { // 魔数不匹配会抛出异常
          byteBuf.resetReaderIndex(); // 重置readIndex指针
          throw new RuntimeException("magic number error:" + magic);
      }
      // 依次读取消息版本、附加信息、消息ID以及消息体长度四部分
      byte version = byteBuf.readByte();
      byte extraInfo = byteBuf.readByte();
      long messageId = byteBuf.readLong();
      int size = byteBuf.readInt();
      Object request = null;
      // 心跳消息是没有消息体的，无须读取
      if (!Constants.isHeartBeat(extraInfo)) {
          // 对于非心跳消息，没有积累到足够的数据是无法进行反序列化的
          if (byteBuf.readableBytes() < size) {
              byteBuf.resetReaderIndex();
              return;
          }
          // 读取消息体并进行反序列化
          byte[] payload = new byte[size];
          byteBuf.readBytes(payload);
          // 这里根据消息头中的extraInfo部分选择相应的序列化和压缩方式
          Serialization serialization = 
            SerializationFactory.get(extraInfo);
          Compressor compressor = CompressorFactory.get(extraInfo);
          // 经过解压缩和反序列化得到消息体
          request = serialization.deserialize(
              compressor.unCompress(payload), Request.class);
      }
      // 将上面读取到的消息头和消息体拼装成完整的Message并向后传递
      Header header = new Header(magic, version, extraInfo, 
          messageId, size);
      Message message = new Message(header, request);
      out.add(message);
  }
}

class DemoRpcEncoder extends MessageToByteEncoder<Message>{
  @Override
  protected void encode(ChannelHandlerContext channelHandlerContext,
              Message message, ByteBuf byteBuf) throws Exception {
      Header header = message.getHeader();
      // 依次序列化消息头中的魔数、版本、附加信息以及消息ID
      byteBuf.writeShort(header.getMagic());
      byteBuf.writeByte(header.getVersion());
      byteBuf.writeByte(header.getExtraInfo());
      byteBuf.writeLong(header.getMessageId());
      Object content = message.getContent();
      if (Constants.isHeartBeat(header.getExtraInfo())) {
          byteBuf.writeInt(0); // 心跳消息，没有消息体，这里写入0
          return;
      }
      // 按照extraInfo部分指定的序列化方式和压缩方式进行处理
      Serialization serialization = 
          SerializationFactory.get(header.getExtraInfo());
      Compressor compressor = 
          CompressorFactory.get(header.getExtraInfo());
      byte[] payload = compressor.compress(
          serialization.serialize(content));
      byteBuf.writeInt(payload.length); // 写入消息体长度
      byteBuf.writeBytes(payload); // 写入消息体
  }
}