netty（十）初识Netty-Handler & Pipeli
一、简介
ChannelHandler 用来处理 Channel 上的各种事件，分为入站、出站两种。
所有 ChannelHandler 被连成一串，就是 Pipeline。
- 入站处理器通常是 ChannelInboundHandlerAdapter 的子类，主要用来读取客户端数据，写回结果。
- 出站处理器通常是 ChannelOutboundHandlerAdapter 的子类，主要对写回结果进行加工。
可以将netty中的各种组件进行一个比如，如下所示：
netty各组件关系.png
二、代码分析
我们通过代码的形式，来展示Handler和Pipeline之间的关系，以及ChannelInboundHandlerAdapter 和ChannelOutboundHandlerAdapter 的使用方式。
客户端代码，使用前面文章用到的客户端，支持控制台输入内容：
public static void main(String[] args) throws Exception {
// 将group提出来，不能匿名方式，为了后面调动shutdownGracefully()方法
NioEventLoopGroup group = new NioEventLoopGroup();
ChannelFuture channelFuture = new Bootstrap()
.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) {
ch.pipeline().addLast(new StringEncoder());
}
})
.connect("localhost", 8080);
// 同步等待连接
Channel channel = channelFuture.sync().channel();
new Thread(() -> {
Scanner scanner = new Scanner(System.in);
while (true) {
String line = scanner.nextLine();
if ("q".equals(line)) {
System.out.println("关闭channel");
// close 异步操作 1s 之后
channel.close();
break;
}
channel.writeAndFlush(line);
}
}, "input").start();
// 处理channel关闭后的操作
ChannelFuture closeFuture = channel.closeFuture();
//异步 - EventLoopGroup线程优雅关闭
closeFuture.addListener((ChannelFutureListener) future -> group.shutdownGracefully());
}
2.1 ChannelInboundHandlerAdapter
服务端代码，首先添加三个入栈处理器，并分别指定名称为h1，h2，h3，分别打印1,2,3
public static void main(String[] args) {
new ServerBootstrap().group(new NioEventLoopGroup())
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel nioServerSocketChannel) {
ChannelPipeline pipeline = nioServerSocketChannel.pipeline();
pipeline.addLast("h1", new ChannelInboundHandlerAdapter() {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("1");
super.channelRead(ctx, msg);
}
});
pipeline.addLast("h2", new ChannelInboundHandlerAdapter() {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("2");
super.channelRead(ctx, msg);
}
});
pipeline.addLast("h3", new ChannelInboundHandlerAdapter() {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("3");
super.channelRead(ctx, msg);
}
});
}
}).bind(8080);
}
分别启动客户端和服务端，客户端随便输入内容，服务端得到如下输出：
1
2
3
由上所示我们可以得到一个结论，我们添加的入站handler是按照添加顺序进行执行的。
通过简单的源码跟踪：
private void addLast0(AbstractChannelHandlerContext newCtx) {
//获取当前处理器链表中的尾节点的前一个处理器
AbstractChannelHandlerContext prev = this.tail.prev;
//将prev设置为新处理器的前一个处理器
newCtx.prev = prev;
//将新处理器的尾结点设置为tail
newCtx.next = this.tail;
//将prev的下一个节点设置为新处理器
prev.next = newCtx;
//将尾结点的前一个处理器设置为新处理器
this.tail.prev = newCtx;
}
我们能得到一个结论，我们通过addLast方法添加的handler其实是添加在链表当中，其中每一个节点都有其对象的头和尾节点，所以我们前面添加的三个处理器会如下所示排列：
Handler在Pipeline中的关系.png
2.2 ChannelOutboundHandlerAdapter
接下来我们在上面的基础上，增加ChannelOutboundHandlerAdapter 处理器，此处理器只有在进行channel数据写入才会执行，这个就不掩饰了，我们直接在代码当中添加写入代码。
public static void main(String[] args) {
new ServerBootstrap().group(new NioEventLoopGroup())
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel nioServerSocketChannel) {
ChannelPipeline pipeline = nioServerSocketChannel.pipeline();
// 入站处理器
pipeline.addLast("h1", new ChannelInboundHandlerAdapter() {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("1");
super.channelRead(ctx, msg);
}
});
pipeline.addLast("h2", new ChannelInboundHandlerAdapter() {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("2");
super.channelRead(ctx, msg);
}
});
pipeline.addLast("h3", new ChannelInboundHandlerAdapter() {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("3");
super.channelRead(ctx, msg);
//服务端channel调用写入方法，出站处理器才会生效
nioServerSocketChannel.writeAndFlush(ctx.alloc().buffer().writeBytes("helloworld".getBytes()));
}
});
//出站处理器
pipeline.addLast("h4", new ChannelOutboundHandlerAdapter() {
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
System.out.println("4");
super.write(ctx, msg, promise);
}
});
pipeline.addLast("h5", new ChannelOutboundHandlerAdapter() {
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
System.out.println("5");
super.write(ctx, msg, promise);
}
});
pipeline.addLast("h6", new ChannelOutboundHandlerAdapter() {
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
System.out.println("6");
super.write(ctx, msg, promise);
}
});
}
}).bind(8080);
}
结果：
1
2
3
6
5
4
如上所示，发现出站处理器的处理顺序是6、5、4，从后向前处理的。
其实ChannelPipeline的实现是一个双向链表，所以实现了上述出站、入站的功能。
2.3 上述代码过程分析
如前面的代码所示，我们可以在Pipeline中添加很多个handler，并使其按照一定的顺序去执行，其主要的一一四就在于，我可以在每个处理器当中，对收到的数据进行不同类型的处理。
比如h1中，我对收到的内容转字符串，将其发送给h2处理器，然后h2处理器将收到的字符串转成一个java对象，等等。
那么handler之间是如何传递处理后数据的呢？
在前面的入站handler当中，都有一行如下代码：
super.channelRead(ctx, msg);
此方法如下所示：
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ctx.fireChannelRead(msg);
}
内部执行了ctx.fireChannelRead(msg) ，也就是通过这行代码将处理过的消息传递给下一个处理器进行处理的。
并且此方法只能唤醒的是下一个入站处理器，如我们前面的代码，在h3使用这行代码唤醒h4是无效的，因为h4是一个出站处理器。所以我们可以删除该行代码，只使用以下这个即可：
nioServerSocketChannel.writeAndFlush(ctx.alloc().buffer().writeBytes("helloworld".getBytes()));
注意：
我们上面的代码使用的服务端的channel（nioServerSocketChannel）的writeAndFlush方法。
而在每个处理器当中传递过来的对象有一个ctx，而这个ctx也有writeAndFlush方法：
image.png
如果你使用了这个ctx，并将其放在h3中，那么我们会发现执行结果只会输除前三个入站处理器的结果。
为什么？
因为使用ctx这个writeAndFlush方法，pipeline会从当前处理器向前去寻找有没有出站处理器，而我们的h3、h2、h1都是入站处理器，所以没有出站处理器的输出。
而nioServerSocketChannel会从尾巴向前面找，所以找到了h6、h5、h4。
验证
在h3当中执行ctx的writeAndFlush方法，注释其中的nioServerSocketChannel调用，并将h3放到h5的后面去执行，按照前面的结论，应该会打印1、2、3、5、4。
pipeline.addLast("h3", new ChannelInboundHandlerAdapter() {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("3");
//h4是一个出站处理器，此处调用没有效果。
//super.channelRead(ctx, msg);
//服务端channel调用写入方法，出站处理器才会生效
//nioServerSocketChannel.writeAndFlush(ctx.alloc().buffer().writeBytes("helloworld".getBytes()));
ctx.writeAndFlush(ctx.alloc().buffer().writeBytes("helloworld".getBytes()));
}
});
执行代码结果,符合预期：
1
2
3
5
4
