编写网络应用程序时，因为数据在网络中传输的都是二进制字节码数据，在发送数据时就需要编码，接收数据时就需要解码
codec(编码器)的组成部分有两个：decoder(解码器) 和 encoder(编码器)，encoder负责把业务数据转换成字节码数据，decoder负责把字节码数据转换成业务数据
编解码器都实现了ChannelInboundHandler或者ChannelOutboundHandler接口
image.png
StringDecoder继承关系

编码器与解码器需要放置在自定义业务处理Handler之前，当进行解码操作时（服务端收到数据），对PipeLine相当于入栈操作；进行编码操作时，对PipeLine相当于出栈操作
37VT%6`H{}5J~%}B22D(HBW.png
编码器与解码器顺序一
image.png
调用顺序二

不论解码器handler还是编码器handler，接收的消息类型必须与待处理的消息类型一致，否则该handler不会被执行；在解码器进行数据解码时，需要判断缓存区(ByteBuf)的数据是否足够，否则接收到的结果会期望结果可能不一致

常用编码器与解码器：

  1. 数据编解码器:
  2.     StringEnvoder                        对字符串编码
  3.     StringDecoder                        对字符串解码
  4.     ObjectEncoder                        只能对Java对象进行编码
  5.     ObjectDecoder                        只能对Java对象进行解码
  6.     HttpServerCodec                  Http服务编码&解码器（将消息解析为HttpRequest和HttpContent两个部分)
  9. 特殊编&解码器:
  10.     LineBasedFrameDecoder                        使用行位控制符（\n或者\r\n）作为分割符来解码数据
  11.     DelimiterBasedFrameDecoder            使用自定义特殊字符作为消息分割符来解码数据
  12.     LengthFieldBasedFrameDecoder        指定长度来标识整包信息达到自动处理粘包和半包问题
  15. 数据包较大时:
  16.     ZlibDecoder                        压缩编码器（数据较大时可以在发送时进行压缩，服务端需要指定对应压缩解码器）
  17.     ZlibEncoder                        压缩解码器
  20. 传输服务编解码器:
  21.     Protobuf                                谷歌编码解码器，支持非Java对象编码解码，适合PRC数据交换服务

HTTP编解码器：

使用 HttpServerCodec 对请求数据进行解析，请求数据将会被 HttpServerCodec 解析为 HttpRequest（请求头）和HttpContent（请求体）两个部分，最后使用 DefaultFullHttpResponse 返回结果，源码地址

  1. package http.coder;
  4. import io.netty.bootstrap.ServerBootstrap;
  5. import io.netty.buffer.ByteBuf;
  6. import io.netty.channel.*;
  7. import io.netty.channel.nio.NioEventLoopGroup;
  8. import io.netty.channel.socket.SocketChannel;
  9. import io.netty.channel.socket.nio.NioServerSocketChannel;
  10. import io.netty.handler.codec.FixedLengthFrameDecoder;
  11. import io.netty.handler.codec.http.*;
  12. import io.netty.util.CharsetUtil;
  14. import java.nio.charset.StandardCharsets;
  16. import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_LENGTH;
  19. /* Http编解码器使用案例 - 服务端 */
  20. public class HttpCoderServer {
  23.     public static void main(String[] args) throws Exception{
  25.         EventLoopGroup bossGroup = new NioEventLoopGroup(1);
  26.         EventLoopGroup workerGroup = new NioEventLoopGroup();
  28.         try {
  29.             ServerBootstrap serverBootstrap = new ServerBootstrap();
  30.             serverBootstrap.group(bossGroup,workerGroup)
  31.                     .channel(NioServerSocketChannel.class)
  32.                     .childHandler(new ChannelInitializer<SocketChannel>() {
  33.                         @Override
  34.                         protected void initChannel(SocketChannel ch) throws Exception {
  35.                             ChannelPipeline pipeline = ch.pipeline();
  36.                             pipeline.addLast(new HttpServerCodec());        //将消息分为HttpRequest请求头和HttpContent请求体两部分
  37.                             /* 只处理请求头数据 */
  38.                             pipeline.addLast(new SimpleChannelInboundHandler<HttpRequest>() {
  39.                                 @Override
  40.                                 protected void channelRead0(ChannelHandlerContext ctx, HttpRequest msg) throws Exception {
  41.                                     System.out.println("请求头内容为: "+msg);
  42.                                 }
  43.                             });
  44.                             /* 只处理请求体数据 */
  45.                             pipeline.addLast(new SimpleChannelInboundHandler<HttpContent>() {
  46.                                 @Override
  47.                                 protected void channelRead0(ChannelHandlerContext ctx, HttpContent msg) throws Exception {
  49.                                     ByteBuf buf = msg.content();
  50.                                     String s = buf.toString(CharsetUtil.UTF_8);
  51.                                     System.out.println("请求体内容为: "+s);
  53.                                     /* 返回响应结果 */
  54.                                     DefaultFullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_0, HttpResponseStatus.OK);
  55.                                     byte[] info = "数据已接收".getBytes(StandardCharsets.UTF_8);
  56.                                     response.content().writeBytes(info);                        //设置返回内容
  57.                                     response.headers().setInt(CONTENT_LENGTH,info.length);        //设置返回内容长度
  58.                                     ctx.writeAndFlush(response);
  60.                                 }
  61.                             });
  63.                         }
  64.                     });
  66.             ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();
  67.             channelFuture.channel().closeFuture().sync();
  69.         }finally {
  70.             bossGroup.shutdownGracefully();
  71.             workerGroup.shutdownGracefully();
  72.         }
  74.     }
  76. }

使用PostMan进行数据发送测试，服务端成功收到消息
image.png

实体类数据传输：

Netty 默认只能使用 byte 类型进行数据传输，当对象为实体类时无法直接通过强转进行解析，需要使用第三方编解码器进行操作后才能使用，常用方案有以下两种：

JBoss编解码器：

源码参考

引入依赖：

  1. <!-- Netty Java序列化框架marshalling -->
  2. <dependency>
  3.   <groupId>org.jboss.marshalling</groupId>
  4.   <artifactId>jboss-marshalling</artifactId>
  5.   <version>1.3.0.CR9</version>
  6. </dependency>
  7. <dependency>
  8.   <groupId>org.jboss.marshalling</groupId>
  9.   <artifactId>jboss-marshalling-serial</artifactId>
  10.   <version>1.3.0.CR9</version>
  11. </dependency>

JBoss解码器：

  1. package jboss;
  3. import io.netty.handler.codec.marshalling.*;
  4. import org.jboss.marshalling.MarshallerFactory;
  5. import org.jboss.marshalling.Marshalling;
  6. import org.jboss.marshalling.MarshallingConfiguration;
  8. /**
  9. * Marshalling工厂，Java实体类编解码器
  10. */
  11. public final class MarshallingCodeCFactory {
  13.     /**
  14.     * 创建Jboss Marshalling解码器MarshallingDecoder
  15.     * @return MarshallingDecoder
  16.     */
  17.     public static MarshallingDecoder buildMarshallingDecoder() {
  18.         //首先通过Marshalling工具类的精通方法获取Marshalling实例对象 参数serial标识创建的是java序列化工厂对象。
  19.         final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial");
  20.         //创建了MarshallingConfiguration对象，配置了版本号为5 
  21.         final MarshallingConfiguration configuration = new MarshallingConfiguration();
  22.         configuration.setVersion(5);
  23.         //根据marshallerFactory和configuration创建provider
  24.         UnmarshallerProvider provider = new DefaultUnmarshallerProvider(marshallerFactory, configuration);
  25.         //构建Netty的MarshallingDecoder对象，俩个参数分别为provider和单个消息序列化后的最大长度
  26.         MarshallingDecoder decoder = new MarshallingDecoder(provider, 1024 * 1024 * 1);
  27.         return decoder;
  28.     }
  30.     /**
  31.     * 创建Jboss Marshalling编码器MarshallingEncoder
  32.     * @return MarshallingEncoder
  33.     */
  34.     public static MarshallingEncoder buildMarshallingEncoder() {
  35.         final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial");
  36.         final MarshallingConfiguration configuration = new MarshallingConfiguration();
  37.         configuration.setVersion(5);
  38.         MarshallerProvider provider = new DefaultMarshallerProvider(marshallerFactory, configuration);
  39.         //构建Netty的MarshallingEncoder对象，MarshallingEncoder用于实现序列化接口的POJO对象序列化为二进制数组
  40.         MarshallingEncoder encoder = new MarshallingEncoder(provider);
  41.         return encoder;
  42.     }
  45. }

实体类：

  1. package jboss;
  3. import lombok.Data;
  4. import lombok.experimental.Accessors;
  6. import java.io.Serializable;
  9. /* Netty 传输数据实体类 */
  10. @Data
  11. @Accessors(chain = true)
  12. public class TranslatorData implements Serializable {
  14.     private String id;
  15.     private String name;
  16.     private String message;     //传输消息具体内容
  18. }

服务端：

  1. package jboss;
  4. import io.netty.bootstrap.ServerBootstrap;
  5. import io.netty.channel.*;
  6. import io.netty.channel.nio.NioEventLoopGroup;
  7. import io.netty.channel.socket.SocketChannel;
  8. import io.netty.channel.socket.nio.NioServerSocketChannel;
  9. import io.netty.handler.codec.string.StringDecoder;
  12. /* Jboss解析实体类-Netty服务端 */
  13. public class NettyServer {
  16.     public static void main(String[] args) throws InterruptedException {
  18.         EventLoopGroup bossGroup = new NioEventLoopGroup();
  19.         EventLoopGroup workGroup = new NioEventLoopGroup();
  21.         //启动器，负责组装Netty组件进行启动服务器
  22.         ServerBootstrap bootstrap = new ServerBootstrap();
  24.         try {
  25.             bootstrap.group(bossGroup,workGroup)
  26.                     .channel(NioServerSocketChannel.class)
  27.                     .option(ChannelOption.SO_BACKLOG,1024)
  28.                     .childHandler(new ChannelInitializer<SocketChannel>() {
  30.                         //连接建立后开始执行
  31.                         @Override
  32.                         protected void initChannel(SocketChannel ch) throws Exception {
  33.                             ChannelPipeline pipeline = ch.pipeline();
  34.                             //添加Java实体类编解码器
  35.                             pipeline.addLast(MarshallingCodeCFactory.buildMarshallingEncoder());
  36.                             pipeline.addLast(MarshallingCodeCFactory.buildMarshallingDecoder());
  37.                             //自定义Handler，通常实现子类ChannelInboundHandlerAdapter或SimpleChannelInboundHandler<参数>进行实现
  38.                             pipeline.addLast(new ChannelInboundHandlerAdapter(){
  39.                                 //当有读事件时执行
  40.                                 @Override
  41.                                 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  42.                                     TranslatorData info = (TranslatorData) msg;
  43.                                     System.out.println("ID："+info.getId()+"，名称："+info.getName()+"，消息："+info.getMessage());
  44.                                     //创建新数据
  45.                                     TranslatorData response = new TranslatorData();
  46.                                     response.setId("resp: " + info.getId());
  47.                                     response.setName("resp: " + info.getName());
  48.                                     response.setMessage("resp: " + info.getMessage());
  49.                                     //返回数据
  50.                                     ctx.writeAndFlush(response);
  51.                                 }
  52.                             });
  53.                         }
  55.                     });
  57.             System.out.println("服务器 is ready .....");
  58.             //启动服务器绑定一个端口并且同步，生成一个 ChannelFuture 对象
  59.             ChannelFuture future = bootstrap.bind(8765).sync();
  60.             //对关闭通道进行监听（当有关闭通道的消息时才进行监听）
  61.             future.channel().closeFuture().sync();
  63.         }finally {
  64.             bossGroup.shutdownGracefully();
  65.             workGroup.shutdownGracefully();
  66.         }
  69.     }
  72. }

客户端：

  1. package jboss;
  3. import io.netty.bootstrap.Bootstrap;
  4. import io.netty.channel.*;
  5. import io.netty.channel.nio.NioEventLoopGroup;
  6. import io.netty.channel.socket.SocketChannel;
  7. import io.netty.channel.socket.nio.NioSocketChannel;
  8. import io.netty.handler.logging.LogLevel;
  9. import io.netty.handler.logging.LoggingHandler;
  10. import io.netty.util.ReferenceCountUtil;
  12. /* Jboss解析实体类-Netty客户端 */
  13. public class NettyClient {
  16.    public static void main(String[] args) {
  18.        EventLoopGroup workGroup = new NioEventLoopGroup();
  19.        Bootstrap bootstrap = new Bootstrap();
  20.        try {
  22.            bootstrap.group(workGroup)
  23.                    .channel(NioSocketChannel.class)
  24.                    //表示缓存区动态调配（自适应）
  25.                    .option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT)
  26.                    .handler(new LoggingHandler(LogLevel.INFO))
  27.                    .handler(new ChannelInitializer<SocketChannel>() {
  28.                        @Override
  29.                        protected void initChannel(SocketChannel sc) throws Exception {
  30.                            ChannelPipeline pipeline = sc.pipeline();
  31.                            //添加Java实体类编解码器
  32.                            pipeline.addLast(MarshallingCodeCFactory.buildMarshallingEncoder());
  33.                            pipeline.addLast(MarshallingCodeCFactory.buildMarshallingDecoder());
  34.                            //添加自定义处理器
  35.                            pipeline.addLast(new ChannelInboundHandlerAdapter(){
  36.                                @Override
  37.                                public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  38.                                    TranslatorData response = (TranslatorData)msg;
  39.                                    System.out.println("Client端: id= " + response.getId() + ", name= " + response.getName() + ", message= " + response.getMessage());
  40.                                }
  41.                            });
  42.                        }
  43.                    });
  45.            //绑定端口，同步等等请求连接
  46.            ChannelFuture future = bootstrap.connect("127.0.0.1", 8765).sync();
  47.            Channel channel = future.channel();
  48.            System.out.println("客户端尝试连接....");
  51.            /*模拟数据发送*/
  52.            for(int i =0; i <10; i++){
  53.                TranslatorData request = new TranslatorData();
  54.                request.setId("" + i);
  55.                request.setName("请求消息名称 " + i);
  56.                request.setMessage("请求消息内容 " + i);
  57.                channel.writeAndFlush(request);
  58.            }
  60.        } catch (InterruptedException e) {
  61.            e.printStackTrace();
  62.        }
  64.    }
  67. }

测试：

image.png

Protobuf：

传统的数据传输是使用 Http + JSON 的方式，引入Netty后可以使用 TCP + Protobuf 的方式进行数据交互。需要将类定义为 protobuf 的 .proto 格式，protobuf会在编译阶段生成它能识别的Java对象进行传输使用。生成的文件将包含外部类和内部类，实现数据推送的为内部类数据，支持跨平台发送数据
image.png
客户端需要添加Proto编码器，服务端需要proto解码器
image.png
IDEA 对应插件

使用ProtoBuf需要先添加依赖：

  1.         <dependency>
  2.             <groupId>com.google.protobuf</groupId>
  3.             <artifactId>protobuf-java</artifactId>
  4.             <version>3.19.3</version>
  5.         </dependency>

单数据类型发送：

客户端与服务端都使用同一个实体类进行交互，源码地址

1、创建Protobuf传输对象
创建一个文件名为 StudentPOJO 的 proto 文件

  1. syntax = "proto3";                                                            //指定Protobuf版本
  2. option java_outer_classname = "StudentPOJO";    //文件名，同时也是外部类名
  3. //真正发送的实体类对象，内部参数需要参照 protobyf官方语言指南
  4. message Student{
  5.   int32 id = 1;   //相当于 private int id ，1相当于属性序号
  6.   string name = 2;
  7. }

2、编译文件
访问 GitHub下载 Proto-3.19.3-win 资源包，解压后对刚才编写的 proto 文件进行编译，编译后放入项目中

  1. protoc.exe --java_out=. 文件名.proto
  1. ![image.png](https://cdn.nlark.com/yuque/0/2022/png/21405095/1645091863467-74cc33f2-e975-447b-87e4-1ee5cecc90b6.png#clientId=u91c76975-c5b2-4&crop=0&crop=0&crop=1&crop=1&from=paste&height=183&id=ub582e2a3&margin=%5Bobject%20Object%5D&name=image.png&originHeight=275&originWidth=658&originalType=binary&ratio=1&rotation=0&showTitle=false&size=18382&status=done&style=stroke&taskId=u87cd738f-9eb3-4de8-aa3d-c5f41f4c5b9&title=&width=438.6666666666667)<br />**       编译proto文件为Java文件**

3、创建客户端与服务端
客户端添加上 protobuf 的编码器，服务端添加上 protobuf 的解码器即可

客户端：

  1. package protobuf;
  3. import io.netty.bootstrap.Bootstrap;
  4. import io.netty.channel.ChannelFuture;
  5. import io.netty.channel.ChannelInitializer;
  6. import io.netty.channel.ChannelPipeline;
  7. import io.netty.channel.EventLoopGroup;
  8. import io.netty.channel.nio.NioEventLoopGroup;
  9. import io.netty.channel.socket.SocketChannel;
  10. import io.netty.channel.socket.nio.NioSocketChannel;
  11. import io.netty.handler.codec.protobuf.ProtobufEncoder;
  14. /* 客户端代码 */
  15. public class NettyClient {
  18.     public static void main(String[] args) throws Exception {
  20.         //客户端需要一个事件循环组
  21.         EventLoopGroup group = new NioEventLoopGroup();
  23.         try {
  24.             //客户端启动对象，客户端使用的是 Bootstrap 而不是 ServerBootstrap
  25.             Bootstrap bootstrap = new Bootstrap();
  26.             //设置相关参数
  27.             bootstrap.group(group)                 //设置线程组
  28.                     .channel(NioSocketChannel.class)        //设置客户端通道的实现类
  29.                     .handler(new ChannelInitializer<SocketChannel>() {
  31.                         @Override
  32.                         protected void initChannel(SocketChannel ch) throws Exception {
  33.                             ChannelPipeline pipeline = ch.pipeline();
  34.                             pipeline.addLast("proEncoder",new ProtobufEncoder());    //自定义处理器
  35.                             pipeline.addLast(new NettyClientHandler());    //自定义处理器
  36.                         }
  38.                     });
  39.             System.out.println("客户端 ok...");
  40.             //启动客户端并连接到服务端
  41.             ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6666).sync();
  42.             //给关闭通道进行监听(关闭通道事件发生后触发)
  43.             channelFuture.channel().closeFuture().sync();
  44.         } finally {
  45.             group.shutdownGracefully();
  46.         }
  49.     }
  52. }

客户端Handler：

  1. package protobuf;
  3. import io.netty.buffer.ByteBuf;
  4. import io.netty.buffer.Unpooled;
  5. import io.netty.channel.ChannelHandlerContext;
  6. import io.netty.channel.ChannelInboundHandlerAdapter;
  7. import io.netty.util.CharsetUtil;
  9. import java.nio.charset.StandardCharsets;
  12. /* 自定义管道 Handler */
  13. public class NettyClientHandler extends ChannelInboundHandlerAdapter {
  16.     /* 当通道就绪就会触发该方法
  17.     *     利用ProtoBuf发送数据
  18.     *  */
  19.     @Override
  20.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  21.         StudentPOJO.Student student = StudentPOJO.Student.newBuilder().setId(10).setName("小明同学").build();
  22.         System.out.println("客户端发送Proto数据");
  23.         ctx.writeAndFlush(student);
  24.     }
  27.     /*当通道有读取数据事件时触发
  28.      *  参数一: ChannelHandlerContext 上下文对象，含有管道 PipeLine、通道
  29.      *  参数二: Object 客户端发送的数据，默认是Object需要转换
  30.      */
  31.     @Override
  32.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  34.         ByteBuf buf = (ByteBuf) msg;
  35.         System.out.println("服务器回复的消息: "+buf.toString(CharsetUtil.UTF_8));
  36.         System.out.println("服务器的地址: "+ctx.channel().remoteAddress());
  38.     }
  41.     /* 异常发生时触发 */
  42.     @Override
  43.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  44.        cause.printStackTrace();
  45.        ctx.close();
  46.     }
  49. }

服务端：

  1. package protobuf;
  3. import io.netty.bootstrap.ServerBootstrap;
  4. import io.netty.channel.*;
  5. import io.netty.channel.nio.NioEventLoopGroup;
  6. import io.netty.channel.socket.SocketChannel;
  7. import io.netty.channel.socket.nio.NioServerSocketChannel;
  8. import io.netty.handler.codec.protobuf.ProtobufDecoder;
  11. /* 服务端代码 */
  12. public class NettyServer {
  15.     public static void main(String[] args) throws InterruptedException {
  17.         /* 创建 BossGroup 和 WorkerGroup 线程组
  18.          *   BossGroup 只处理连接请求，真正的和客户端业务处理会交给 WorkGrouop 完成
  19.          *     bossGroup 和 workerGroup 含有的子线程的个数默认为 CPU最大线程数 X 2，可以手动指定
  20.          */
  21.         EventLoopGroup bossGroup = new NioEventLoopGroup(1);        //创建子线程的个数为1的 bossGroup
  22.         EventLoopGroup workerGroup = new NioEventLoopGroup(8);      //创建子线程的个数为8的 workerGroup
  23.         try {
  24.             //创建服务器端的启动对象，配置参数
  25.             ServerBootstrap bootstrap = new ServerBootstrap();
  27.             //使用链式参数配置启动参数
  28.             bootstrap.group(bossGroup,workerGroup)                          //设置两个线程组
  29.                     .channel(NioServerSocketChannel.class)                  //使用 NioSocketChannel 作为服务器通道实现
  30.                     .option(ChannelOption.SO_BACKLOG,120)             //设置线程队列等待连接的个数
  31.                     .childOption(ChannelOption.SO_KEEPALIVE,true)     //保持活动连接状态
  32.                     .childHandler(new ChannelInitializer<SocketChannel>() { //创建一个通道初始化对象
  34.                         //给PipeLine设置处理器
  35.                         @Override
  36.                         protected void initChannel(SocketChannel ch) throws Exception {
  37.                             ChannelPipeline pipeline = ch.pipeline();
  38.                             //指定ProtoBuf解码器对哪种对象进行解码
  39.                             pipeline.addLast("proDecoder",new ProtobufDecoder(StudentPOJO.Student.getDefaultInstance()));
  40.                             pipeline.addLast(new NettyServerHandler());    //将自定义处理器加入到PipeLine
  41.                         }
  43.                     });     //给 WorkerGroup 的 EventLoop 对应的管道设置处理器
  45.             System.out.println("服务器 is ready .....");
  47.             //启动服务器绑定一个端口并且同步，生成一个 ChannelFuture 对象
  48.             ChannelFuture cf = bootstrap.bind(6666).sync();
  49.             //给CF注册监听器，监控关心的事件
  50.             cf.addListener(new ChannelFutureListener() {
  51.                 @Override
  52.                 public void operationComplete(ChannelFuture channelFuture) throws Exception {
  53.                     if(cf.isSuccess()){
  54.                         System.out.println("监听端口成功");
  55.                     }else{
  56.                         System.out.println("监听端口失败");
  57.                     }
  58.                 }
  59.             });
  60.             //对关闭通道进行监听（当有关闭通道的消息时才进行监听）
  61.             cf.channel().closeFuture().sync();
  63.         } finally {
  64.             bossGroup.shutdownGracefully();     //关闭资源
  65.             workerGroup.shutdownGracefully();   //关闭
  66.         }
  69.     }
  72. }

服务端Handler：

  1. package protobuf;
  3. import io.netty.buffer.Unpooled;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.channel.ChannelInboundHandlerAdapter;
  7. import java.nio.charset.StandardCharsets;
  8. import java.time.LocalDateTime;
  9. import java.util.concurrent.TimeUnit;
  12. /** 自定义管道 Handler */
  13. public class NettyServerHandler extends ChannelInboundHandlerAdapter {
  15.     /*当有读取事件时该方法将被触发*/
  16.     @Override
  17.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  19.         //读取从客户端发送的StudentPojo.Student
  20.         StudentPOJO.Student student = (StudentPOJO.Student) msg;
  21.         System.out.println("客户端发送的数据 id="+student.getId()+" 名称: "+student.getName());
  23.     }
  26.     /* 数据读取完毕后触发
  27.      *   参数一: ChannelHandlerContext 上下文对象，含有管道 PipeLine、通道
  28.      */
  29.     @Override
  30.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
  32.         //对发送数据进行编码后写入到缓存并刷新
  33.         ctx.writeAndFlush(Unpooled.copiedBuffer("服务端连接成功,欢迎!"+LocalDateTime.now(),StandardCharsets.UTF_8));
  35.     }
  38.     /* 处理异常，一般为关闭通道 */
  39.     @Override
  40.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  41.         ctx.close();        //和 ctx.channel().close() 是一个意思
  42.     }
  46. }

测试：
image.png
服务端接收并解码客户端发送的Protobuf封装数据

多数据类型发送：

客户端可能发送不同的实体类对象，服务端不同实体类对象的业务操作逻辑也不相同
相较于单数据类型发送，需要再proto文件上添加枚举类来控制具体选择的 Java 对象
源码地址

1、创建ProtoBuf传输对象
创建一个文件名为 StudentPOJO 的 proto 文件

  1. syntax = "proto3";
  2. option optimize_for = SPEED;                      //加快解析
  3. option java_package = "protobuf.morepojo";        //指定生成到哪个包下
  4. option java_outer_classname = "MyStudentInfo";    //文件名，同时也是外部类名
  5. //真正发送的实体类对象，内部参数需要参照 protobyf官方语言指南
  7. //protobuf 可以使用 message 管理其他的message
  8. message MyMessage{
  10.     //定义一个枚举类型，根据枚举类型的值来判断取要取哪个对象实例
  11.     enum DataType{
  12.         StudentType = 0;  //在proto3要求enum内值的编号从0开始
  13.         TeacherType = 1;
  14.     }
  16.     DataType data_type = 1; //将枚举值定义为属性，利用data_type判断实际传的是哪个枚举类型，每个属性的编号在proto中必须是唯一的
  18.     //oneof表示实际调用dataBody时，只能选择其中的一个进行操作
  19.     oneof dataBody{
  20.         Student student = 2;
  21.         Teacher teacher = 3;
  22.     }
  24. }
  28. message Student{
  29.   int32 id = 1;   //相当于 private int id , 1相当于属性序号
  30.   string name = 2;
  31. }
  32. message Teacher{
  33.   string name = 1;
  34.   int32 age = 2;
  35. }

2、编译文件
访问 GitHub下载 Proto-3.19.3-win 资源包，解压后对刚才编写的 proto 文件进行编译，编译后放入项目中

  1. protoc.exe --java_out=. 文件名.proto
  1. ![image.png](https://cdn.nlark.com/yuque/0/2022/png/21405095/1645091863467-74cc33f2-e975-447b-87e4-1ee5cecc90b6.png#clientId=u91c76975-c5b2-4&crop=0&crop=0&crop=1&crop=1&from=paste&height=183&id=MBiIJ&margin=%5Bobject%20Object%5D&name=image.png&originHeight=275&originWidth=658&originalType=binary&ratio=1&rotation=0&showTitle=false&size=18382&status=done&style=stroke&taskId=u87cd738f-9eb3-4de8-aa3d-c5f41f4c5b9&title=&width=438.6666666666667)<br />**       编译proto文件为Java文件**

3、创建客户端和服务端
客户端代码和服务端代码与单数据类型发送案例中一致，只有Handler不同

客户端Handler：

  1. package protobuf.morepojo;
  3. import io.netty.buffer.Unpooled;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.channel.ChannelInboundHandlerAdapter;
  6. import io.netty.channel.SimpleChannelInboundHandler;
  7. import protobuf.StudentPOJO;
  9. import java.nio.charset.StandardCharsets;
  10. import java.time.LocalDateTime;
  13. /** 自定义管道 Handler
  14.  *      继承父类ChannelInboundHandlerAdapter的区别就是读取事件发生时的方法中的msg不会被指定成具体的类型
  15.  * */
  16. public class NettyServerHandler extends SimpleChannelInboundHandler<MyStudentInfo.MyMessage> {
  19.     /*当有读取事件时该方法将被触发*/
  20.     @Override
  21.     protected void channelRead0(ChannelHandlerContext ctx, MyStudentInfo.MyMessage msg) throws Exception {
  23.         //根据dataType来显示不同的信息
  24.         MyStudentInfo.MyMessage.DataType dataType = msg.getDataType();
  25.         if(dataType == MyStudentInfo.MyMessage.DataType.StudentType){
  27.             MyStudentInfo.Student student = msg.getStudent();
  28.             System.out.println("获取到的学生信息: "+student.getName()+" id: "+student.getId());
  30.         }else if(dataType == MyStudentInfo.MyMessage.DataType.TeacherType){
  32.             MyStudentInfo.Teacher teacher = msg.getTeacher();
  33.             System.out.println("获取到的教师信息: "+teacher.getName()+" 年龄: "+teacher.getAge());
  35.         }else{
  36.             System.out.println("传输的类型不正确");
  37.         }
  39.     }
  42.     /* 数据读取完毕后触发
  43.      *   参数一: ChannelHandlerContext 上下文对象，含有管道 PipeLine、通道
  44.      */
  45.     @Override
  46.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
  48.         //对发送数据进行编码后写入到缓存并刷新
  49.         ctx.writeAndFlush(Unpooled.copiedBuffer("服务端连接成功,欢迎!"+LocalDateTime.now(),StandardCharsets.UTF_8));
  51.     }
  54.     /* 处理异常，一般为关闭通道 */
  55.     @Override
  56.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  57.         ctx.close();        //和 ctx.channel().close() 是一个意思
  58.     }
  62. }

客户端Handler：

  1. package protobuf.morepojo;
  3. import io.netty.buffer.ByteBuf;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.channel.ChannelInboundHandlerAdapter;
  6. import io.netty.util.CharsetUtil;
  7. import protobuf.StudentPOJO;
  9. import java.util.Random;
  12. /* 自定义管道 Handler */
  13. public class NettyClientHandler extends ChannelInboundHandlerAdapter {
  16.     /* 当通道就绪就会触发该方法
  17.     *     利用ProtoBuf发送数据
  18.     *  */
  19.     @Override
  20.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  22.         /* 随机发送Student或Teacher对象*/
  23.         int random = new Random().nextInt(3);
  24.         MyStudentInfo.MyMessage myMessage= null;
  25.         if(random%2==0){
  26.             myMessage = MyStudentInfo.MyMessage.newBuilder()
  27.               .setDataType(MyStudentInfo.MyMessage.DataType.StudentType)
  28.               .setStudent(MyStudentInfo.MyMessage.newBuilder().getStudentBuilder().setId(998).setName("马超").build())
  29.               .build();
  30.         }else{
  31.             myMessage = MyStudentInfo.MyMessage.newBuilder()
  32.                     .setDataType(MyStudentInfo.MyMessage.DataType.StudentType)
  33.                     .setTeacher(MyStudentInfo.MyMessage.newBuilder().getTeacherBuilder().setAge(25).setName("诸葛老师"))
  34.                     .build();
  35.         }
  37.         ctx.writeAndFlush(myMessage);
  39.     }
  42.     /*当通道有读取数据事件时触发
  43.      *  参数一: ChannelHandlerContext 上下文对象，含有管道 PipeLine、通道
  44.      *  参数二: Object 客户端发送的数据，默认是Object需要转换
  45.      */
  46.     @Override
  47.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  49.         ByteBuf buf = (ByteBuf) msg;
  50.         System.out.println("服务器回复的消息: "+buf.toString(CharsetUtil.UTF_8));
  51.         System.out.println("服务器的地址: "+ctx.channel().remoteAddress());
  53.     }
  56.     /* 异常发生时触发 */
  57.     @Override
  58.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  59.        cause.printStackTrace();
  60.        ctx.close();
  61.     }
  64. }

测试：
服务端成功获取到客户端随机发送的数据
image.png

自定义编码器与解码器：

可以继承 ByteToMessageDecoder 或者 ReplayingDecode 实现自定义解码器。ByteToMessageDecoder 不会对消息进行拆分；ReplayingDecode 会自动对数据进行拆分，但需要指定参数，参数代表用户状态管理的类型，参数为void时代表不需要状态管理
通过继承 MessageToByteEncoder<数据类型> 实现自定义编码器，数据类型为channel中msg数据的类型
通过继承 ByteToMessageCodec<数据类型> 实现自定义编/解码器，数据类型为channel中msg数据的类型

ReplayingDecoder缺点：

1、并不是所有的ByteBuf操作都被支持，如果调用了一个不被支持的方法，将会抛出一个UnsupportedOperationException.
2、ReplayingDecoder在某些情况下可能稍慢于其父类ByteToMessageDecoder，例如网络缓慢并且消息格
式复杂时，消息会被拆成了多个碎片，速度变慢

案例：

使用自定义编码器与解码器对数据进行操作

自定义解码器：

  1. package inboundhandlerandoutboundhandler;
  3. import io.netty.buffer.ByteBuf;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.handler.codec.ReplayingDecoder;
  7. import java.util.List;
  10. /* 自定义解码器,参数代表用户状态管理的类型，其中void代表不需要状态管理 */
  11. public class MyByteToLongDecoder extends ReplayingDecoder<Void> {
  14.     /*  decode会根据接收的数据被调用多次，直到确定没有新的元素被泰诺健爱到List
  15.             或者是ByteBuf没有更多的可读字节为止
  16.      * @param ctx:  上下文对象
  17.      * @param in:   入栈的ByteBuf
  18.      * @param out:  List集合，将解码后的数据传给下一个ChannelInboundHandler进行处理，该处理器方法也会被调用多次
  19.      */
  20.     @Override
  21.     protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
  23.         System.out.println("MyByteToLongDecoder方法被调用");
  24.         out.add(in.readLong());
  26.     }
  29. }

自定义编码器：

  1. package inboundhandlerandoutboundhandler;
  3. import io.netty.buffer.ByteBuf;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.handler.codec.MessageToByteEncoder;
  8. /* 自定义编码器 */
  9. public class MyLongToByteEncoder extends MessageToByteEncoder<Long> {
  11.     //编码方法
  12.     @Override
  13.     protected void encode(ChannelHandlerContext ctx, Long msg, ByteBuf out) throws Exception {
  15.         System.out.println("MyLongToByteEncoder的Encode方法被调用");
  16.         System.out.println("编码后的Msg=" + msg);
  17.         out.writeLong(msg);
  19.     }
  21. }

服务端：

  1. package inboundhandlerandoutboundhandler;
  3. import io.netty.bootstrap.ServerBootstrap;
  4. import io.netty.channel.ChannelFuture;
  5. import io.netty.channel.ChannelInitializer;
  6. import io.netty.channel.ChannelPipeline;
  7. import io.netty.channel.EventLoopGroup;
  8. import io.netty.channel.nio.NioEventLoopGroup;
  9. import io.netty.channel.socket.SocketChannel;
  10. import io.netty.channel.socket.nio.NioServerSocketChannel;
  12. public class MyServer {
  14.     public static void main(String[] args) throws Exception{
  16.         EventLoopGroup bossGroup = new NioEventLoopGroup(1);
  17.         EventLoopGroup workerGroup = new NioEventLoopGroup();
  19.         try {
  20.             ServerBootstrap serverBootstrap = new ServerBootstrap();
  21.             serverBootstrap.group(bossGroup,workerGroup)
  22.                     .channel(NioServerSocketChannel.class)
  23.                     .childHandler(new ChannelInitializer<SocketChannel>() {
  24.                         @Override
  25.                         protected void initChannel(SocketChannel ch) throws Exception {
  26.                             ChannelPipeline pipeline = ch.pipeline();
  27.                             pipeline.addLast(new MyByteToLongDecoder());               //入站的handler进行解码 MyByteToLongDecoder
  28.                             pipeline.addLast(new MyLongToByteEncoder());               //出站的handler进行编码
  29.                             pipeline.addLast(new MyServerHandler());                   //自定义的handler 处理业务逻辑
  30.                         }
  31.                     });
  33.             ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();
  34.             channelFuture.channel().closeFuture().sync();
  35.         }finally {
  36.             bossGroup.shutdownGracefully();
  37.             workerGroup.shutdownGracefully();
  38.         }
  40.     }
  42. }

服务端Handler：

  1. package inboundhandlerandoutboundhandler;
  3. import io.netty.channel.ChannelHandlerContext;
  4. import io.netty.channel.SimpleChannelInboundHandler;
  6. /* 自定义服务端Handler */
  7. public class MyServerHandler extends SimpleChannelInboundHandler<Long> {
  10.     @Override
  11.     protected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception {
  12.         System.out.println("从客户端" + ctx.channel().remoteAddress() + " 读取到long " + msg);
  13.     }
  15.     /* 数据读取完毕后触发 */
  16.     @Override
  17.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
  18.         System.out.println("全部接收完毕，开始返回指定数据给客户端");
  19.         ctx.writeAndFlush(98765L);        //给客户端返回数据
  20.     }
  22.     /* 当有异常的触发 */
  23.     @Override
  24.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  25.         cause.printStackTrace();
  26.         ctx.close();
  27.     }
  30. }

客户端：

  1. package inboundhandlerandoutboundhandler;
  3. import io.netty.bootstrap.Bootstrap;
  4. import io.netty.channel.ChannelFuture;
  5. import io.netty.channel.ChannelInitializer;
  6. import io.netty.channel.ChannelPipeline;
  7. import io.netty.channel.EventLoopGroup;
  8. import io.netty.channel.nio.NioEventLoopGroup;
  9. import io.netty.channel.socket.SocketChannel;
  10. import io.netty.channel.socket.nio.NioSocketChannel;
  12. public class MyClient {
  15.     public static void main(String[] args)  throws  Exception{
  17.         EventLoopGroup group = new NioEventLoopGroup();
  19.         try {
  20.             Bootstrap bootstrap = new Bootstrap();
  21.             bootstrap.group(group).channel(NioSocketChannel.class)
  22.                     .handler(new ChannelInitializer<SocketChannel>() {
  23.                         @Override
  24.                         protected void initChannel(SocketChannel ch) throws Exception {
  25.                             ChannelPipeline pipeline = ch.pipeline();
  26.                             pipeline.addLast(new MyLongToByteEncoder());           //加入一个出站的handler 对数据进行一个编码
  27.                             pipeline.addLast(new MyByteToLongDecoder());           //这时一个入站的解码器(入站handler )
  28.                             pipeline.addLast(new MyClientHandler());               //加入一个自定义的handler处理业务
  29.                         }
  30.                     }); 
  32.             ChannelFuture channelFuture = bootstrap.connect("localhost", 7000).sync();
  33.             channelFuture.channel().closeFuture().sync();
  35.         }finally {
  36.             group.shutdownGracefully();
  37.         }
  39.     }
  42. }

客户端Handler：

  1. package inboundhandlerandoutboundhandler;
  3. import io.netty.buffer.Unpooled;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.channel.SimpleChannelInboundHandler;
  6. import io.netty.util.CharsetUtil;
  9. /* 自定义客户端Handler */
  10. public class MyClientHandler  extends SimpleChannelInboundHandler<Long> {
  13.     @Override
  14.     protected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception {
  15.         System.out.println("收到服务器消息=" + msg);
  16.     }
  19.     @Override
  20.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  21.         System.out.println("MyClientHandler 发送数据");
  22.         //ctx.writeAndFlush(123456L); //发送的是一个long
  23.         ctx.writeAndFlush(Unpooled.copiedBuffer("abcdefghijklmnop", CharsetUtil.UTF_8));
  24.     }
  26. }

测试：

启动客户端与服务端，客户端将消息编码后发送给服务端，服务端收到消息后解码进行输出；服务端收到全部消息后返回数据到编码器，编码器编码后再传递给客户端
image.png
客户端与服务端成功使用自定义编码器与解码器