Netty知多少

Author: haoransun
Wechat: SHR—97

一、概述

官方网站：http://netty.io/

• Netty是一款异步事件驱动的网络通信框架，用来快速开发可维护的高性能、高扩展的协议服务器和客户端。

• Netty是一个基于NI0客户端、服务器框架，使用它可以轻松快速的的构建网络应用，如协议服务器和客户端。它极大地简化了TCP和UDP套接字服务器网络编程开发。

• 高性能、稳定、高度灵活，社区活跃，有大量成功的商业实践项目

  二、Netty编程模型

  Maven坐标

  <dependency>
      <groupId>io.netty</groupId>
      <artifactId>netty-all</artifactId>
      <version>5.0.0.Alpha2</version>
  </dependency>

服务器

• 服务器编程步骤：

1.创建服务器引导对象
2.创建请求转发和IO处理的线程池组
3.绑定线程池
4.设置服务的实现类
5.初始化通讯管道（重点）
6.绑定端口启动服务
7.关闭服务
8.释放资源

• 服务器示例代码：

/**
 * @author: haoransun
 * @date: 2018/05/05
 * 基于Netty网络通信框架的服务器编程模型
  * 1. 创建服务器的引导对象
  * 2. 创建请求接收和IO处理的线程池
  * 3. 绑定线程池
  * 4. 指定服务的实现类
  * 5. 初始化通讯管道（***重点***）
  * 6. 绑定端口，并启动服务
  * 7. 关闭服务
  * 8. 释放资源（释放线程池资源）
  *
 */
public class NettyServerBootstrap {
	public static void main(String[] args) throws InterruptedException {
		//1. 创建服务器的引导对象
		ServerBootstrap serverBootstrap = new ServerBootstrap();
		
		//2. 创建请求接受和IO处理的线程池
			//请求接受的线程池
		NioEventLoopGroup bossGroup = new NioEventLoopGroup();
			//IO处理的线程池
		NioEventLoopGroup workerGroup = new NioEventLoopGroup();
		
		//3. 绑定线程池
		serverBootstrap.group(bossGroup,workerGroup);
		
		//4. 指定服务的实现类
		serverBootstrap.channel(NioServerSocketChannel.class);
		
		//5. 初始化通讯管道（***重点***） Handler处理器按成IO的处理
		serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
		/**
		 * 通道初始化方法 绑定处理器
		 * 处理数据：通道+处理器
		 * @param socketChannel
		 * @throws Exception
		 */
		 
		 @Override
		 protected void initChannel(SocketChannel socketChannel) throws Exception {
		 ChannelPipeline pipeline = socketChannel.pipeline();
		 
		 //配置拆包粘包处理器
		 //消息头  所占字节数量 2     2^16=65536
		 pipeline.addLast(new LengthFieldPrepender(2));
		 //参数一:消息包的最大长度
		 // 参数⼀一：消息包的最⼤大⻓长度 参数⼆二：⻓长度域的偏移量量 参数三：⻓长度字段所占
		 //字节⼤大⼩小 参数四：⻓长度补偿 参数五：从数据帧中跳过的字节数
		 pipeline.addLast(new LengthFieldBasedFrameDecoder(65535,0,2,0,2));
		 
		 pipeline.addLast(new ObjectToByteBufEncoder());
		 pipeline.addLast(new ByteBufToObjectDecoder());
		 pipeline.addLast(new NettyServerChannelHandlerAdapter());
		 }
	  });
	  
		//6. 绑定端口，并启动服务
		System.out.println("服务器已启动！正在8888端口等待客户端的连接");
		ChannelFuture channelFuture = serverBootstrap.bind(8888).sync();
		
		//7. 关闭服
		channelFuture.channel().closeFuture().sync();
		
		//8. 释放资源
		workerGroup.shutdownGracefully();
		bossGroup.shutdownGracefully();
	}
}

• 服务器 通道处理类 示例代码：

/**
 * @author: haoransun
 * @date: 2018/05/05
 * Netty服务器  通道处理类
  */

public class NettyServerChannelHandlerAdapter extends ChannelHandlerAdapter{

	/**
	* 异常回调方法
	* @param ctx 通道处理器的上下文  获取通道的相关信息
	* @param cause
	* @throws Exception
	*/
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		System.out.println("服务器错误");
		cause.printStackTrace();
		ctx.close();
	}

	/**
	* 处理请求的回调方法
	* @param ctx 通道处理器的上下文  获取通道的相关信息
	* @param msg 通道中传输的数据
	* @throws Exception
	*  NIO      ByteBuffer 
	*           Netty    ByteBuf
	*                    读写指针
	*/
	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		//传输的数据类型为ByteBuf
		//ByteBuf byteBuf = (ByteBuf) msg;
		//System.out.println("服务器收到客户端数据："+byteBuf.toString(CharsetUtil.UTF_8));
		
		//byteBuf.clear();
		//======================================
		System.out.println("接受到的请求数据:"+msg);
		
		//======================================
		//响应数据
		//byteBuf.writeBytes("地瓜地瓜，我是土豆...".getBytes());
		//ctx.writeAndFlush(byteBuf);
		
		//响应一个对象类型的数据
		//Netty默认使用ByteBuff类型传输数据 对象不支持
		
		//通道异步处理的结果对象
		//ChannelFuture channelFuture = ctx.writeAndFlush(new Date());
		//常用监听器
		//在操作失败时触发异常
		//   channelFuture.addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
		//添加监听，操作完成  关闭通道
		//        channelFuture.addListener(ChannelFutureListener.CLOSE);
		
		//添加监听 操作失败或者不成功 关闭通道
		//        channelFuture.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
		//=========================================================
		for (int i=0;i< 100 ;i++){
			ctx.writeAndFlush("你好，客户端，我是服务器"+i);
		}
	}

}

客户端

• 客户端编程步骤：

1.初始化客户端引导程序
2.创建IO处理线程池
3.设置客户端的实现类
4.绑定线程池
5.初始化客户端通道（重点）
6.连接服务器
7.关闭通道
8.释放资源

• 客户端示例代码：

/**
 * @author: haoransun
 * @date: 2018/05/05
 * 基于Netty的Client程序
  * 1. 创建客户端引导对象
  * 2. 创建IO处理线程池
  * 3. 绑定线程池
  * 4. 指定客户端的实现类
  * 5. 初始化客户端通道（****重点****）
  * 6. 连接服务器
  * 7. 关闭服务
  * 8. 释放资源
  */

public class NettyClientBootstrap {
	public static void main(String[] args) throws InterruptedException {
		//1. 创建客户端引导类
		Bootstrap bootstrap = new Bootstrap();
		
		//2. 创建客户端IO处理线程池
		NioEventLoopGroup workGroup = new NioEventLoopGroup();
		
		//3. 绑定线程池
		bootstrap.group(workGroup);
		
		//4. 指定客户端的实现类
		bootstrap.channel(NioSocketChannel.class);
		
		//5. 初始化客户端通讯通道（*****）
		bootstrap.handler(new ChannelInitializer<SocketChannel>() {
		/**
		* 客户端发起请求 接收响应
		* @param ch
		* @throws Exception
		*/
		
		@Override
		protected void initChannel(SocketChannel ch) throws Exception {
		
		ChannelPipeline channelPipeline = ch.pipeline();
		
		//参数一:负责在待发送的ByteBuf消息头上增加一个⻓度字段，⽤用来表示消息体的⻓度 参数代表：⻓度字段所占字节⼤小
		channelPipeline.addLast(new LengthFieldPrepender(2));
		// 参数⼀：消息包的最⼤长度 参数⼆：⻓度域的偏移量量 参数三：⻓度字段所占字节⼤小 参数四：⻓度补偿 参数五：从数据帧中跳过的字节数
		
		channelPipeline.addLast(new LengthFieldBasedFrameDecoder(65535,0,2,0,2));
		
		channelPipeline.addLast(new ObjectToByteBufEncoder());
		channelPipeline.addLast(new ByteBufToObjectDecoder());
		channelPipeline.addLast(new NettyClientChannelHandlerAdapter());
		 }
		});
		
		//6. 绑定服务器
		ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8888).sync();
		
		//7. 关闭通道
		channelFuture.channel().closeFuture().sync();
		
		//8. 释放资源
		workGroup.shutdownGracefully();
	}
}

• 客户端 通道处理类 示例代码：

/**
 * @author: haoransun
 * @date: 2018/05/05
 * Netty 客户端的通道处理器类
  */

public class NettyClientChannelHandlerAdapter extends ChannelHandlerAdapter {

	/**
	* 异常回调方法
	* @param ctx
	* @param cause
	* @throws Exception
	*/  @Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		System.out.println("客户端错误！");
		cause.printStackTrace();
		ctx.close();
		}
	
	/**
	* 通道可用回调方法（发起请求的回调方法）
	* @param ctx
	* @throws Exception
	*/  @Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		ByteBuf byteBuf = Unpooled.buffer();
		byteBuf.writeBytes("你好，服务器-.- ".getBytes());
		ctx.writeAndFlush(byteBuf);
		
		//编解码测试数据
		ctx.writeAndFlush("Hello,World");
	}
	
	/**
	* 读事件回调方法（接收到响应后的回调）
	* @param ctx
	* @param msg
	* @throws Exception
	*             NIO      ByteBuffer
	*             Netty    ByteBuf
	*                       读写指针
	*/
	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
	
		System.out.println("响应数据已到达...");
		
		//传输的数据类型为ByteBuf
		//ByteBuf byteBuf = (ByteBuf) msg;
		//System.out.println("收到服务器响应："+byteBuf.toString(CharsetUtil.UTF_8));
		//ctx.close();
		
		//获取对象类型的响应结果
		//Date date = (Date)msg;
		//System.out.println("接收到的响应结果为："+date);
		//ctx.close();
		
		//====================================================
		System.out.println(msg);
	}
}

三、异常处理

模拟问题

1.修改 NettyserverChannelHandlerAdapter 第44行成 ctx.writeandFlush（new Date（））

2.启动Server、Client测试，测试发现客户端无法获取服务器响应

Netty默认情况下，只支持ByteBuf类型传输，不支持对象类型传输，直接写入对象会产生异常，通过添加istener处理异常

添加监听器

ChannelFuture channelFuture = ctx.writeAndFlush(new Date());

//添加监听，操作完成  关闭通道
channelFuture.addListener(ChannelFutureListener.CLOSE);

//添加监听，操作失败时或者不成功时触发异常，关闭通道
channelFuture.addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);

//添加监听，操作失败时或者不成功时触发异常，关闭通道
 channelFuture.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);

四、编解码

在异常处理中，我们发现Netty默认采用ByteBuf 传输数据，如果我们需要传输对象类型，需要自定义Channel，添加编解码器，以支持对象类型数据传输。

导入Maven依赖

<dependencies>   
<!-- https://mvnrepository.com/artifact/io.netty/netty-all -->    
	<dependency>        
		<groupId>io.netty</groupId>        
		<artifactId>netty-all</artifactId>        
		<version>5.0.0.Alpha2</version>   
	</dependency>

	<dependency>        
		<groupId>commons-lang</groupId>        
		<artifactId>commons-lang</artifactId>        
		<version>2.5</version>    
	</dependency> 
</dependencies>

定制解码器

/**
 * @author: haoransun
 * @date: 2018/05/05
 * 编码器
 *   对象-- >ByteBuff 出站 outBound
 */
 
public class ObjectToByteBufEncoder extends MessageToMessageEncoder {
	/**
	* 编码方法：
	*     序列化方案 jdk原生的序列化支持 commons-lang
	*
	*     也可以用序列化
	*               fastjson   Object ---> json
	*
	* @param ctx 通道处理器的上下文对象
	* @param msg 出站的数据 Object
	* @param out 编码后的消息列表(将当前编码后的数据添加到消息列表中
	*            Netty会自动将该数据添加到下一个处理器中)
	* @throws Exception
	*/
	@Override
	protected void encode(ChannelHandlerContext ctx, Object msg, List out) throws Exception {
	
	//1. 将出站数据进行序列化
	byte[] bytes = SerializationUtils.serialize((Serializable) msg);
	//2. 将序列化后的字节数组  包装到ByteBuf对象中
	ByteBuf byteBuf = Unpooled.buffer();
	byteBuf.writeBytes(bytes);
	//将ByteBuf添加到消息队列中 交给下一个Handler处理
	out.add(byteBuf);
  }
}

定制解码器

/**
 * @author: haoransun
 * @date: 2018/05/05
 * 解码器
 *   ByteBuf -----> Object
 */

public class ByteBufToObjectDecoder  extends MessageToMessageDecoder {

	/**
	* 
	*  fastjosn  Json---->Object 
	* 
	* 
	* @param ctx 通道处理器的上下文对象
	* @param msg 通达中传输的数据 ByteBuf
	* @param list 消息列表
	* @throws Exception
	*/
	
	@Override
	protected void decode(ChannelHandlerContext ctx, Object msg, List list) throws Exception {
		
		//1. 将通道中传输的数据强转成ByteBuf类型
		ByteBuf byteBuf = (ByteBuf)msg;
		
		//2. 获取ByteBuf中可读的数据
		//填充到数组里
		byte[] bytes = new byte[byteBuf.readableBytes()];
		//对可读的字节数组进行读取
		byteBuf.readBytes(bytes);
		
		//3. 将可读的数据反序列化成一个Object对象
		Object o = SerializationUtils.deserialize(bytes);
		
		//4. 将反序列化后的结果添加到消息列表中，交给下一个Handler进行处理
		list.add(o);
   }
}

组装通道处理器

//5. 初始化通讯管道（服务器）
serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
	/**
	* 通道初始化方法 绑定处理器
	* 处理数据：通道+处理器
	* @param socketChannel
	* @throws Exception
	*/
	@Override
	protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
		//配置拆包粘包处理器
		//消息头  所占字节数量 2     2^16=65536 
		pipeline.addLast(new LengthFieldPrepender(2));
		
		//参数一：消息包的最⼤长度 
		//参数二：⻓长度域的偏移量量 
		//参数三：长度字段所占字节⼤小 
		//参数四：长度补偿 
		//参数五：从数据帧中跳过的字节数
		pipeline.addLast(new LengthFieldBasedFrameDecoder(65535,0,2,0,2));
		
		pipeline.addLast(new ObjectToByteBufEncoder());
		pipeline.addLast(new ByteBufToObjectDecoder());
		pipeline.addLast(new NettyServerChannelHandlerAdapter());
   }
});


//5. 初始化通讯管道（客户端）
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
	/**
	* 客户端发起请求 接收响应
	* @param ch
	* @throws Exception
	*/
	
	@Override
	protected void initChannel(SocketChannel ch) throws Exception {
		ChannelPipeline channelPipeline = ch.pipeline();
	
		//参数一:负责在待发送的ByteBuf消息头上增加一个⻓度字段，⽤来表示消息体的⻓度 参数代表：⻓度字段所占字节⼤小
		channelPipeline.addLast(new LengthFieldPrepender(2));
		//参数⼀：消息包的最⼤长度 
		//参数⼆：⻓度域的偏移量量 
		//参数三：⻓度字段所占字节⼤小 
		//参数四：⻓度补偿 
		//参数五：从数据帧中跳过的字节数
		channelPipeline.addLast(new LengthFieldBasedFrameDecoder(65535,0,2,0,2));
		
		channelPipeline.addLast(new ObjectToByteBufEncoder());
		channelPipeline.addLast(new ByteBufToObjectDecoder());
		channelPipeline.addLast(new NettyClientChannelHandlerAdapter());
   }
});

修改服务器和客户端事件回调方法（读写直接操作对象）

/**
 * 读事件回调方法
 * @param ctx
 * @param msg
 * @throws Exception
 */
@Override
public void channelRead(ChannelHandlerContext ctx,Object msg) throws Exception{
	System.out.println("服务器收到客户端数据："+ msg);
	//写入对象
	ChannelFuture channelFuture = ctx.writeAndFlush(new Date());
	
	channelFuture.addListener(ChannelFutureListener.CLOSE);
	channelFuture.addListener(ChannelFutureListener.CLOSE_ON_FAILURE)
	channelFuture.addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);

}

五、TCP粘包/拆包问题

模拟问题

• 继续修改NettyServerChannelHandlerAdapter，尝试给客户端响应多个对象，如图所示：

• 测试发现客户端解码过程中产生异常或者出现丢失数据情况

问题原因

• TCP是个“流”协议，所谓流，就是没有分界线的一串数据。TCP底层并不了解上层业务数据的具体含义，它会根据TCP缓冲区的实际情况进行包的划分，所以在业务上，一个完整的数据包可能会被TCP拆分成多个包进行发送，也有可能把多个小的包封装成一个大的数据包发送，这就是所谓的TCP拆包和粘包问题。

• 现在假设客户端向服务端连续发送了两个数据包，用packet1和packet2来表示，那么服务端收到的数据可以分为三种，现列举如下：

第一种情况，接收端正常收到两个数据包，即没有发生拆包和粘包的现象

第二种情况，接收端只收到一个数据包，由于TCP是不会出现丢包的，所以这一个数据包中包含了发送端发送的两个数据包的信息，这种现象即为粘包。这种情况由于接收端不知道这两个数据包的界限，所以对于接收端来说很难处理

第三种情况，这种情况有两种表现形式，如下图。接收端收到了两个数据包，但是这两个数据包要么是不完整的，要么就是多出来一块，这种情况即发生了拆包和粘包。这两种情况如果不加特殊处理，对于接收端同样是不好处理的。

粘包和拆包问题的解决策略

由于底层的TCP无法理解上层的业务数据，所以在底层是无法保证数据包不被拆分和重组的，所以只能通过上层的应用协议栈设计解决，常用的解决策略如下：

• 消息定长，例如每个报文的大小固定长度200字节，如果不够，空位补空格
• 在包尾增加回车换行符进行分割，例如FTP协议
• 将消息分为消息头和消息体，消息头中记录了消息体的长度
• 更复杂的应用层协议

使用Netty的编解码器解决粘包和拆包原理

• o.netty.handler.codec.LengthFieldPrepender

• 编码器：负责在待发送的ByteBuf消息头上增加一个长度字段，用来标识消息的长度*

• io.nety.handler.codec.LengthFieldBasedFrameDecoder

• 解码器：基于消息长度的半包的解码器*

• 通讯管道配置

代码实现

• Client

• Server

• 测试

Netty原理示意图

简单的自定义RPC设计