netty权威指南之七 其他协议

最近通过学习@李林峰 大神的 《netty权威指南》

对Java io 有了进一步的了解，我对其代码加了一些注解和个人的看法。

对netty有了进一步的了解，所以后面有一些对rocketmq和dubbo中协议部分的分析

还是那句话 学习使人进步  

前面学习了基本的codec   

FixedLengthFrameDecoder  

DelimiterBasedFrameDecoder  

LineBasedFrameDecoder 

当然netty，已经帮我们实现了一些常见的协议

今天我们学习下 一些简单的协议

serialization  java序列化 

@Override protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception { ByteBuf frame = (ByteBuf) super.decode(ctx, in); if (frame == null) { return null; } return new CompactObjectInputStream( new ByteBufInputStream(frame), classResolver).readObject(); }

@Override protected void encode(ChannelHandlerContext ctx, Serializable msg, ByteBuf out) throws Exception { int startIdx = out.writerIndex(); ByteBufOutputStream bout = new ByteBufOutputStream(out); bout.write(LENGTH_PLACEHOLDER); ObjectOutputStream oout = new CompactObjectOutputStream(bout); oout.writeObject(msg); oout.flush(); oout.close(); int endIdx = out.writerIndex(); out.setInt(startIdx, endIdx - startIdx - 4); }

protobuf  为啥加ProtobufVarint32FrameDecoder 和 ProtobufVarint32LengthFieldPrepender

ch.pipeline().addLast( new ProtobufVarint32FrameDecoder()); ch.pipeline().addLast( new ProtobufDecoder( SubscribeRespProto.SubscribeResp .getDefaultInstance())); ch.pipeline().addLast( new ProtobufVarint32LengthFieldPrepender()); ch.pipeline().addLast(new ProtobufEncoder()); @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { in.markReaderIndex(); final byte[] buf = new byte[5]; for (int i = 0; i < buf.length; i ++) { if (!in.isReadable()) { in.resetReaderIndex(); return; } buf[i] = in.readByte(); if (buf[i] >= 0) { int length = CodedInputStream.newInstance(buf, 0, i + 1).readRawVarint32(); if (length < 0) { throw new CorruptedFrameException("negative length: " + length); } if (in.readableBytes() < length) { in.resetReaderIndex(); return; } else { out.add(in.readBytes(length)); return; } } } // Couldn't find the byte whose MSB is off. throw new CorruptedFrameException("length wider than 32-bit"); } @Override protected void encode( ChannelHandlerContext ctx, ByteBuf msg, ByteBuf out) throws Exception { int bodyLen = msg.readableBytes(); int headerLen = CodedOutputStream.computeRawVarint32Size(bodyLen); out.ensureWritable(headerLen + bodyLen); CodedOutputStream headerOut = CodedOutputStream.newInstance(new ByteBufOutputStream(out)); headerOut.writeRawVarint32(bodyLen); headerOut.flush(); out.writeBytes(msg, msg.readerIndex(), bodyLen); }

marshalling  

protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception { ByteBuf frame = (ByteBuf) super.decode(ctx, in); if (frame == null) { return null; } Unmarshaller unmarshaller = provider.getUnmarshaller(ctx); ByteInput input = new ChannelBufferByteInput(frame); try { unmarshaller.start(input); Object obj = unmarshaller.readObject(); unmarshaller.finish(); return obj; } finally { // Call close in a finally block as the ReplayingDecoder will throw an Error if not enough bytes are // readable. This helps to be sure that we do not leak resource unmarshaller.close(); } } protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception { Marshaller marshaller = provider.getMarshaller(ctx); int lengthPos = out.writerIndex(); out.writeBytes(LENGTH_PLACEHOLDER); ChannelBufferByteOutput output = new ChannelBufferByteOutput(out); marshaller.start(output); marshaller.writeObject(msg); marshaller.finish(); marshaller.close(); out.setInt(lengthPos, out.writerIndex() - lengthPos - 4); }