【Netty】七、Netty自定义编码和解码器

浅浅的花香味﹌ 2023-10-09 15:07 64阅读 0赞

Netty自定义编码和解码

  • 一、Netty自定义编码和解码)
  • 服务端代码
    • ProtocolServer
    • ProtocolServerHandler
  • 客户端
    • ProtocolClient
    • ProtocolClientHandler
  • 编解码器
    • RpcDecoder 、RpcEncoder

一、Netty自定义编码和解码)

编解码器的作用就是讲原始字节数据与自定义的消息对象进行互转，网络中都是以字节码的数据形式来传输数据的，服务器编码数据后发送到客户端，客户端需要对数据进行解码，因为编解码器由两部分组成：

  • Decoder(解码器)
  • Encoder(编码器)
    自定义编码器和解码器，可以参考netty里面内置的一些编码器和解码器的代码，比如：
    StringDecoder、StringEncoder
    ObjectDecoder、ObjectEncoder
    如下的例子是采用Fastjson将一个RpcMessage 的java对象在网络中传输所进行的编码和解码，更多场景可以采用该思路进行；

  • 解码：

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

    1. ByteBuf frame = in.retainedDuplicate();
    2. final String content = frame.toString(CharsetUtil.UTF_8);
    3. RpcMessage message = JSON.parseObject(content, RpcMessage.class);
    4. out.add(message);
    5. in.skipBytes(in.readableBytes());

    }

    编码：

    1. @Override
    2. protected void encode(ChannelHandlerContext ctx, RpcMessage msg, ByteBuf out) throws Exception {
    4.     out.writeBytes(JSON.toJSONString(msg).getBytes(CharsetUtil.UTF_8));
    5. }

    服务端代码

    ProtocolServer

    1. /**
    2.  * * Netty自定义编码和解码 服务端
    3.  */
    4. public class ProtocolServer {
    7.     public static void main(String[] args) {
    9.         ProtocolServer server = new ProtocolServer();
    10.         server.openSever(6666);
    11.     }
    13.     public void openSever(int port) {
    15.         ServerBootstrap bootstrap = new ServerBootstrap();
    17.         EventLoopGroup bootGroup = new NioEventLoopGroup(1); //connect \accept \read \write
    18.         EventLoopGroup workGroup = new NioEventLoopGroup();
    20.         bootstrap.group(bootGroup, workGroup);
    21.         bootstrap.channel(NioServerSocketChannel.class);
    23.         bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
    25.             @Override
    26.             protected void initChannel(SocketChannel ch) throws Exception {
    28.                 ChannelPipeline pipeline = ch.pipeline();
    30.                 //解决粘包拆包
    31.                 pipeline.addLast(new LengthFieldBasedFrameDecoder(65535, 0, 4, 0, 4));
    32.                 pipeline.addLast(new LengthFieldPrepender(4));
    34.                 //自定义的编码和解码
    35.                 pipeline.addLast(new RpcDecoder());
    36.                 pipeline.addLast(new RpcEncoder());
    38.                 //业务处理handler
    39.                 pipeline.addLast(ProtocolServerHandler.INSTANCE);
    40.             }
    41.         });
    42.         try {
    44.             System.out.println("服务启动成功");
    45.             ChannelFuture f = bootstrap.bind(port).sync();
    46.             f.channel().closeFuture().sync();
    47.         } catch (InterruptedException e) {
    49.             e.printStackTrace();
    50.         } finally {
    52.             bootGroup.shutdownGracefully();
    53.             workGroup.shutdownGracefully();
    54.         }
    55.     }
    56. }

    ProtocolServerHandler

    1. import com.mytest.protocol.codec.RpcMessage;
    2. import io.netty.channel.ChannelHandler;
    3. import io.netty.channel.ChannelHandlerContext;
    4. import io.netty.channel.ChannelInboundHandlerAdapter;
    6. @ChannelHandler.Sharable
    7. public class ProtocolServerHandler extends ChannelInboundHandlerAdapter {
    10.     public static final ProtocolServerHandler INSTANCE = new ProtocolServerHandler();
    12.     @Override
    13.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    15.         System.out.println("服务端接收消息");
    17.         if (msg instanceof RpcMessage) {
    19.             RpcMessage rpcMessage = (RpcMessage) msg;
    20.             System.out.println("服务端接收到的消息：" + rpcMessage);
    22.             //ctx.write(msg);
    23.         }
    24.     }
    26.     @Override
    27.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    29.         ctx.flush();
    30.     }
    32.     @Override
    33.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    35.         cause.printStackTrace();
    36.     }
    37. }

    客户端

    ProtocolClient

    1. import com.mytest.protocol.codec.RpcDecoder;
    2. import com.mytest.protocol.codec.RpcEncoder;
    3. import com.mytest.protocol.handler.ProtocolClientHandler;
    4. import io.netty.bootstrap.Bootstrap;
    5. import io.netty.channel.ChannelFuture;
    6. import io.netty.channel.ChannelInitializer;
    7. import io.netty.channel.ChannelPipeline;
    8. import io.netty.channel.EventLoopGroup;
    9. import io.netty.channel.nio.NioEventLoopGroup;
    10. import io.netty.channel.socket.SocketChannel;
    11. import io.netty.channel.socket.nio.NioSocketChannel;
    12. import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
    13. import io.netty.handler.codec.LengthFieldPrepender;
    15. import java.io.IOException;
    18. /**
    19.  * Netty自定义编码和解码 客户端
    20.  */
    21. public class ProtocolClient {
    24.     public static void main(String[] args) throws IOException {
    26.         ProtocolClient client = new ProtocolClient("127.0.0.1", 6666);
    27.     }
    29.     public ProtocolClient(String host, int port) {
    31.         EventLoopGroup group = new NioEventLoopGroup(1);
    33.         Bootstrap bootstrap = new Bootstrap();
    34.         bootstrap.group(group);
    35.         bootstrap.channel(NioSocketChannel.class);
    36.         bootstrap.remoteAddress(host, port);
    37.         bootstrap.handler(new ChannelInitializer<SocketChannel>() {
    39.             @Override
    40.             protected void initChannel(SocketChannel ch) throws Exception {
    42.                 ChannelPipeline pipeline = ch.pipeline();
    44.                 //netty的粘包和拆包
    45.                 pipeline.addLast(new LengthFieldBasedFrameDecoder(65535, 0, 4, 0, 4));
    46.                 pipeline.addLast(new LengthFieldPrepender(4));
    48.                 //自定义的编码和解码
    49.                 pipeline.addLast(new RpcDecoder());
    50.                 pipeline.addLast(new RpcEncoder());
    52.                 //业务处理的handler
    53.                 pipeline.addLast(ProtocolClientHandler.INSTANCE);
    54.             }
    55.         });
    56.         try {
    58.             ChannelFuture f = bootstrap.connect().sync();
    59.             f.channel().closeFuture().sync();
    60.         } catch (Exception e) {
    62.             e.printStackTrace();
    63.         } finally {
    65.             System.out.println("连接关闭，资源释放");
    66.             group.shutdownGracefully();
    67.         }
    68.     }
    69. }

    ProtocolClientHandler

    1. import com.mytest.protocol.codec.RpcMessage;
    2. import com.mytest.protocol.codec.RpcMessageType;
    3. import io.netty.channel.ChannelHandler;
    4. import io.netty.channel.ChannelHandlerContext;
    5. import io.netty.channel.ChannelInboundHandlerAdapter;
    7. @ChannelHandler.Sharable
    8. public class ProtocolClientHandler extends ChannelInboundHandlerAdapter {
    11.     public static final ProtocolClientHandler INSTANCE = new ProtocolClientHandler();
    13.     @Override
    14.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
    17.         //调用远程，传输一个java对象
    18.         RpcMessage rpcMessage = new RpcMessage();
    19.         rpcMessage.setContent("消息");
    20.         rpcMessage.setSender("发送者");
    21.         StringBuffer stringBuffer = new StringBuffer();
    22.         for (int i=0; i<100; i++) {
    24.             stringBuffer.append("这是一个消息字名字的");
    25.         }
    26.         rpcMessage.setReceiver(stringBuffer.toString());
    27.         rpcMessage.setRpcMessageType(RpcMessageType.LOGIN);
    28.         rpcMessage.setTime(System.currentTimeMillis());
    30.         for (int i=0; i<20; i++) {
    32.             ctx.writeAndFlush(rpcMessage);
    33.         }
    34.         System.out.println("启动后消息发送完毕");
    35.     }
    37.     @Override
    38.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    40.         System.out.println("客户端接收");
    41.         if (msg instanceof RpcMessage) {
    43.             RpcMessage rpcMessage = (RpcMessage) msg;
    44.             System.out.println("客户端接收到的消息：" + rpcMessage);
    45.         }
    46.     }
    48.     @Override
    49.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    51.         ctx.flush();
    52.     }
    54.     @Override
    55.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    57.         cause.printStackTrace();
    58.         ctx.close();
    59.     }
    60. }

    编解码器

    RpcDecoder 、RpcEncoder

    1. /**
    2.  * 网络通信传输的 字节流数据 的解码器（字节流解码成程序能识别的数据（比如解码成一个java对象））
    3.  *
    4.  */
    5. public class RpcDecoder extends ByteToMessageDecoder {
    8.     @Override
    9.     protected void decode(ChannelHandlerContext ctx, ByteBuf byteBuf, List<Object> out) throws Exception {
    12.         //复制一份 字节流数据
    13.         ByteBuf frame = byteBuf.retainedDuplicate();
    15.         //把字节流数据转成字符串
    16.         final String content = frame.toString(CharsetUtil.UTF_8);
    18.         //把字符串通过fastjson转成java对象
    19.         RpcMessage message = JSON.parseObject(content, RpcMessage.class);
    21.         //把得到的java对象传给下一个handler
    22.         out.add(message);
    24.         //buf的数据已经读取过了，跳过已经读取的数据，更新一下buf的index下标
    25.         byteBuf.skipBytes(byteBuf.readableBytes());
    26.     }
    27. }
    29. import com.alibaba.fastjson.JSON;
    30. import io.netty.buffer.ByteBuf;
    31. import io.netty.channel.ChannelHandlerContext;
    32. import io.netty.handler.codec.MessageToByteEncoder;
    33. import io.netty.util.CharsetUtil;
    35. public class RpcEncoder extends MessageToByteEncoder<RpcMessage> {
    38.     @Override
    39.     protected void encode(ChannelHandlerContext ctx, RpcMessage msg, ByteBuf out) throws Exception {
    41.         out.writeBytes(JSON.toJSONString(msg).getBytes(CharsetUtil.UTF_8));
    42.     }
    43. }
    45. @Data
    46. public class RpcMessage {
    49.     private RpcMessageType rpcMessageType; //消息类型[LOGIN]或者[SYSTEM]或者[LOGOUT]
    51.     private long time;               //消息发送时间
    53.     private String sender;           //发送人
    55.     private String receiver;         //接收人
    57.     private Object content;          //消息内容
    59. }
    61. public enum RpcMessageType {
    64.     /**
    65.      * 系统消息
    66.      */
    67.     SYSTEM("SYSTEM"),
    69.     /**
    70.      * 登录消息
    71.      */
    72.     LOGIN("LOGIN"),
    74.     /**
    75.      * 登出消息
    76.      */
    77.     LOGOUT("LOGOUT"),
    79.     /**
    80.      * 调用消息
    81.      */
    82.     INVOKE("INVOKE");
    84.     private String name;
    86.     public static boolean isRpcMessageType(String content) {
    88.         return content.matches("^\\[(SYSTEM|LOGIN|LOGIN|INVOKE)\\]");
    89.     }
    91.     RpcMessageType(String name) {
    93.         this.name = name;
    94.     }
    96.     public String getName() {
    98.         return this.name;
    99.     }
    101.     public String toString() {
    103.         return this.name;
    104.     }
    105. }

发表评论

表情：
评论列表 （有 0 条评论，64人围观）

还没有评论，来说两句吧...

相关阅读

    相关 Netty 编解码器

    Netty 编解码 在网络中数据是以字节码二进制的形式传输的，所以我们在使用 Netty 传输数据的时候，需要将我们传输的数据转换为 二进制的形式 进行传输，所以不管是我

    梦里梦外;梦里梦外;/ 0 赞/ 324 阅读

    相关 Netty——编解码器

    什么是编解码器 每个网络应用程序都必须定义如何解析在两个节点之间来回传输的原始字节，以及如何将其和 目标应用程序的数据格式做相互转换。这种转换逻辑由编解码器处理，编解码器

    比眉伴天荒比眉伴天荒/ 0 赞/ 647 阅读