netty 4.1.45 protobuf 编解码实现

迈不过友情╰ 2023-07-04 06:23 2阅读 0赞

### 文章目录 ###

*  PB协议简单使用
 *  netty使用PB协议进行编解码
 *   *  定义请求响应协议
 *  服务端代码编写
 *  客户端改造
 *  运行结果

> **注：更多netty相关文章请访问博主专栏： [netty专栏][netty]**

# PB协议简单使用 #

关于PB协议的使用介绍请移步至： [protocol buffer 3 (Protobuf3) ( java 版本 ) 使用入门][protocol buffer 3 _Protobuf3_ _ java _ _]。里面介绍了如何使用PB协议。

# netty使用PB协议进行编解码 #

## 定义请求响应协议 ##

*  请求 SubscribeReq.proto

syntax="proto3";
    option java_package = "com.example.pojo";
    option java_outer_classname = "SubscribeReqProto";
     
    message SubscribeReq {
    	int32 subReqId = 1;
    	string userName = 2;
    	string productName = 3;
    	repeated string address = 4;
    }

*  响应 SubscribeResp.proto

syntax="proto2";
    option java_package="com.example.pojo";
    option java_outer_classname="SubscribeRespProto";
     
    message SubscribeResp{
    	required int32 subReqId = 1;
    	required int32 respCode = 2;
    	required string desc = 3;		
    }

使用`protoc --java_out {Java文件地址} {proto文件地址}`编译成Java文件。将Java文件拷贝到工程目录下。文件代码非常多，请读者自行编译查看。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70]

# 服务端代码编写 #

还是在之前的代码基础上进行改造。

最主要的就是添加PB协议的编解码器，netty已经为我们封装好了。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 1]

*  ProtobufVarint32FrameDecoder 半包处理
 *  ProtobufDecoder 只负责解码，不负责拆包粘包处理，所以在这之前要添加ProtobufVarint32FrameDecoder
 *  ProtobufEncoder：PB协议编码器

然后就是使用`SubscribeReqProto.SubscribeReq`来进行消息的传输。代码的编写还是比较简单的，完整的代码如下：

package com.example;
    
    import com.example.pojo.SubscribeReqProto;
    import com.example.pojo.SubscribeRespProto;
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.handler.codec.ByteToMessageDecoder;
    import io.netty.handler.codec.DelimiterBasedFrameDecoder;
    import io.netty.handler.codec.LineBasedFrameDecoder;
    import io.netty.handler.codec.MessageToByteEncoder;
    import io.netty.handler.codec.protobuf.ProtobufDecoder;
    import io.netty.handler.codec.protobuf.ProtobufEncoder;
    import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
    import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
    import io.netty.handler.codec.string.StringDecoder;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    import java.util.List;
    
    /** * 自定义消息解码器 */
    public class MyNettyServerWithPB { 
        public static void main(String[] args) { 
            //配置服务端NIO线程组
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workGroup = new NioEventLoopGroup();
    
            try { 
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap.group(bossGroup, workGroup)//配置主从线程组
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG, 128)//配置一些TCP的参数
                        .childHandler(new MyChildHandlerWithPB());//添加自定义的channel
                //绑定8080端口
                ChannelFuture channelFuture = serverBootstrap.bind(8080).sync();
                //服务端监听端口关闭
                ChannelFuture future = channelFuture.channel().closeFuture().sync();
            } catch (Exception e) { 
                e.printStackTrace();
            } finally { 
                //netty优雅停机
                bossGroup.shutdownGracefully();
                workGroup.shutdownGracefully();
            }
        }
    }
    
    class MyChildHandlerWithPB extends ChannelInitializer<SocketChannel> { 
        @Override
        protected void initChannel(SocketChannel socketChannel) { 
            socketChannel.pipeline()
                    .addLast(new ProtobufVarint32FrameDecoder())//用于半包处理
                    //ProtobufDecoder解码器，参数是SubscribeReq，也就是需要接收到的消息解码为SubscribeReq类型的对象
                    .addLast(new ProtobufDecoder(SubscribeReqProto.SubscribeReq.getDefaultInstance()))
                    .addLast(new ProtobufVarint32LengthFieldPrepender())
                    .addLast(new ProtobufEncoder())
                    .addLast(new TimeWithPBServerHandler());
        }
    }
    
    
    /** * TimeServerHandler这个才是服务端真正处理请求的服务方法 */
    class TimeWithPBServerHandler extends ChannelInboundHandlerAdapter { 
    
        private int count = 0;
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
            SubscribeReqProto.SubscribeReq req = (SubscribeReqProto.SubscribeReq) msg;
    
            SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            String currentTimeStr = "TOM".equalsIgnoreCase(req.getUserName()) ?
                    format.format(new Date()) + "" : "BAD ORDER";
    
            //受到一次请求，count加1
            System.out.println("第 " + count++ + " 次收到客户端请求：" + req.toString() + " 返回响应：" + currentTimeStr);
    
            SubscribeRespProto.SubscribeResp.Builder resp = SubscribeRespProto.SubscribeResp.newBuilder();
            resp.setRespCode(200)
                    .setSubReqId(req.getSubReqId())
                    .setDesc(currentTimeStr);
    
            ctx.write(resp);//将消息发送到发送缓冲区
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) { 
            ctx.flush();//将消息从发送缓冲区中写入socketchannel中
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 
            ctx.close();
        }
    }

# 客户端改造 #

客户端同理也是要配置编解码器  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 2]

完整代码如下：

package com.example;
    
    import com.example.pojo.SubscribeReqProto;
    import com.example.pojo.SubscribeRespProto;
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.handler.codec.protobuf.ProtobufDecoder;
    import io.netty.handler.codec.protobuf.ProtobufEncoder;
    import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
    import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
    
    import java.util.Arrays;
    import java.util.List;
    import java.util.UUID;
    
    /** * 自定义消息解码器 */
    public class MyNettyClientWithPB { 
        public static void main(String[] args) { 
            //客户端NIO线程组
            EventLoopGroup group = new NioEventLoopGroup();
    
            try { 
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.group(group)
                        .channel(NioSocketChannel.class)
                        .handler(new ChannelInitializer<SocketChannel>() { 
                            @Override
                            protected void initChannel(SocketChannel ch) { 
                                ch.pipeline()
                                        .addLast(new ProtobufVarint32FrameDecoder())
                                        .addLast(new ProtobufDecoder(SubscribeRespProto.SubscribeResp.getDefaultInstance()))
                                        .addLast(new ProtobufVarint32LengthFieldPrepender())
                                        .addLast(new ProtobufEncoder())
                                        .addLast(new TimeWithPBClientHandler());
    
                            }
                        });
    
                //异步链接服务器
                ChannelFuture future = bootstrap.connect("127.0.0.1", 8080).sync();
                //等等客户端链接关闭
                future.channel().closeFuture().sync();
            } catch (Exception e) { 
                e.printStackTrace();
            } finally { 
                //优雅停机
                group.shutdownGracefully();
            }
        }
    }
    
    
    //客户端业务逻辑处理类
    class TimeWithPBClientHandler extends ChannelInboundHandlerAdapter { 
        private int count = 0;
    
        /** * 客户端与服务器TCP链路链接成功后调用该方法 * * @param ctx */
        @Override
        public void channelActive(ChannelHandlerContext ctx) { 
            for (int i = 0; i < 100; i++) { 
                List<String> list = Arrays.asList(new String[]{ "杭州","北京"});
                SubscribeReqProto.SubscribeReq.Builder req = SubscribeReqProto.SubscribeReq.newBuilder();
                req.setSubReqId(i)
                        .setUserName("TOM")
                        .setProductName(UUID.randomUUID().toString())
                        .addAllAddress(list);
    
                ctx.writeAndFlush(req);//写入缓冲并且发送到socketchannel
            }
        }
    
        /** * 读取到服务端相应后执行该方法 * * @param ctx * @param msg * @throws Exception */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) { 
            SubscribeRespProto.SubscribeResp resp = (SubscribeRespProto.SubscribeResp) msg;
            System.out.println("第 " + count++ + " 次受到服务端返回：" + resp.getDesc());
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 
            System.out.println("Unexpected exception from downstream : " + cause.getMessage());
            ctx.close();
        }
    }

# 运行结果 #

依次启动服务端和客户端，运行结果如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 3]

> **注：更多netty相关文章请访问博主专栏： [netty专栏][netty]**

[netty]: https://blog.csdn.net/fgyibupi/category_9705823.html
[protocol buffer 3 _Protobuf3_ _ java _ _]: https://blog.csdn.net/liubenlong007/article/details/96715004
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200209171329152.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200209171707591.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200209171933352.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200209172504946.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70