使用Netty框架完成客户端和服务端收发Protobuf消息

前言

本周继续学习尼恩编著的《Netty、Redis、ZooKeeper高并发实战》，一些资源也贴在这里，自己以后想看还可以找到，这个是在博客园的一个入口https://www.cnblogs.com/crazymakercircle/p/9904544.html。
这周主要学习了Netty客户端和服务端通信，书是由浅入深的在进行，从Socket NOI通信到 Reactor反应器模式，再到Netty框架，示例代码都在https://gitee.com/crazymaker/netty_redis_zookeeper_source_code.git 中可以看到，书结合源代码，自己在动手试验一下，感觉还是有些收获。 今天的示例代码就是实践出一个客户端和服务端传递protobuf的例子。
Netty 服务端

先来看一下服务端代码，

1. @Slf4j
2. public class ProtoBufServer {
4.     private final int serverPort;
5.     ServerBootstrap b = new ServerBootstrap();
7.     public ProtoBufServer(int port) {
8.         this.serverPort = port;
9.     }
11.     public void runServer() {
12.         //创建reactor 线程组
13.         EventLoopGroup bossLoopGroup = new NioEventLoopGroup(1);
14.         EventLoopGroup workerLoopGroup = new NioEventLoopGroup();
16.         try {
17.             //1 设置reactor 线程组
18.             b.group(bossLoopGroup, workerLoopGroup);
19.             //2 设置nio类型的channel
20.             b.channel(NioServerSocketChannel.class);
21.             //3 设置监听端口
22.             b.localAddress(serverPort);
23.             //4 设置通道的参数
24.             b.option(ChannelOption.SO_KEEPALIVE, true);
25.             b.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
26.             b.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
28.             //5 装配子通道流水线
29.             b.childHandler(new ChannelInitializer<SocketChannel>() {
30.                 //有连接到达时会创建一个channel
31.                 protected void initChannel(SocketChannel ch) throws Exception {
32.                     // pipeline管理子通道channel中的Handler
33.                     // 向子channel流水线添加3个handler处理器
35.                     // protobufDecoder仅仅负责编码，并不支持读半包，所以在之前，一定要有读半包的处理器。
36.                     // 有三种方式可以选择：
37.                     // 使用netty提供ProtobufVarint32FrameDecoder
38.                     // 继承netty提供的通用半包处理器 LengthFieldBasedFrameDecoder
39.                     // 继承ByteToMessageDecoder类，自己处理半包
41.                     // 半包的处理
42.                     ch.pipeline().addLast(new ProtobufVarint32FrameDecoder());
43.                     // 需要解码的目标类
44.                     ch.pipeline().addLast(new ProtobufDecoder(MarketPriceProto.MarketPrice.getDefaultInstance()));
45.                     ch.pipeline().addLast(new ProtobufBussinessHandler());
46.                 }
47.             });
48.             // 6 开始绑定server
49.             // 通过调用sync同步方法阻塞直到绑定成功
50.             ChannelFuture channelFuture = b.bind().sync();
51.             log.info(" 服务器启动成功，监听端口: " +
52.                     channelFuture.channel().localAddress());
54.             // 7 等待通道关闭的异步任务结束
55.             // 服务监听通道会一直等待通道关闭的异步任务结束
56.             ChannelFuture closeFuture = channelFuture.channel().closeFuture();
57.             closeFuture.sync();
58.         } catch (Exception e) {
59.             e.printStackTrace();
60.         } finally {
61.             // 8 优雅关闭EventLoopGroup，
62.             // 释放掉所有资源包括创建的线程
63.             workerLoopGroup.shutdownGracefully();
64.             bossLoopGroup.shutdownGracefully();
65.         }
67.     }
69.     //服务器端业务处理器
70.     static class ProtobufBussinessHandler extends ChannelInboundHandlerAdapter {
71.         @Override
72.         public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
73.             MarketPriceProto.MarketPrice protoMsg = (MarketPriceProto.MarketPrice) msg;
74.             //经过pipeline的各个decoder，到此Person类型已经可以断定
75.             log.info("收到一个 MsgProtos.Msg 数据包 =》");
76.             log.info("protoMsg.getId():=" + protoMsg.getId());
77.             log.info("protoMsg.getClose():=" + protoMsg.getClose());
78.         }
79.     }
82.     public static void main(String[] args) throws InterruptedException {
83.         int port = SERVER_PORT;
84.         new ProtoBufServer(port).runServer();
85.     }
86. }

复制代码

代码中有注释解释，我在这里加一下说明
代码中有两个EventLoopGroup bossLoopGroup和EventLoopGroup workerLoopGroup，使用两个是什么原因呢？ 一个是负责处理连接监听事件， 一个负责处理数据IO事件和Handler业务处理，通俗点解释就是一个负责接客，一个负责服务客户。如果只有一个人就会忙不过来，让后面的人等很久。
b.childHandler这个就是我们具体的如何处理接收到的消息，他们都继承ChannelInboundHandlerAdapter，通过PipeLine把消息进行处理。我们从通道里面拿到的都是字节码，那么要转成我们需要的Protobuf类，就需要用到这些处理类
前两个处理类ProtobufVarint32FrameDecoder和ProtobufDecoder都是Netty提供的，一个是为了解决半包问题，半包问题是因为在TCP传输的时候对数据包进行了拆包或者分包，收到的时候如果直接处理，就会有问题，需要我们在应用层进行二次封装。 在这个示例中如果我们不使用ProtobufVarint32FrameDecoder，客户端也不用，那么就会出现有的可以解析出来，有的报错的情况： com.google.protobuf.InvalidProtocolBufferException: Protocol message contained an invalid tag (zero).
可以用搬家来打比分，我们把一个家从一个地方搬到另外一个地方，还要求布局一样，汽车运输的时候可能要分好几次，那么我们可以先记住位置，然后随便装车搬过去，过去后先暂存，再按记录的位置进行还原，这样来保证一模一样。
ProtobufBussinessHandler 是我们自定义的Handler继承了ChannelInboundHandlerAdapter，通过它我们拿到channel里面的数据，转换成我们的具体的Protobuf对象。这里只是简单的打印出来。 如果要把它继续传下去，需要调用 super.channelRead(ctx,msg)传递下去。
客户端

1. @Slf4j
2. public class ProtoBufScanClient {
3.     private int serverPort;
4.     private String serverIp;
5.     Bootstrap b = new Bootstrap();
7.     public ProtoBufScanClient(String ip, int port) {
8.         this.serverPort = port;
9.         this.serverIp = ip;
10.     }
12.     public void runClient() {
13.         //创建reactor 线程组
14.         EventLoopGroup workerLoopGroup = new NioEventLoopGroup();
16.         try {
17.             //1 设置reactor 线程组
18.             b.group(workerLoopGroup);
19.             //2 设置nio类型的channel
20.             b.channel(NioSocketChannel.class);
21.             //3 设置监听端口
22.             b.remoteAddress(serverIp, serverPort);
23.             //4 设置通道的参数
24.             b.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
26.             //5 装配通道流水线
27.             b.handler(new ChannelInitializer<SocketChannel>() {
28.                 //初始化客户端channel
29.                 protected void initChannel(SocketChannel ch) throws Exception {
30.                     // 客户端channel流水线添加2个handler处理器
31.                     ch.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
32.                     ch.pipeline().addLast(new ProtobufEncoder());
33.                 }
34.             });
35.             ChannelFuture f = b.connect();
36.             f.addListener((ChannelFuture futureListener) ->
37.             {
38.                 if (futureListener.isSuccess()) {
39.                     log.info("EchoClient客户端连接成功!");
41.                 } else {
42.                     log.info("EchoClient客户端连接失败!");
43.                 }
45.             });
47.             // 阻塞,直到连接完成
48.             f.sync();
49.             Channel channel = f.channel();
52.             Scanner scanner = new Scanner(System.in);
53.             log.info("请输入发送内容:");
56.             GenericFutureListener sendCallBack = new GenericFutureListener() {
58.                 @Override
59.                 public void operationComplete(Future future) throws Exception {
60.                     if (future.isSuccess()) {
61.                         log.info("发送成功!");
63.                     } else {
64.                         log.info("发送失败!");
65.                     }
66.                 }
67.             };
70.             while (scanner.hasNext()) {
71.                 //获取输入的内容
72.                 String next = scanner.next();
73.                 String[] values = next.split(",");
74.                 if(values.length != 5)
75.                 {
76.                     log.info("格式不正确!");
77.                 }
78.                 else {
79.                     MarketPriceProto.MarketPrice msg = build(values);
80.                     ChannelFuture writeAndFlushFuture = channel.writeAndFlush(msg);
81.                     writeAndFlushFuture.addListener(sendCallBack);
82.                 }
84.                 log.info("请输入发送内容:");
86.             }
89.             channel.flush();
92.             // 7 等待通道关闭的异步任务结束
93.             // 服务监听通道会一直等待通道关闭的异步任务结束
94.             ChannelFuture closeFuture = channel.closeFuture();
95.             closeFuture.sync();
97.         } catch (Exception e) {
98.             e.printStackTrace();
99.         } finally {
100.             // 优雅关闭EventLoopGroup，
101.             // 释放掉所有资源包括创建的线程
102.             workerLoopGroup.shutdownGracefully();
103.         }
105.     }
107.     //构建ProtoBuf对象
108.     public MarketPriceProto.MarketPrice build(String[] values) {
110.         MarketPriceProto.MarketPrice.Builder builder = MarketPriceProto.MarketPrice.newBuilder();
112.         builder.setId(values[0]);
115.         builder.setOpen(Double.valueOf(values[1]));
116.         builder.setHigh(Double.valueOf(values[2]));
117.         builder.setLow(Double.valueOf(values[3]));
118.         builder.setClose(Double.valueOf(values[4]));
120.         return builder.build();
121.     }
123.     public static void main(String[] args) throws InterruptedException {
124.         int port = SERVER_PORT;
125.         String ip = SOCKET_SERVER_IP;
126.         new ProtoBufScanClient(ip, port).runClient();
127.     }
128. }

复制代码

Netty客户端代码和服务端很接近，这里它只用了一个线程组，客户端只有它一个使用，和服务端模式不一样，一个就够了。
客户端读取控制台输入数据，然后构造成MarketPriceProto.MarketPrice，它的Proto定义如下

1. // [开始声明]
2. syntax = "proto3";
3. //定义protobuf的包名称空间
4. package com.ken.netty.protocol;
5. // [结束声明]
7. // [开始 java 选项配置]
8. option java_package = "com.ken.netty.protocol";
9. option java_outer_classname = "MarketPriceProto";
10. // [结束 java 选项配置]
12. // [开始 消息定义]
13. message MarketPrice {
14.   string id = 1;
15.   double open = 2;
16.   double high = 3;
17.   double low = 4;
18.   double close = 5;
19. }

复制代码

channel.writeAndFlush（msg）,  这里我们直接把Protobuf对象传递进channel, ProtobufEncoder会对它进行编码。
总结

代码比较简单，这是在有书的帮助和源代码帮助下实现的，不通过书这个也不是这么容易理解的。 作者还实现了一个及时通信的例子https://gitee.com/crazymaker/SimpleCrayIM。  也需要花些时间来学习一下。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！