1.基本说明
-
netty的组件设计:Netty的主要组件有
Channel
、EventLoop
、ChannelFuture
、ChannelHandler
、ChannelPip
e等 -
ChannelHandler
充当了处理入站和出站数据的应用程序逻辑的容器。例如,实现ChannelInboundHandler接口(或ChannelInboundHandlerAdapter),你就可以接收入站事件和数据,这些数据会被业务逻辑处理。当要给客户端发送响应时,也可以从ChannelInboundHandler冲刷数据。**业务逻辑通常写在一个或者多个ChannelInboundHandler中。**ChannelOutboundHandler原理一样,只不过它是用来处理出站数据的
-
ChannelPipeline
提供了ChannelHandler链的容器。以客户端应用程序为例,如果事件的运动方向是从客户端到服务端的,那么我们称这些事件为出站
的,即客户端发送给服务端的数据会通过pipeline中的一系列ChannelOutboundHandler,并被这些Handler处理,反之则称为入站的
2. 编码解码器
- 当Netty发送或者接受一个消息的时候,就将会发生一次数据转换。
- 入站消息会被解码:从字节转换为另一种格式(比如java对象);
- 如果是出站消息,它会被编码成字节。
- **Netty提供一系列实用的编解码器,他们都实现了ChannelInboundHadnler或者ChannelOutboundHandler接口。**在这些类中,channelRead方法已经被重写了。
- 以入站为例,对于每个从入站Channel读取的消息,这个方法会被调用。随后,它将调用由解码器所提供的decode()方法进行解码,并将已经解码的字节转发给ChannelPipeline中的下一个ChannelInboundHandler。
2.1 解码器-ByteToMessageDecoder
-
继承关系图
-
由于不可能知道远程节点是否会 一次性发送一个完整的信息, tcp有可能出现粘包拆包的问题, 这个类会对入站数据进行缓冲, 直到它准备好被处理.
-
一个关于ByteToMessageDecoder实例分析
这个例子,**每次入站从ByteBuf中读取4字节,将其解码为一个int,然后将它添加到下一个List中。**当没有更多元素可以被添加到该List中时,它的内容将会被发送给下一个ChannelInboundHandler。int在被添加到List中时,会被自动装箱为Integer。在调用readInt()方法前必须验证所输入的ByteBuf是否具有足够的数据
public class ToIntegerDecoder extends ByteToMessageDecoder { @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { if (in.readableBytes() >= 4) { out.add(in.readInt()); } } }
案例分析:
- 已知一个int4个字节,那我们在解码的时候就需要将4个字节组成一个int然后添加到下一个handler可以读取到的list中
- 在执行readInt方法时,ByteBuf中的readerIndex指针会向后移动,所以我们只需要一直读取即可
- 同理:
2.2 Netty的handler链的调用机制-实例1
实例要求:
使用自定义的编码器和解码器来 说明Netty的handler 调用机制 客户端发送long -> 服务器 服务端发送long -> 客户端
因为数据是从客户端到服务端,我先从客户端开始写(数据流转比较清除)
2.2.1 MyClient
- 需要一个MyClientInitializer()
package site.zhourui.nioAndNetty.netty.inboundhandlerandoutboundhandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
/**
* @author zr
* @date 2023/7/1 16:40
*/
public class MyClient {
public static void main(String[] args) throws Exception{
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group).channel(NioSocketChannel.class)
.handler(new MyClientInitializer()); //自定义一个初始化类
ChannelFuture channelFuture = bootstrap.connect("localhost", 8888).sync();
channelFuture.channel().closeFuture().sync();
}finally {
group.shutdownGracefully();
}
}
}
2.2.2 MyClientInitializer
- 因为数据是从客户端到服务器端是出站,所以需要编码,我们需要加入自定义编码器MyLongToByteEncoder
- 处理后续业务需要MyClientHandler
package site.zhourui.nioAndNetty.netty.inboundhandlerandoutboundhandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
/**
* @author zr
* @date 2023/7/1 16:40
*/
public class MyClientInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
//加入一个出站的handler 对数据进行一个编码
pipeline.addLast(new MyLongToByteEncoder());
//加入一个自定义的handler , 处理业务
pipeline.addLast(new MyClientHandler());
}
}
2.2.3 MyLongToByteEncoder
- 继承MessageToByteEncoder<T>
- 重写encode方法来自定义编码方式,将数据编码
package site.zhourui.nioAndNetty.netty.inboundhandlerandoutboundhandler;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
/**
* @author zr
* @date 2023/7/1 16:44
*/
public class MyLongToByteEncoder extends MessageToByteEncoder<Long> {
//编码方法
@Override
protected void encode(ChannelHandlerContext ctx, Long msg, ByteBuf out) throws Exception {
System.out.println("MyLongToByteEncoder encode 被调用");
System.out.println("msg=" + msg);
out.writeLong(msg);
}
}
2.2.4 MyClientHandler
这里就是将我们编码的数据在channel激活的时候发送一个long类型的数据到服务端
package site.zhourui.nioAndNetty.netty.inboundhandlerandoutboundhandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
/**
* @author zr
* @date 2023/7/1 16:41
*/
public class MyClientHandler extends SimpleChannelInboundHandler<Long> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception {
}
//重写channelActive 发送数据
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("MyClientHandler 发送数据");
//ctx.writeAndFlush(Unpooled.copiedBuffer(""))
ctx.writeAndFlush(123456L); //发送的是一个long
}
}
2.2.5 MyServer
将处理器封装为MyServerInitializer
package site.zhourui.nioAndNetty.netty.inboundhandlerandoutboundhandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
/**
* @author zr
* @date 2023/7/1 15:16
*/
public class MyServer {
public static void main(String[] args) throws InterruptedException {
NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
NioEventLoopGroup workerGroup = new NioEventLoopGroup(8);
try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(bossGroup,workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new MyServerInitializer());
ChannelFuture channelFuture = serverBootstrap.bind(8888).sync();
channelFuture.channel().closeFuture().sync();
}finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
2.2.6 MyServerInitializer
- 因为数据是从客户端到服务器端是出站,所以需要解码,我们需要加入自定义解码器MyByteToLongDecoder
- 处理后续业务需要MyServerHandler
package site.zhourui.nioAndNetty.netty.inboundhandlerandoutboundhandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
/**
* @author zr
* @date 2023/7/1 15:21
*/
public class MyServerInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();//一会下断点
//入站的handler进行解码 MyByteToLongDecoder
pipeline.addLast(new MyByteToLongDecoder());
//自定义的handler 处理业务逻辑
pipeline.addLast(new MyServerHandler());
System.out.println("xx");
}
}
2.2.7 MyByteToLongDecoder
- 继承ByteToMessageDecoder
- 重写decode方法来自定义解码方式,将数据解码
- decode 会根据接收的数据,被调用多次, 直到确定没有新的元素被添加到list或者是ByteBuf 没有更多的可读字节为止
- 如果list out 不为空,就会将list的内容传递给下一个 channelinboundhandler处理, 该处理器的方法也会被调用多次
- 参数说明:
- ctx:上下文对象
- in: 入站的 ByteBuf
- out:List 集合,将解码后的数据传给下一个handler
package site.zhourui.nioAndNetty.netty.inboundhandlerandoutboundhandler;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageCodec;
import io.netty.handler.codec.ByteToMessageDecoder;
import java.util.List;
/**
* @author zr
* @date 2023/7/1 16:36
*/
public class MyByteToLongDecoder extends ByteToMessageDecoder {
/**
*
* decode 会根据接收的数据,被调用多次, 直到确定没有新的元素被添加到list
* , 或者是ByteBuf 没有更多的可读字节为止
* 如果list out 不为空,就会将list的内容传递给下一个 channelinboundhandler处理, 该处理器的方法也会被调用多次
*
* @param ctx 上下文对象
* @param in 入站的 ByteBuf
* @param out List 集合,将解码后的数据传给下一个handler
* @throws Exception
*/
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
System.out.println("MyByteToLongDecoder 被调用");
//因为 long 8个字节, 需要判断有8个字节,才能读取一个long
if (in.readableBytes()>=8){
out.add(in.readLong());
}
}
}
2.2.8 MyServerHandler
- MyByteToLongDecoder解码的数据会传到下一个handler,那么数据类型就确定了所以SimpleChannelInboundHandler<Long>
- 将数据打印到控制台
package site.zhourui.nioAndNetty.netty.inboundhandlerandoutboundhandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
/**
* @author zr
* @date 2023/7/1 16:34
*/
public class MyServerHandler extends SimpleChannelInboundHandler<Long> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception {
System.out.println("从客户端" + ctx.channel().remoteAddress() + " 读取到long " + msg);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
2.2.9 测试
-
启动服务端
-
启动客户端
编码器和解码器都被调用,服务端拿到客户端编码后的数据解码后输出成功
2.2.10 特殊情况测试-发送数据与与编解码器的类型不一致
当我们客户端发送的数据为一个16个字节的字符串的时候会发生什么情况?
package site.zhourui.nioAndNetty.netty.inboundhandlerandoutboundhandler;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.CharsetUtil;
/**
* @author zr
* @date 2023/7/1 16:41
*/
public class MyClientHandler extends SimpleChannelInboundHandler<Long> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception {
}
//重写channelActive 发送数据
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("MyClientHandler 发送数据");
//ctx.writeAndFlush(Unpooled.copiedBuffer(""))
// ctx.writeAndFlush(123456L); //发送的是一个long
//分析
//1. "abcdabcdabcdabcd" 是 16个字节
//2. 该处理器的前一个handler 是 MyLongToByteEncoder
//3. MyLongToByteEncoder 父类 MessageToByteEncoder
//4. 父类 MessageToByteEncoder
/*
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
ByteBuf buf = null;
try {
if (acceptOutboundMessage(msg)) { //判断当前msg 是不是应该处理的类型,如果是就处理,不是就跳过encode
@SuppressWarnings("unchecked")
I cast = (I) msg;
buf = allocateBuffer(ctx, cast, preferDirect);
try {
encode(ctx, cast, buf);
} finally {
ReferenceCountUtil.release(cast);
}
if (buf.isReadable()) {
ctx.write(buf, promise);
} else {
buf.release();
ctx.write(Unpooled.EMPTY_BUFFER, promise);
}
buf = null;
} else {
ctx.write(msg, promise);
}
}
4. 因此我们编写 Encoder 是要注意传入的数据类型和处理的数据类型一致
*/
ctx.writeAndFlush(Unpooled.copiedBuffer("abcdabcdabcdabcd", CharsetUtil.UTF_8));
}
}
-
启动服务端
-
启动客户端
服务端输出了两次
MyByteToLongDecoder 被调用
代表MyByteToLongDecoder
的decode方法被调用了两次
- decode 会根据接收的数据,被调用多次, 直到确定没有新的元素被添加到list或者是ByteBuf 没有更多的可读字节为止
- 如果list out 不为空,就会将list的内容传递给下一个 channelinboundhandler处理, 该处理器的方法也会被调用多次
我们发现没有客户端没有打印
MyLongToByteEncoder 被调用
,代表MyLongToByteEncoder的encode方法没有被调用
2.3 Netty的handler链的调用机制-实例2
实例要求:
使用自定义的编码器和解码器来 说明Netty的handler 调用机制 客户端发送long -> 服务器 服务端发送long -> 客户端
实现图
实例1已经实现客户端向服务端发送数据,我们只需要在示例1的基础上在服务端打印客户端发送的数据的同时发送编码数据,然后客户端在解码读取数据即可
2.3.1 MyServerInitializer
加上编码器
2.3.2 MyServerHandler
发送数据
2.3.3 MyClientInitializer
客户端加上解码器
2.3.4 MyClientHandler
打印数据
2.3.5 测试
-
启动服务端
-
启动客户端
服务端也成功接收到了数据
- 结论:
- 不论解码器handler 还是 编码器handler 即接 收的消息类型必须与待处理的消息类型一致, 否则该handler不会被执行
- 在解码器 进行数据解码时,需要判断 缓存 区(ByteBuf)的数据是否足够,否则接收到的结果会和期望结果可能不一致
2.2.6 调用链debug
-
为了方便查看handler,最好起一个名字
-
启动服务端和客户端,代码走到断点查看head为整个调用链的(pipeline)的头结点
-
点击next查看下一个handler,是我们的
MyServerInitializer
-
点击next查看下一个handler,是我们的
MyByteToLongDecoder
-
点击next查看下一个handler,是我们的
MyLongToByteEncoder
-
点击next查看下一个handler,是我们的
MyServerHandler
-
点击next查看下一个handler,这是netty的默认尾节点next已经为空了
第四步和第五步只会执行一个,因为他们继承的父类不同出入站,出站和入站只会执行对应的一个
3. 解码器-ReplayingDecoder
- public abstract class ReplayingDecoder
extends ByteToMessageDecoder - ReplayingDecoder扩展了ByteToMessageDecoder类,使用这个类,我们不必调用readableBytes()方法。参数S指定了用户状态管理的类型,其中Void代表不需要状态管理(由ByteToMessageDecoder来帮我们自动识别并管理)
- ReplayingDecoder使用方便,但它也有一些局限性:
- 并不是所有的 ByteBuf 操作都被支持,如果调用了一个不被支持的方法,将会抛出一个
UnsupportedOperationException
。 - ReplayingDecoder 在某些情况下可能稍慢于 ByteToMessageDecoder,例如网络缓慢并且消息格式复杂时,消息会被拆成了多个碎片,速度变慢
- 并不是所有的 ByteBuf 操作都被支持,如果调用了一个不被支持的方法,将会抛出一个
3.1 应用实例
3.1.1 MyByteToLongDecoder2
- ReplayingDecoder<Void>代表没有指定类型,ReplayingDecoder来帮我们识别并管理
- 读取数据的时候也不需要判断数据是否足够读取,内部会进行处理判断
package site.zhourui.nioAndNetty.netty.inboundhandlerandoutboundhandler;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ReplayingDecoder;
import java.util.List;
public class MyByteToLongDecoder2 extends ReplayingDecoder<Void> {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
System.out.println("MyByteToLongDecoder2 被调用");
//在 ReplayingDecoder 不需要判断数据是否足够读取,内部会进行处理判断
out.add(in.readLong());
}
}
3.1.2 将实例2中的MyByteToLongDecoder替换为MyByteToLongDecoder2
MyClientInitializer
MyServerInitializer
3.1.3 测试
仍然能实现实例2的功能
4. 其它编解码器
4.1 LineBasedFrameDecoder
这个类在Netty内部也有使用,它使用行尾控制字符(\n或者\r\n)作为分隔符来解析数据。
4.2 DelimiterBasedFrameDecoder
使用自定义的特殊字符作为消息的分隔符。
4.3 HttpObjectDecoder
一个HTTP数据的解码器,之前的笔记中使用过
4.4 LengthFieldBasedFrameDecoder
通过指定长度来标识整包消息,这样就可以自动的处理黏包和半包消息。
5. Log4j 整合到Netty
-
在Maven 中添加对Log4j的依赖 在 pom.xml
<dependency> <groupId>log4j</groupId> <artifactId>log4j</artifactId> <version>1.2.17</version> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> <version>1.7.25</version> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-log4j12</artifactId> <version>1.7.25</version> <scope>test</scope> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-simple</artifactId> <version>1.7.25</version> <scope>test</scope> </dependency>
-
配置 Log4j , 在 resources/log4j.properties
log4j.rootLogger=DEBUG, stdout log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout.layout=org.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern=[%p] %C{1} - %m%n
-
随便启动一个服务
为了方便以后学习查看控制台清楚一点可以先关闭日志功能
把步骤2和步骤2的内容注释掉,当然你需要日志功能也可以打开