Netty 拆包粘包和服务启动流程分析

川长思鸟来 2022-06-01 12:38 188阅读 0赞

![在这里插入图片描述][aHR0cHM6Ly9pbWcyMDIwLmNuYmxvZ3MuY29tL2Jsb2cvODA2OTU2LzIwMjAwNy84MDY5NTYtMjAyMDA3MzEyMzQ1MDc3MjItMTExNjgwOTkyMi5naWY]

# Netty 拆包粘包和服务启动流程分析 #

通过本章学习，笔者希望你能掌握EventLoopGroup的工作流程，ServerBootstrap的启动流程，ChannelPipeline是如何操作管理Channel。只有清楚这些，才能更好的了解和使用Netty。还在等什么，快来学习吧！

知识结构图：

![Netty][]

技术：Netty，拆包粘包，服务启动流程  
说明：若你对NIO有一定的了解，对于本章知识来说有很大的帮助！[NIO教程][NIO]  
源码：https://github.com/ITDragonBlog/daydayup/tree/master/Netty/netty-stu

## Netty 重要组件 ##

这里让你清楚了解 ChannelPipeline，ChannelHandlerContext，ChannelHandler，Channel 四者之间的关系。  
这里让你清楚了解 NioEventLoopGroup，NioEventLoop，Channel 三者之间的关系。  
这里让你清楚了解 ServerBootstrap，Channel 两者之间的关系。  
看懂了这块的理论知识，后面Netty拆包粘包的代码就非常的简单。

### Channel ###

**Channel** : Netty最核心的接口。NIO通讯模式中通过Channel进行Socket套接字的读，写和同时读写操作。  
**ChannelHandler** : 因为直接使用Channel会比较麻烦，所以在Netty编程中通过ChannelHandler间接操作Channel，从而简化开发。  
**ChannelPipeline** : 可以理解为一个管理ChandlerHandler的链表。对Channel进行操作时，Pipeline负责从尾部依次调用每一个Handler进行处理。每个Channel都有一个属于自己的ChannelPipeline。  
**ChannelHandlerContext** : ChannelPipeline通过ChannelHandlerContext间接管理每个ChannelHandler。

如下图所示，结合代码，在服务器初始化和客户端创建连接的过程中加了四个Handler，分别是日志事务，字符串分割解码器，接受参数转字符串解码器，处理任务的Handler。  
![format_png][]

### NioEventLoopGroup ###

**EventLoopGroup** : 本质是个线程池，继承了ScheduledExecutorService 定时任务线程池。  
**NioEventLoopGroup** : 是用来处理NIO通信模式的线程池。每个线程池有N个NioEventLoop来处理Channel事件，每一个NioEventLoop负责处理N个Channel。  
**NioEventLoop** : 负责不停地轮询IO事件，处理IO事件和执行任务，类比多路复用器，细化分三件事。  
1 轮询注册到Selector上所有的Channel的IO事件  
2 处理产生网络IO事件的Channel  
3 处理队列中的任务

![format_png 1][]

### ServerBootstrap ###

本章重点，Netty是如何通过NIO辅助启动类来初始化Channel的？先看下面的源码。

@Override
    void init(Channel channel) throws Exception { 
    	final Map<ChannelOption<?>, Object> options = options0();
    	synchronized (options) { 
    		setChannelOptions(channel, options, logger);
    	}
    	final Map<AttributeKey<?>, Object> attrs = attrs0();
    	synchronized (attrs) { 
    		for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) { 
    			@SuppressWarnings("unchecked")
    			AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();
    			channel.attr(key).set(e.getValue());
    		}
    	}
    	ChannelPipeline p = channel.pipeline();
    	final EventLoopGroup currentChildGroup = childGroup;
    	final ChannelHandler currentChildHandler = childHandler;
    	final Entry<ChannelOption<?>, Object>[] currentChildOptions;
    	final Entry<AttributeKey<?>, Object>[] currentChildAttrs;
    	synchronized (childOptions) { 
    		currentChildOptions = childOptions.entrySet().toArray(newOptionArray(childOptions.size()));
    	}
    	synchronized (childAttrs) { 
    		currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(childAttrs.size()));
    	}
    	p.addLast(new ChannelInitializer<Channel>() { 
    		@Override
    		public void initChannel(final Channel ch) throws Exception { 
    			final ChannelPipeline pipeline = ch.pipeline();
    			ChannelHandler handler = config.handler();
    			if (handler != null) { 
    				pipeline.addLast(handler);
    			}
    			ch.eventLoop().execute(new Runnable() { 
    				@Override
    				public void run() { 
    					pipeline.addLast(new ServerBootstrapAcceptor(
    							ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
    				}
    			});
    		}
    	});
    }

服务器启动和连接过程：  
第一步：是给Channel设置options和attrs，  
第二步：复制childGroup，childHandler，childOptions和childAttrs等待服务器和客户端连接，  
第三步：实例化一个ChannelInitializer，添加到Pipeline的末尾。  
第四步：当Channel注册到NioEventLoop时，ChannelInitializer触发initChannel方法，pipeline装入自定义的Handler，给Channel设置一下child配置。

小结：  
1 **group，options，attrs，handler，是在服务器端初始化时配置，是AbstractBootstrap的方法。**  
2 **childGroup，childOption，childAttr，childHandler，是在服务器与客户端建立Channel后配置，是ServerBootstrap的方法。**  
3 **Bootstrap 和 ServerBootstrap 都继承了AbstractBootstrap类。**  
4 **若不设置childGroup，则默认取group值。**  
5 **Bootstrap 和 ServerBootstrap 启动服务时，都会执行验证方法，判断必填参数是否都有配置。**

![format_png 2][]

## Netty 拆包粘包 ##

这里通过介绍Netty拆包粘包问题来对Netty进行入门学习。  
在基于流的传输中，即便客户端发送独立的数据包，操作系统也会将其转换成**一串字节队列**，而服务端一次**读取到的字节数又不确定**。再加上网络传输的快慢。服务端很难完整的接收到数据。  
常见的拆包粘包方法有三种  
1 **服务端设置一次接收字节的长度**。若服务端接收的字节长度不满足要求则一直处于等待。客户端为满足传输的字节长度合格，可以考虑使用空格填充。  
2 **服务端设置特殊分隔符**。客户端通过特殊分隔符粘包，服务端通过特殊分隔符拆包。  
3 **自定义协议**。数据传输一般分消息头和消息体，消息头中包含了数据的长度。服务端先接收到消息头，得知需要接收N个数据，然后服务端接收直到数据为N个为止。  
本章采用第二种，用特殊分隔符的方式。

### 创建服务端代码流程 ###

第一步：准备两个线程池。一个用于接收事件的boss线程池，另一个用于处理事件的worker线程池。  
第二步：服务端实例化ServerBootstrap NIO服务辅助启动类。用于简化提高开发效率。  
第三步：配置服务器启动参数。比如channel的类型，接收channel的EventLoop，初始化的日志打印事件，建立连接后的事件（拆包，对象转字符串，自定义事件），初始化的配置和建立连接后的配置。  
第四步：绑定端口，启动服务。Netty会根据第三步配置的参数启动服务。  
第五步：关闭资源。

package com.itdragon.delimiter;
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.handler.codec.DelimiterBasedFrameDecoder;
    import io.netty.handler.codec.FixedLengthFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.logging.LogLevel;
    import io.netty.handler.logging.LoggingHandler;  
    public class ITDragonServer {   
          
        private static final Integer PORT = 8888; 							// 被监听端口号
        private static final String DELIMITER = "_$"; 						// 拆包分隔符 
          
        public static void main(String[] args) {   
            EventLoopGroup bossGroup = new NioEventLoopGroup(); 			// 用于接收进来的连接
            EventLoopGroup workerGroup = new NioEventLoopGroup(); 			// 用于处理进来的连接
            try {   
                ServerBootstrap serverbootstrap = new ServerBootstrap(); 	// 启动NIO服务的辅助启动类
                serverbootstrap.group(bossGroup, workerGroup) 				// 分别设置bossGroup, workerGroup 顺序不能反
                .channel(NioServerSocketChannel.class) 						// Channel的创建工厂，启动服务时会通过反射的方式来创建一个NioServerSocketChannel对象
                .handler(new LoggingHandler(LogLevel.INFO))					// handler在初始化时就会执行，可以设置打印日志级别
                .childHandler(new ChannelInitializer<SocketChannel>() {   	// childHandler会在客户端成功connect后才执行，这里实例化ChannelInitializer
                    @Override  
                    protected void initChannel(SocketChannel socketChannel) throws Exception {  	// initChannel方法执行后删除实例ChannelInitializer，添加以下内容
                        ByteBuf delimiter = Unpooled.copiedBuffer(DELIMITER.getBytes());  		// 获取特殊分隔符的ByteBuffer
                        socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(128, delimiter)); // 设置特殊分隔符用于拆包 
    // socketChannel.pipeline().addLast(new FixedLengthFrameDecoder(8)); 设置指定长度分割
                        socketChannel.pipeline().addLast(new StringDecoder());  				// 设置字符串形式的解码 
                        socketChannel.pipeline().addLast(new ITDragonServerHandler());			// 自定义的服务器处理类，负责处理事件
                    }  
                })  
                .option(ChannelOption.SO_BACKLOG, 128) 						// option在初始化时就会执行，设置tcp缓冲区 
                .childOption(ChannelOption.SO_KEEPALIVE, true); 			// childOption会在客户端成功connect后才执行，设置保持连接 
                ChannelFuture future = serverbootstrap.bind(PORT).sync(); 	// 绑定端口， 阻塞等待服务器启动完成，调用sync()方法会一直阻塞等待channel的停止
                future.channel().closeFuture().sync(); 						// 等待关闭 ，等待服务器套接字关闭
            } catch (Exception e) {   
                e.printStackTrace();  
            } finally {   
                workerGroup.shutdownGracefully(); 							// 关闭线程组，先打开的后关闭 
                bossGroup.shutdownGracefully();  
            }  
        }  
    }

### 核心参数说明 ###

**NioEventLoopGroup** : 是用来处理I/O操作的多线程事件循环器。 Netty提供了许多不同的EventLoopGroup的实现用来处理不同传输协议。  
**ServerBootstrap** : 启动NIO服务的辅助启动类。先配置Netty服务端启动参数，执行bind(PORT)方法才算真正启动服务。  
**group** : 注册EventLoopGroup  
**channel** : channelFactory，用于配置通道的类型。  
**handler** : 服务器始化时就会执行的事件。  
**childHandler** : 服务器在和客户端成功连接后会执行的事件。  
**initChannel** : channelRegistered事件触发后执行，删除ChannelInitializer实例，添加该方法体中的handler。  
**option** : 服务器始化的配置。  
**childOption** : 服务器在和客户端成功连接后的配置。  
**SocketChannel** : 继承了Channel，通过Channel可以对Socket进行各种操作。  
**ChannelHandler** : 通过ChannelHandler来间接操纵Channel，简化了开发。  
**ChannelPipeline** : 可以看成是一个ChandlerHandler的链表。  
**ChannelHandlerContext** : ChannelPipeline通过ChannelHandlerContext来间接管理ChannelHandler。

### 自定义服务器处理类 ###

第一步：继承 ChannelInboundHandlerAdapter，其父类已经实现了ChannelHandler接口，简化了开发。  
第二步：覆盖 chanelRead()事件处理方法 ，每当服务器从客户端收到新的数据时，该方法会在收到消息时被调用。  
第三步：释放 ByteBuffer，ByteBuf是一个引用计数对象，这个对象必须显示地调用release()方法来释放。  
第四步：异常处理，即当Netty由于IO错误或者处理器在处理事件时抛出的异常时触发。在大部分情况下，捕获的异常应该被记录下来并且把关联的channel给关闭掉。

package com.itdragon.delimiter;
    import com.itdragon.utils.ITDragonUtil;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.util.ReferenceCountUtil;  
      
    public class ITDragonServerHandler extends ChannelInboundHandlerAdapter{   
    	private static final String DELIMITER = "_$"; // 拆包分隔符 
        @Override  
        public void channelRead(ChannelHandlerContext chc, Object msg) {   
            try {   
            	// 普通读写数据
            	/* 设置字符串形式的解码 new StringDecoder() 后可以直接使用 ByteBuf buf = (ByteBuf) msg; byte[] req = new byte[buf.readableBytes()]; buf.readBytes(req); String body = new String(req, "utf-8"); */
            	System.out.println("Netty Server : " + msg.toString());
                // 分隔符拆包 
                String response = ITDragonUtil.cal(msg.toString())+ DELIMITER;  
                chc.channel().writeAndFlush(Unpooled.copiedBuffer(response.getBytes()));
            } catch (Exception e) {   
                e.printStackTrace();  
            } finally {   
                ReferenceCountUtil.release(msg); // 写入方法writeAndFlush ，Netty已经释放了
            }  
        }  
        // 当出现Throwable对象才会被调用
        @Override  
        public void exceptionCaught(ChannelHandlerContext chc, Throwable cause) {   
            // 这个方法的处理方式会在遇到不同异常的情况下有不同的实现，比如你可能想在关闭连接之前发送一个错误码的响应消息。 
            cause.printStackTrace();  
            chc.close();  
        }  
    }

### 客户端启动流程 ###

第一步：创建一个用于发送请求的线程池。  
第二步：客户端实例化Bootstrap NIO服务启动辅助类，简化开发。  
第三步：配置参数，粘包，发送请求。  
第四步：关闭资源。  
值得注意的是，和ServerBootstrap不同，它并没有childHandler和childOption方法。

package com.itdragon.delimiter;
    import io.netty.bootstrap.Bootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;  
    import io.netty.channel.ChannelFuture;  
    import io.netty.channel.ChannelInitializer;  
    import io.netty.channel.ChannelOption;  
    import io.netty.channel.nio.NioEventLoopGroup;  
    import io.netty.channel.socket.SocketChannel;  
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.handler.codec.DelimiterBasedFrameDecoder;
    import io.netty.handler.codec.FixedLengthFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;  
      
    public class ITDragonClient {   
          
        private static final Integer PORT = 8888;  
        private static final String HOST = "127.0.0.1";  
        private static final String DELIMITER = "_$"; // 拆包分隔符 
          
        public static void main(String[] args) {   
            NioEventLoopGroup group = new NioEventLoopGroup();  
            try {   
                Bootstrap bootstrap = new Bootstrap();  
                bootstrap.group(group)  
                .channel(NioSocketChannel.class)  
                .handler(new ChannelInitializer<SocketChannel>() { 
                    @Override  
                    protected void initChannel(SocketChannel socketChannel) throws Exception {  
                    	ByteBuf delimiter = Unpooled.copiedBuffer(DELIMITER.getBytes());  
                        // 设置特殊分隔符
                        socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(128, delimiter));  
                        // 设置指定长度分割 不推荐，两者选其一
    // socketChannel.pipeline().addLast(new FixedLengthFrameDecoder(8)); 
                        socketChannel.pipeline().addLast(new StringDecoder()); 
                        socketChannel.pipeline().addLast(new ITDragonClientHandler());  
                    }  
                })  
                .option(ChannelOption.SO_KEEPALIVE, true);  
                  
                ChannelFuture future = bootstrap.connect(HOST, PORT).sync(); // 建立连接 
                future.channel().writeAndFlush(Unpooled.copiedBuffer(("1+1"+DELIMITER).getBytes()));  
                future.channel().writeAndFlush(Unpooled.copiedBuffer(("6+1"+DELIMITER).getBytes()));  
                future.channel().closeFuture().sync();  
            } catch (Exception e) {   
                e.printStackTrace();  
            } finally {   
                group.shutdownGracefully();  
            }  
        }  
    }

### 客户端请求接收类 ###

和服务器处理类一样，这里只负责打印数据。

package com.itdragon.delimiter;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.util.ReferenceCountUtil;  
    public class ITDragonClientHandler extends ChannelInboundHandlerAdapter{   
              
    	@Override 
        public void channelRead(ChannelHandlerContext chc, Object msg) {   
            try {   
            	/* 设置字符串形式的解码 new StringDecoder() 后可以直接使用 ByteBuf buf = (ByteBuf) msg; byte[] req = new byte[buf.readableBytes()]; buf.readBytes(req); String body = new String(req, "utf-8"); */
                System.out.println("Netty Client :" + msg);  
            } catch (Exception e) {   
                e.printStackTrace();  
            } finally {   
                ReferenceCountUtil.release(msg);
            }  
        }  
        public void exceptionCaught(ChannelHandlerContext chc, Throwable cause) {   
            cause.printStackTrace();  
            chc.close();  
        }  
    }

打印结果

一月 29, 2018 11:31:10 上午 io.netty.handler.logging.LoggingHandler channelRegistered
    信息: [id: 0xcf3a3ac1] REGISTERED
    一月 29, 2018 11:31:11 上午 io.netty.handler.logging.LoggingHandler bind
    信息: [id: 0xcf3a3ac1] BIND: 0.0.0.0/0.0.0.0:8888
    一月 29, 2018 11:31:11 上午 io.netty.handler.logging.LoggingHandler channelActive
    信息: [id: 0xcf3a3ac1, L:/0:0:0:0:0:0:0:0:8888] ACTIVE
    一月 29, 2018 11:31:18 上午 io.netty.handler.logging.LoggingHandler channelRead
    信息: [id: 0xcf3a3ac1, L:/0:0:0:0:0:0:0:0:8888] READ: [id: 0xf1b8096b, L:/127.0.0.1:8888 - R:/127.0.0.1:4777]
    一月 29, 2018 11:31:18 上午 io.netty.handler.logging.LoggingHandler channelReadComplete
    信息: [id: 0xcf3a3ac1, L:/0:0:0:0:0:0:0:0:8888] READ COMPLETE
    Netty Server : 1+1
    Netty Server : 6+1
    
    Netty Client :2
    Netty Client :7

从日志中可以看出Channel的状态从REGISTERED —> ACTIVE —> READ —> READ COMPLETE。服务端也是按照特殊分割符拆包。

## 总结 ##

看完本章，你必须要掌握的三个知识点：NioEventLoopGroup，ServerBootstrap，ChannelHandlerAdapter  
1 **NioEventLoopGroup 本质就是一个线程池，管理多个NioEventLoop，一个NioEventLoop管理多个Channel。**  
2 **NioEventLoop 负责不停地轮询IO事件，处理IO事件和执行任务。**  
3 **ServerBootstrap 是NIO服务的辅助启动类，先配置服务参数，后执行bind方法启动服务。**  
4 **Bootstrap 是NIO客户端的辅助启动类，用法和ServerBootstrap类似。**  
5 **Netty 使用FixedLengthFrameDecoder 固定长度拆包，DelimiterBasedFrameDecoder 分隔符拆包。**

到这里，Netty的拆包粘包，以及Netty的重要组件，服务器启动流程到这里就结束了，如果觉得不错可以点一个\*\* “推荐” \*\* ，也可以\*\* “关注” \*\*我哦。

## 优质文章 ##

http://blog.csdn.net/spiderdog/article/category/1800249  
https://www.jianshu.com/p/c5068caab217

[aHR0cHM6Ly9pbWcyMDIwLmNuYmxvZ3MuY29tL2Jsb2cvODA2OTU2LzIwMjAwNy84MDY5NTYtMjAyMDA3MzEyMzQ1MDc3MjItMTExNjgwOTkyMi5naWY]: /images/20220601/5f3e74c53a794504bc4b5ac110b156fd.png
[Netty]: /images/20220601/c7f623ad7bd74c3db4a77e1094369aa4.png
[NIO]: http://www.cnblogs.com/itdragon/p/8337234.html
[format_png]: /images/20220601/15119b854a3f4debbd90e0084b50c9df.png
[format_png 1]: /images/20220601/e2018697332c49608b7a72eb6c54e58b.png
[format_png 2]: /images/20220601/8cdaee25df564748a95a4a4e72f52578.png