Netty应用(七) 之 Handler Netty服务端编程总结
目录
15.Handler
15.1 handler的分类
15.1.1 按照方向划分
15.1.2 handler的结构
15.2 输入方向ChannelInboundHandlerAdapter
15.2.1 输出方向Handler的顺序
15.2.2 多个输入方向Handler之间的数据传递
15.2.2.1 handler消失了
15.2.2.2 手动编写netty提供的new StringDecoder();这一Handler
15.2.2.3 责任链设计模式
15.2.2.4 ctx上下文对象
15.3 输出方向ChannelOutboundHandlerAdapter
15.3.1 输出方向Handler的顺序
15.3.2 总结 【自己多测试测试，真正底层细节看源码，现在只能测试，然后瞎猜】
15.3.3 ctx和ch的writeAndFlush()
15.4 关于head和tail节点
16.netty服务端编程总结
16.1 服务端关于handler和childHandler
16.2 为什么叫孩子处理器？childHandler
16.3 客户端关于bootstrap.handler
17.作图总结 [橘子哥的图]
18.EmbeddedChannel
15.Handler
Handler是程序员接触最多的地方。最重要的编码环节。
为何说重要，因为我们前面可以知道serverBootstrap.group(new NioEventLoopGroup());服务端在结束了该操作后，实际上就开启了一个线程池在处理连接ACCEPT事件了，等到连接建立后，后续的IO操作会把数据发送过来这个数据实际上就是我们业务中要处理的对象，那么这个处理就是在Handler里面处理的，这个处理就是你业务逻辑的所在地，至于你怎么实现，是发mq还是写库，还是做什么处理，这个是另外一件事。但是这里的Handler就是拿到网络传输数据的地方，也就是以前所说的SocketChannel的地方，而这个Handler通常都是一组，它有很多实现，许多个Handler组成了一个pipeline流水线，每个Handler各司其职，每一种Handler会完成功能中的一件事。通过pipeline流水线来组合各种Handler，实现一系列的功能
15.1 handler的分类
15.1.1 按照方向划分
我们说的handler是有方向的，可以按照读入数据和写出数据的方向划分
读入方向
也就是站在一个角度，数据是流入，举个例子。
我在看服务端的时候，接收客户端过来的数据，对于服务端来说这属于数据流入，也就是读入数据。这就是读入方向。而此时对于客户端来说，数据就是写出方向的。
对于读入数据来说，都属于ChannelInboundHandlerAdapter，我们接收数据的一系列Handler都是这个ChannelInboundHandlerAdapter的子类实现。
写出方向
如果服务端此时要给客户端发数据。这就属于服务端的写出方向，这都属于
ChannelOutboundHandlerAdapter。
这个方向是相对的，你服务端写出数据，对于客户端就是写入。不管怎么看，你如果属于读入(吃数据)，就是在拿到数据之后做ChannelInboundHandlerAdapter的处理，如果你是往出写数据，也就是吐数据，写出去之前，那就是要做ChannelOutboundHandlerAdapter的处理。
15.1.2 handler的结构
pipeline中的各个handler是用双向链表组成的，这个链表中间是你所有的配置的handler，实际上一头一尾还有一个head Handler和一个tail Handler，这两个Handler是Netty自带的Handler，负责来管理这个双向链表
1.Handler作用：用于处理接收数据后 或 发送数据前这两个时间点的数据，是程序员使用netty最重要的战斗场地
2.通过Pipeline把多个handler有机的整合成了一个整体
读取接收数据：ChannelInboundHandler子类
写出数据：ChannelOutboundHandler子类
3.Pipeline中，执行相同种类的Handler有固定顺序，不同种类的Handler不讲究先后顺序
4.Handler传递数据：
super.channelRead(ctx,msg);
super.channelRead(ctx,msg);底层为ctx.fireChannelRead(s)
最后一个Handler不需要传递数据，所以最后一个Handler无需调用该方法
5.pipeline中的各个Handler是使用双向链表组成的，这个链表中间就是你所有的配置的Handler，实际上一头一尾: Head Handler，tail Handler这两个Handler来进行管理整个双向链表
15.2 输入方向ChannelInboundHandlerAdapter
15.2.1 输出方向Handler的顺序
我们说handler的处理是被pipeline流水线管理的。当你把handler一个个的添加到pipeline之后。就是按照你添加的顺序执行的。因为他的添加方法就是addLast，不断的往后面追加，所以就是先来先执行的。
我们来看一下这个顺序性。
- 服务端
package com.messi.netty_core_02.netty04;
import com.sun.security.ntlm.Server;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.ServerSocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class MyNettyServer2 {
private static final Logger log = LoggerFactory.getLogger(MyNettyServer2.class);
public static void main(String[] args) {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(new NioEventLoopGroup(1),new NioEventLoopGroup(8));
DefaultEventLoopGroup defaultEventLoopGroup = new DefaultEventLoopGroup();
serverBootstrap.channel(NioServerSocketChannel.class);
serverBootstrap.childHandler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(defaultEventLoopGroup,"handler1",new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("handler1_____________________________");
log.info("msg:{}",msg);
super.channelRead(ctx,msg) ;
}
});
pipeline.addLast(defaultEventLoopGroup,"handler2",new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("handler2_____________________________");
super.channelRead(ctx,msg) ;
}
});
}
});
serverBootstrap.bind(8000);
}
}- 客户端
package com.messi.netty_core_02.netty04;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LoggingHandler;
import java.net.InetSocketAddress;
public class MyNettyClient2 {
public static void main(String[] args) throws InterruptedException {
Bootstrap bootstrap = new Bootstrap();
bootstrap.channel(NioSocketChannel.class);
NioEventLoopGroup group = new NioEventLoopGroup();
bootstrap.group(group);
bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new LoggingHandler());
pipeline.addLast(new StringEncoder());
}
});
Channel channel = bootstrap.connect(new InetSocketAddress(8000)).sync().channel();
channel.writeAndFlush("leomessi");
System.out.println("MyNettyClient2.main");
group.shutdownGracefully();
}
}- debug客户端进行测试
1.先给客户端设置断点：
2.启动服务端
3.debug启动客户端
4.看控制台打印输出：
5.debug客户端，让客户端多次发数据给服务端
见下图：
你可以观察到一个输出现象就是：无论你客户端发送多少次数据给服务端，处理你这个客户端发送数据的线程都是同一个DefaultEventLoop(处理该NioSocketChannel对应的业务逻辑)或同一个NioEventLoop（处理该NioSocketChannel对应的IO事件逻辑）
所以你可以得出一个结论：客户端与服务端建立连接后，NioServerSocketChannel会分配一个NioSocketChannel给该客户端与服务端作为交互通道。之后，同一个SocketChannel的任务（IO事件任务或业务逻辑任务）会让同一个线程去做，比如说：同一个NioEventLoop处理读写事件，同一个DefaultEventLoop处理业务Handler
- 细节
为什么要引入DefaultEventLoopGroup来处理业务逻辑？
其实很简单，就是为了提高吞吐。因为NioEventLoopGroup通过Reactor模型划分为Boss和Worker分别去处理连接，read，write等IO事件。再复习下Reactor模型的设计吧。当一个客户端请求连接服务端时，服务端的Boss线程进行处理这一次连接，一旦这一次的连接建立后，在程序层面，NioServerSocketChannel就会生成一个NioSocketChannel，Boss线程就会把该连接所对应的NioSocketChannel中IO事件，业务逻辑的交互操作全部交给worker线程去做，我们知道连接只需要建立一次，所以boss线程压力较小，所以boss线程一般只有一个或两个。worker线程压力过大，所以一般根据计算机CPU核数去具体设置，但是呢当客户端过多时，一个worker线程是要进行处理多个客户端连接后的所有操作的，如果一个worker线程在处理完某一个客户端的写出数据的操作后，又得接着去处理该客户端触发的业务逻辑，假设说这个业务逻辑很复杂很耗时，你这个worker线程是不是就阻塞了。。。我们之前就说过worker线程数量是有限制的，所以为了提高系统吞吐量，worker线程只处理IO事件，对于业务处理耗时操作，会异步新开启一个新线程去处理。worker线程会直接返回一个确认告诉客户端，客户端也可以继续向下执行它的业务逻辑，对于客户端而言这也很高效。当异步线程处理完这一业务操作后，需要返回业务处理结果，此时会拿到worker线程给的客户端信息进行回调客户端的回调方法，然后把业务处理结果返回给客户端。可见，异步线程是要开启的，那么这个异步线程怎么做呢？其实就是DefaultEventLoopGroup这一线程池去做啦，为什么呢？还是没说为什么，其实很好理解，使用DefaultEventLoopGroup可以简化开发，最重要的就是更好的和netty体系进行融合！
但是注意：只有显示指定使用DefaultEventLoopGroup的Handler才可以使用defaultEventLoop线程去处理对应的Handler业务，否则还是使用NioEventLoop线程去处理，如下图所示：
ofcourse，当然，假设客户端多次进行发数据给服务端，服务端同样使用相同的defaultEventLoop线程或NioEventLoop线程去处理对应的Handler，不会改变的。
如下这个例子：
15.2.2 多个输入方向Handler之间的数据传递
15.2.2.1 handler消失了
15.2.2.2 手动编写netty提供的new StringDecoder();这一Handler
15.2.2.3 责任链设计模式
多个Handler组成最终的处理链路，这就是责任链设计模式，把各个工作分到每个部分里面，放到链路中挨个处理，你不往下传(不调用super.channelRead(ctx,msg))，就类似于filter过滤器中不往下写(return true)。后面就不会再执行了，链路就断开了。
而且你如果是处于最后一个handler比如我的handler2，他处于最后一个handler了，其实他不往下传了，也就可以不写这个了。写了也没事。反正后面没了。
pipeline中的handler是个双向链表，因为有读入读出，这个后面看看。
总结：
把一个工作做成一个链条，则这一个工作分成若干个步骤，并且每一个步骤都会对数据进行不断的加工处理，会把数据不断的传递给下一个步骤，直到最后一个环节步骤为止。
15.2.2.4 ctx上下文对象
ctx：上下文环境（ChannelHandlerContext类型）
ctx对象管理的是所有Handler，它是Handler运行的环境。ctx管理着数据的传递，也管理着ByteBuf
15.3 输出方向ChannelOutboundHandlerAdapter
15.3.1 输出方向Handler的顺序
前面我们一直说的是输入方向的Hnadler的处理，也就是ChannelInboundHandlerAdapter这个处理。 现在我们再来看一下关于写出数据的操作，也就是ChannelOutboundHandlerAdapter这个处理器操作。我们来看一下代码。
- 客户端代码
package com.messi.netty_core_02.netty04;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LoggingHandler;
import java.net.InetSocketAddress;
public class MyNettyClient2 {
public static void main(String[] args) throws InterruptedException {
Bootstrap bootstrap = new Bootstrap();
bootstrap.channel(NioSocketChannel.class);
NioEventLoopGroup group = new NioEventLoopGroup();
bootstrap.group(group);
bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new LoggingHandler());
pipeline.addLast(new StringEncoder());
}
});
Channel channel = bootstrap.connect(new InetSocketAddress(8000)).sync().channel();
channel.writeAndFlush("leomessi");
System.out.println("MyNettyClient2.main");
group.shutdownGracefully();
}
}- 服务端代码
package com.messi.netty_core_02.netty04;
import com.sun.security.ntlm.Server;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.ServerSocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.CharBuffer;
import java.nio.charset.Charset;
public class MyNettyServer2 {
private static final Logger log = LoggerFactory.getLogger(MyNettyServer2.class);
public static void main(String[] args) {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(new NioEventLoopGroup(1),new NioEventLoopGroup(8));
DefaultEventLoopGroup defaultEventLoopGroup = new DefaultEventLoopGroup();
serverBootstrap.channel(NioServerSocketChannel.class);
serverBootstrap.childHandler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
// pipeline.addLast(new StringDecoder());
pipeline.addLast(defaultEventLoopGroup,"handler1",new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("handler1_____________________________");
ByteBuf byteBuf = (ByteBuf) msg;
CharBuffer decode = Charset.forName("UTF-8").decode(byteBuf.nioBuffer());
log.info("decode:{}",decode);
super.channelRead(ctx,decode);
}
});
pipeline.addLast(defaultEventLoopGroup,"handler2",new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("handler2_____________________________");
log.info("decode:{}",msg);
super.channelRead(ctx,msg) ;
}
});
pipeline.addLast(defaultEventLoopGroup,"handler3",new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("handler3___________________________");
super.channelRead(ctx,msg);
}
});
pipeline.addLast(defaultEventLoopGroup,"handler4",new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("handler4_______________进行write写出，输出处理，其实就是向客户端写数据");
ch.writeAndFlush("服务端向客户端发送数据");
super.write(ctx, msg, promise);
}
});
pipeline.addLast(defaultEventLoopGroup,"handler5",new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("handler5_______________进行write写出，输出处理，其实就是向客户端写数据");
super.write(ctx, msg, promise);
}
});
pipeline.addLast(defaultEventLoopGroup,"handler6",new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("handler6_______________进行write写出，输出处理，其实就是向客户端写数据");
super.write(ctx, msg, promise);
}
});
}
});
serverBootstrap.bind(8000);
}
}- 测试
我们把服务端写出的操作放在handler4中，结果服务端写出对应的Handler没有调用：
修改一下：
我们把服务端写出数据的操作放置到最后一个读取的Handler中：
结果显示可以成功调用写出Handler！
分析：
我们看到执行了输出处理器的操作信息，因为我们接收数据的也就是h3那里写出了数据，就能往下走了，每个写handler里面用了super。write继续往下传递写出数据，但是问题来了。我们add的handler顺序是4,5,6但是执行的顺序却是6,5,4这样的倒序。
实际上我们来看个图。这个图是所有的handler的一个结构，我们说数据从外部进来的时候，数据会从head接收到，然后顺序执行h1 h2 h3这个顺序。
但是输出的操作处理器是从tail开始的，也就是h6 h5 h4这样的顺序。而且当我们先处理接收数据，在处
理写出数据，是按照这样的h1 h2 h3执行完了，看有没有下一个输入，如果没有就直接走到tail了，然后
从tail往前执行输出，执行输出的时候，也就是从tail开始的。然后倒序执行。
注释：head和tail这两个Handler是netty自定义自带的Handler处理器类，负责进行管理整个pipeline流水线，管理所有的Handler处理器类
- 再修改一下
把服务端写出数据的操作放到第一个Handler：
输出：
- 再修改一下
把handler1的向后传递给删除：
把handler5的向后传递给删除：
输出：
由于handler1和handler5的向前传递都断了，所以：读取Handler只输出handler1，输出Handler只输出handler6和handler5
- 修改：基于最原始开头给出的代码，只断开handler4的向前传递
结果表明：对读取Handler无影响
- 修改：基于最原始开头给出的代码，任意修改handler4的位置
根据输出结果可以得出一个结论，可以自己测试一下：
顺序只在同种handler里面产生，不同种类的handler不受顺序影响。你可以这样想，当你此时执行输入Handler时，你把输出Handler都掩盖住，看输入Handler之间的相对位置就是真正的执行顺序！当你执行输出Handler时，同理可得。
15.3.2 总结 【自己多测试测试，真正底层细节看源码，现在只能测试，然后瞎猜】
# 其逻辑一定是从接收到的数据head开始往后走，挨个走过所有的InBound处理器。然后处理完了，从tail走，倒序往前走执行所有的outBound处理器。
# 基于第一条规则，不管你怎么变顺序，哪怕是输出和输入的各种交错add。也是这么个逻辑，输出的处理器顺序不会影响输入的处理器逻辑。而且每个输入的处理器都要super.channelRead(ctx,msg);才能往下一个发。不管123这样的顺序，而是看你add的顺序。而且哪怕你是h1 h4 h2 h3这样，他也是先处理读的h1 h2 h3然后才是h4，因为h4是输出Handler，因为读写处理器是互不影响的。
# 基于前两条规则，顺序只在同种handler里面产生，不同种类的handler不影响
15.3.3 ctx和ch的writeAndFlush()
我们刚才写出数据的时候用的是ch.writeAndFlush("服务端给客户端写的信息...");这个操作，其中ch是NioSocketChannel ch，这是客户端和服务端建立的连接。其实我们的参数里面ctx也有这个写出方法。我们来看一下：
- 服务端代码
package com.messi.netty_core_02.netty04;
import com.sun.security.ntlm.Server;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.ServerSocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.CharBuffer;
import java.nio.charset.Charset;
public class MyNettyServer2 {
private static final Logger log = LoggerFactory.getLogger(MyNettyServer2.class);
public static void main(String[] args) {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(new NioEventLoopGroup(1),new NioEventLoopGroup(8));
DefaultEventLoopGroup defaultEventLoopGroup = new DefaultEventLoopGroup();
serverBootstrap.channel(NioServerSocketChannel.class);
serverBootstrap.childHandler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
// pipeline.addLast(new StringDecoder());
pipeline.addLast(defaultEventLoopGroup,"handler1",new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("handler1_____________________________");
ByteBuf byteBuf = (ByteBuf) msg;
CharBuffer decode = Charset.forName("UTF-8").decode(byteBuf.nioBuffer());
log.info("decode:{}",decode);
super.channelRead(ctx,decode);
}
});
pipeline.addLast(defaultEventLoopGroup,"handler2",new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("handler2_____________________________");
log.info("decode:{}",msg);
super.channelRead(ctx,msg) ;
}
});
pipeline.addLast(defaultEventLoopGroup,"handler3",new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("handler3___________________________");
// ch.writeAndFlush("服务端向客户端发送数据");
ctx.writeAndFlush("服务端向客户端写信息....")
super.channelRead(ctx,msg);
}
});
pipeline.addLast(defaultEventLoopGroup,"handler4",new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("handler4_______________进行write写出，输出处理，其实就是向客户端写数据");
super.write(ctx, msg, promise);
}
});
pipeline.addLast(defaultEventLoopGroup,"handler5",new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("handler5_______________进行write写出，输出处理，其实就是向客户端写数据");
super.write(ctx, msg, promise);
}
});
pipeline.addLast(defaultEventLoopGroup,"handler6",new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("handler6_______________进行write写出，输出处理，其实就是向客户端写数据");
super.write(ctx, msg, promise);
}
});
}
});
serverBootstrap.bind(8000);
}
}- 测试
我们发现h4 h5 h6又没了，原因还是那个图：
之前我们使用ch这一SocketChannel去写出数据，这个ch是本次连接的对象，所以他能感知到所有本次连接的handler，也就是全局的，他是从tail节点往前遍历一直到head节点，但是ctx只是当前h3的上下文对象，他无法感知到其余handler的信息。
ctx代表的是当前这个handler处理器的上下文，也就是h3的上下文，其余的handler不知道h3他的上下文。当h3使用ctx发起写出操作时，他的流程是从当前上下文的handler节点往前走，一直到head节点，所有他会去执行h3前面所有的写出处理器handler。但是此时h3前面没有写出处理器handler，所以就不执行了。那么我们现在修改一下代码，在h3之前注册几个写出handler处理器，比如h5这个handler：
package com.messi.netty_core_02.netty04;
import com.sun.security.ntlm.Server;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.ServerSocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.CharBuffer;
import java.nio.charset.Charset;
public class MyNettyServer2 {
private static final Logger log = LoggerFactory.getLogger(MyNettyServer2.class);
public static void main(String[] args) {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(new NioEventLoopGroup(1),new NioEventLoopGroup(8));
DefaultEventLoopGroup defaultEventLoopGroup = new DefaultEventLoopGroup();
serverBootstrap.channel(NioServerSocketChannel.class);
serverBootstrap.childHandler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
// pipeline.addLast(new StringDecoder());
pipeline.addLast(defaultEventLoopGroup,"handler1",new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("handler1_____________________________");
ByteBuf byteBuf = (ByteBuf) msg;
CharBuffer decode = Charset.forName("UTF-8").decode(byteBuf.nioBuffer());
log.info("decode:{}",decode);
super.channelRead(ctx,decode);
}
});
pipeline.addLast(defaultEventLoopGroup,"handler5",new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("handler5_______________进行write写出，输出处理，其实就是向客户端写数据");
super.write(ctx, msg, promise);
}
});
pipeline.addLast(defaultEventLoopGroup,"handler2",new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("handler2_____________________________");
log.info("decode:{}",msg);
super.channelRead(ctx,msg) ;
}
});
pipeline.addLast(defaultEventLoopGroup,"handler3",new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("handler3___________________________");
// ch.writeAndFlush("服务端向客户端发送数据");
ctx.writeAndFlush("服务端向客户端写信息....");
super.channelRead(ctx,msg);
}
});
pipeline.addLast(defaultEventLoopGroup,"handler4",new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("handler4_______________进行write写出，输出处理，其实就是向客户端写数据");
super.write(ctx, msg, promise);
}
});
pipeline.addLast(defaultEventLoopGroup,"handler6",new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("handler6_______________进行write写出，输出处理，其实就是向客户端写数据");
super.write(ctx, msg, promise);
}
});
}
});
serverBootstrap.bind(8000);
}
}
我们看到如期输出了，来分析一下原因，我们在代码里面依次注册了h1，h5，h2，h3，h4，h6。结构变为如下所示：
当h3这一handler执行完ctx的写出之后，会从当前h3开始执行h3前面的写出处理器，直到head，只会执行h5。因为ctx只具有当前handler3的上下文。
但是把handler3中的ctx.writeXXX修改成ch.writeAndFlush的话，由于ch是整个客户端连接的NioSocketChannel，所以他是面向全局的，也就是他拿到的是tail，然后从tail往前走，去找出全部的输出handler，执行顺序为：h6->h4->h5，前后顺序是相对的。找输出handler时只看输出handler，不看输入！
PS：tail和head是辅助节点，你在代码里面看不到，得去看源码。
当启动服务端的时候，此时就是NioEventLoop里面的线程在做select监听。进入死循环，等客户端连上来才走后面的发送，然后交给handler的流水线做处理
15.4 关于head和tail节点
我们上面说的pipline中有两个handler是netty内置的，叫做head和tail，我们已经知道了，当服务端输入的时候，会按照添加顺序执行inboundHandler，当服务端往出写的时候，会按照添加顺序的反方向执行outBoundHandler。那么问题来了，对于内置handler的head和tail在输入输出的时候到底执行不执行呢？
# 答案是输入的时候执行head->h1->...->tail
# 输出的时候执行的是h6->h4->h5->head
看一下原因，这里的顺序后面源码再看，目前为止只能根据测试结果去总结结论。
我们看一下head节点的源码：
final class HeadContext extends AbstractChannelHandlerContext implements
ChannelOutboundHandler, ChannelInboundHandler
Head的类实现了in和outbound，可见其本质就是OutboundHandler和InboundHandler，所以他其实在输入输出都会和其他的in out一起执行。
我们看一下tail节点的源码：
final class TailContext extends AbstractChannelHandlerContext implements
ChannelInboundHandler
可见tail是一个inbound，所以他只会在输入的时候执行。
补充：
而且这两个节点的类都是内部类，都在DefaultChannelPipeline类中，这就是一个高内聚的体现，如果一个类只在这个类中进行使用，那就在这个类里面定义即可，不对外暴露。
16.netty服务端编程总结
package com.messi.netty_core_02.netty05;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.logging.LoggingHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class NettyServer {
private static final Logger log = LoggerFactory.getLogger(NettyServer.class);
public static void main(String[] args) {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(new NioEventLoopGroup(1),new NioEventLoopGroup(8));
DefaultEventLoopGroup defaultEventLoopGroup = new DefaultEventLoopGroup();
serverBootstrap.channel(NioServerSocketChannel.class);
serverBootstrap.handler(new ChannelInitializer<NioServerSocketChannel>() {
@Override
protected void initChannel(NioServerSocketChannel ch) throws Exception {
}
});
serverBootstrap.childHandler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new StringDecoder());
pipeline.addLast(new LoggingHandler());
pipeline.addLast(defaultEventLoopGroup,"handler1",new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("handler1");
super.channelRead(ctx, msg);
}
});
pipeline.addLast(defaultEventLoopGroup,"handler2",new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("handler2");
ch.writeAndFlush("llll");
super.channelRead(ctx, msg);
}
});
pipeline.addLast(defaultEventLoopGroup,"handler3",new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("handler3");
super.write(ctx, msg, promise);
}
});
pipeline.addLast(defaultEventLoopGroup,"handler4",new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("handler4");
super.write(ctx, msg, promise);
}
});
}
});
serverBootstrap.bind(8000);
}
}package com.messi.netty_core_02.netty05;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LoggingHandler;
import java.net.InetSocketAddress;
public class NettyClient {
public static void main(String[] args) throws InterruptedException {
Bootstrap bootstrap = new Bootstrap() ;
NioEventLoopGroup group = new NioEventLoopGroup();
bootstrap.group(group);
bootstrap.channel(NioSocketChannel.class);
bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
ch.pipeline().addLast(new StringEncoder());
ch.pipeline().addLast(new LoggingHandler());
}
});
Channel channel = bootstrap.connect(new InetSocketAddress(8000)).sync().channel();
channel.writeAndFlush("netty hello");
System.out.println("NettyClient.main");
group.shutdownGracefully();
}
}16.1 服务端关于handler和childHandler
- 我们来看一段代码：
serverBootstrap.handler(new ChannelInitializer<NioServerSocketChannel>() {
@Override
protected void initChannel(NioServerSocketChannel ch) throws Exception {
}
});
serverBootstrap.childHandler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
}
});我们看到这段代码就是我们在服务端启动的时候的设置的东西，但是我们在上面编程的时候只设置了serverBootstrap.childHandler。那么关于handler和childHandler有啥区别呢，下面为了方便我直接简称为h和ch。
我们前面说过serverBootstrap.childHandler也就是ch中实现的是数据的IO处理，也就是对应在NIO中是
socketChannel的功能。其实从他的参数中的new ChannelInitializer()是NioSocketChannel泛型就能知道，他是对应的SC做IO处理的。
而对于我们的服务端，他其实又两个功能，一个是处理IO，一个是accept处理连接的。那么childHandler处理了IO，serverBootstrap.handler就是处理连接的。
其实你看他的参数泛型是NioServerSocketChannel也能知道他对应的是NioServerSocketChannel也就是SSC，他是处理连接的。你不能乱写泛型启动会报错。
其次你也不能不写serverBootstrap.childHandler这一部分，因为服务端你不能不处理IO，你是实际建立IO连接的socketChannel的，所以不写也会报错。
而不写serverBootstrap.handler这个处理连接的代码是不会报错，不影响运行的，他是为了 ServerSocketChannel服务的，你可以写里面也能用pipline，但是他只是为了做连接，没有太多的操作就是accept,源码已经给你封装了，所以你可以不写，但是你要是做一些复杂开发，你要监控SSC的状态，就可以增加pipline在这里，就能监控连接accept的状态和信息。
所以我们就可以知道每一个childHandler就是一个sc，都是一个连接，一个连接就对应一个childHandler，然后他里面的一组pipline的一组handler是每一个childHandler独立拥有一份的，他们不能混着来。其实也好理解，一个连接肯定是自己一组pipline的一组handler。不然就混乱了数据。
16.2 为什么叫孩子处理器？childHandler
每一个客户端过来与ServerSocketChannel进行建立accept连接，ServerSocketChannel会给每一个客户端都建立一个SocketChannel与之对应。所以站在服务端的角度，SocketChannel就是孩子。childHandler()方法是针对服务端-客户端之间建立SocketChannel之后进行的读写事件操作），所以childHandler必须是要建立编码的。你想想你与客户端后续是不是主要进行读写？对吧。然后对于每一个SocketChannel，childHandler都会建立一条pipeline流水线进行处理建立SocketChannel后的读写操作以及其他业务操作，pipeline流水线是由多个Handler处理器类构成（可以为netty自带的Handler也可以为自定义Handler）。
那么与之对应的就是handler()方法，handler方法是处理ServerSocketChannel进行accept()建立连接操作（通常没什么特别的，所以handler()方法可以省略不写）。注：由于accept()建立连接为公共可封装的代码，netty会把ServerSocketChannel.accept()这种代码都给你封装好。
补充：
对于handler()可以省略不写，childHandler()不可以省略不写这一结论，还可以这样理解：
handler()对应的是ServerSocketChannel进行accept建立连接的操作，连接操作是固定的，你想想：客户端与服务端交互，那不必须连接吗？对吧。所以这部分代码可封装。handler()主要处理ServerSocketChannel建立连接时这一时间段的处理，能有啥处理，不就重复建立连接操作的过程吗，直接封装不就完了。在特别复杂的情况下可能会使用handler()做处理，所以可以省略handler()不写。
childHandler()对应的是SocketChannel建立后的读写，读写是不确定的，谁知道你建立连接后是读还是写，还是只读还是只写，所以一般需要自定义，所以childHanlder不可以省略。并且当你读取到数据后，你会在Handler中配置一些处理器类来进行读取数据后的业务处理，写出前同样要进行Handler处理。
每一个SocketChannel都对应一份pipeline流水线（pipeline流水线是许多个handler构成的）
16.3 客户端关于bootstrap.handler
在客户端的代码是这样的：
Bootstrap bootstrap = new Bootstrap();
bootstrap.channel(NioSocketChannel.class);
NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup();
bootstrap.group(nioEventLoopGroup);
bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
ch.pipeline().addLast(new LoggingHandler());
ch.pipeline().addLast(new StringEncoder());
}
});我们看到只有bootstrap.handler，客户端只有这个东西，设计层面上，因为他本身就是一个发起连接的，读写数据的，没有什么接收处理连接的操作（因为客户端是请求服务端建立连接的，而不是服务端请求客户端建立连接！！）。所以他一个就行了。
为什么在客户端代码中编写的handler()方法中实现的泛型是NioSocketChannel而不是服务端中那种NioServerSocketChannel呢？？？？
因为客户端代码是Bootstrap类构建的代码，Bootstrap是ServerBootstrap的父类，ServerBootstrap扩展重写了Bootstrap这个类。所以泛型肯定不一样呀。
17.作图总结 [橘子哥的图]
我理解的就是：
serverBootstrap.handler(多个handler)：其中多个handler是进行客户端-服务端连接操作后，后续的一系列的业务逻辑处理Handler。但是你对于连接后，能有什么操作？没啥操作，所以一般handler方法省略不写。注：连接事件这一网络操作netty都已经帮你封装好了。。。你表层看不见的
serverBootstrap.childhandler(多个handler)：这其中也有多个handler，是客户端-服务端进行IO事件操作后，后续一系列的业务逻辑处理Handler，因为你对于读写事件（读IO后，写IO前这个事件段）一定有其他的业务逻辑可以处理，比如：你读取到的数据可以用来存储MQ还是把它存储起来等，这都属于业务逻辑，你可以把这一系列业务逻辑写在一个个的Handler中。
注：IO事件(read或write)这一网络操作netty都已经帮你封装好了。。。你表层看不见的
对于连接，IO事件，netty都帮你封装好了，你看不见的，你能进行自定义处理的只有Handler操作。所以Handler对于程序员而言，是多么的重要。
18.EmbeddedChannel
前面我们都是启动一个客户端，启动一个服务端然后客户端发消息，服务端或者Inbound接收，或者outbound输出。这样的操作模式，那么我们可以看到我们每次写一个服务端代码都要写一遍handler。属实麻烦，我们可以写成这样。
package com.messi.netty_core_02.netty05;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelOutboundHandlerAdapter;
import io.netty.channel.ChannelPromise;
import io.netty.channel.embedded.EmbeddedChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class TestEmbededHandler {
private static final Logger log = LoggerFactory.getLogger(TestEmbededHandler.class);
public static void main(String[] args) {
ChannelInboundHandlerAdapter h1 = new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("h1 {}",msg) ;
super.channelRead(ctx, msg);
}
};
ChannelInboundHandlerAdapter h2 = new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("h2 {}",msg) ;
super.channelRead(ctx, msg);
}
};
ChannelInboundHandlerAdapter h3 = new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("h3 {}",msg) ;
super.channelRead(ctx, msg);
}
};
ChannelOutboundHandlerAdapter h4 = new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("h4 {}",msg) ;
super.write(ctx, msg, promise);
}
};
ChannelOutboundHandlerAdapter h5 = new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("h5 {}",msg) ;
super.write(ctx, msg, promise);
}
};
ChannelOutboundHandlerAdapter h6 = new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("h6 {}",msg) ;
super.write(ctx, msg, promise);
}
};
//把handler都绑定到Channel上面
EmbeddedChannel channel = new EmbeddedChannel(h1,h2,h3,h4,h5,h6);
//读入操作
channel.writeInbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("llll".getBytes()));
//写出操作
channel.writeOutbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("llll".getBytes()));
}
}
- 测试
- 分析
ChannelInboundHandlerAdapter h1 = new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("h1 {}",msg) ;
super.channelRead(ctx, msg);
}
};
ChannelInboundHandlerAdapter h2 = new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("h2 {}",msg) ;
super.channelRead(ctx, msg);
}
};
ChannelInboundHandlerAdapter h3 = new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("h3 {}",msg) ;
super.channelRead(ctx, msg);
}
};
ChannelOutboundHandlerAdapter h4 = new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("h4 {}",msg) ;
super.write(ctx, msg, promise);
}
};
ChannelOutboundHandlerAdapter h5 = new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("h5 {}",msg) ;
super.write(ctx, msg, promise);
}
};
ChannelOutboundHandlerAdapter h6 = new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("h6 {}",msg) ;
super.write(ctx, msg, promise);
}
};我们可以把上面这段代码分解写成6个类 ，如下所示：
public class InboundHandlerAdapter1 extends ChannelInboundHandlerAdapter {
static Logger logger = LoggerFactory.getLogger(InboundHandlerAdapter1.class);
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws
Exception {
logger.info("HandlerAdapter1 *************msg is {}",msg);
super.channelRead(ctx, msg);
}
}
public class InboundHandlerAdapter2 extends ChannelInboundHandlerAdapter {
static Logger logger = LoggerFactory.getLogger(InboundHandlerAdapter2.class);
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws
Exception {
logger.info("HandlerAdapter2 **************msg is {}",msg);
super.channelRead(ctx, msg);
}
}
public class InboundHandlerAdapter3 extends ChannelInboundHandlerAdapter {
static Logger logger = LoggerFactory.getLogger(InboundHandlerAdapter3.class);
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws
Exception {
logger.info("HandlerAdapter3 *************msg is {}",msg);
super.channelRead(ctx, msg);
}
}
public class OutboundHandlerAdapter4 extends ChannelOutboundHandlerAdapter {
static Logger logger =
LoggerFactory.getLogger(OutboundHandlerAdapter4.class);
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise
promise) throws Exception {
logger.info("HandlerAdapter4 *************msg is {}",msg);
super.write(ctx, msg, promise);
}
}
public class OutboundHandlerAdapter5 extends ChannelOutboundHandlerAdapter {
static Logger logger =
LoggerFactory.getLogger(OutboundHandlerAdapter5.class);
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise
promise) throws Exception {
logger.info("HandlerAdapter5 ************msg is {}",msg);
super.write(ctx, msg, promise);
}
}
public class OutboundHandlerAdapter6 extends ChannelOutboundHandlerAdapter {
static Logger logger =
LoggerFactory.getLogger(OutboundHandlerAdapter6.class);
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise
promise) throws Exception {
logger.info("HandlerAdapter6 *************msg is {}",msg);
super.write(ctx, msg, promise);
}
}然后原始代码只需要把这六个类的对象创建出来然后封装给EmbeddedChannel对象即可。和我们最开始编写的代码是一样的
- 分析2：如果编码改变一下，如下：
// 把handler绑定到Channel上面
EmbeddedChannel embeddedChannel = new EmbeddedChannel(h1,h2,h3,h4,h5,h6);
// 读入操作，和之前的inbound一样
embeddedChannel.writeInbound("inbound netty");
// 写出操作，和之前的outbound一样
embeddedChannel.writeOutbound("outbound netty");我们执行writeInbound就是类似以前的执行多个inboundHandler，按照EmbeddedChannel添加的顺序执行。
而执行writeOutbound就是类似以前的执行多个outboundHandler，按照outBoundHandler的添加反顺序执行。
还有一个注意点：
我们在embeddedChannel.writeInbound("inbound netty");这个操作类似于以前的接收客户端的数据。以前我们客户端是经过编码成为bytebuf发给服务端接收的，然后服务端在走解码器，成为字符串，现在我们就直接写了一个"inbound netty"的字符串，和以前的不太真实一样了，所以需要我们发送的时候编码为bytebuff才能更加真实。
也就是这样像之前那样：
channel.writeInbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("llll".getBytes()));
channel.writeOutbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("llll".getBytes()));
修改后并运行程序，此时则Handler输出的msg类型为：ByteBuf。这是Netty对NIO-ByteBuffer类型的封装