转自：https://zhuanlan.zhihu.com/p/28910259?utm_source=qq&utm_medium=social
首先我们先来看一下netty中的概念，熟悉了基本概念后再来进行netty高性能的分析。
netty中的概念
- ChannelEvent
因为Netty是基于事件驱动的，ChannelEvent就相当于某一个事件，比如说连接成功时打印一句话。 - ChannelPipeline
Pipeline意味着管道、传输途径。也就是说，他是控制ChannelEvent事件分发和传递的。事件在管道中流转，第一站到哪，第二站到哪，到哪是终点，就是用这个ChannelPipeline处理的。比如：开发某项目的流程是先给A设计，然后给B开发。这一套流程相当于是ChannelPipeline。 - ChannelHandler
刚说Pipeline负责把事件分发到相应的站点去处理，这个站点就是指ChannelHandler。事件到了ChannelHandler这里，就要被具体处理了，也就是说，我们具体的业务逻辑一般都是从这里开始的。 - Channel
有了各部门的协调处理，我们还需要一个从整体把握形势的的部门：channel。它够告诉你当前通道的状态，是连通还是关闭、获取通道相关的配置信息、得到Pipeline等。Channel的实现类型决定了你这个通道是同步的还是异步的，例如，NioServerSocketChannel。 - EventLoopGroup
group：群组，Loop：循环，Event：事件，Netty内部都是通过线程在处理各种数据，EventLoopGroup就是用来管理调度他们的，相当于线程池，一般使用NioEventLoopGroup，默认的NioEventLoop个数为cpu核数*2。它是4.x版本提出来的一个新概念，类似于3.x版本中的线程。
小demo
上面了解到了netty中的某些概念，下面通过一个demo来熟悉下netty的使用流程。
server端代码
public class HelloServer {
//服务端监听的端口地址
private static final int port = 9527;
public static void main(String[] args) throws InterruptedException {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
//使用主从Reactor多线程模型
b.group(bossGroup, workerGroup);
//使用nio传输方式
b.channel(NioServerSocketChannel.class);
//添加我们自己的handler处理流程
b.childHandler(new HelloServerInitializer());
// 服务器绑定端口监听
ChannelFuture f = b.bind(port).sync();
// 监听服务器关闭监听
f.channel().closeFuture().sync();
// 可以简写为
/* b.bind(portNumber).sync().channel().closeFuture().sync(); */
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
HelloServerInitializer ：用于定义我们自己的处理流程
public class HelloServerInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
/**
* 获取pipeline，并定义流程
*/
ChannelPipeline pipeline = socketChannel.pipeline();
// 以("\n")为结尾分割的 解码器
pipeline.addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
// 字符串解码 和 编码
pipeline.addLast("decoder", new StringDecoder());
pipeline.addLast("encoder", new StringEncoder());
// 自己的逻辑Handler
pipeline.addLast("handler", new HelloServerHandler());
}
}
HelloServerHandler ：server端自己的handler处理逻辑
public class HelloServerHandler extends SimpleChannelInboundHandler<String> {
/**
* 每一次收到消息时执行此方法
*
* 字符串最后面的"\n"是必须的。
* 因为我们在前面的解码器DelimiterBasedFrameDecoder是一个根据字符串结尾为“\n”来结尾的。
* 假如没有这个字符的话。解码会出现问题。
*
*/
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
// 收到消息直接打印输出
System.out.println(ctx.channel().remoteAddress() + " Say : " + msg);
// 返回客户端消息 - 我已经接收到了你的消息
ctx.writeAndFlush("Received your message !\n");
}
/**
* 覆盖 channelActive 方法
* 在channel被启用的时候触发 (在建立连接的时候)
*
* 这里channeActive的意思是当连接活跃(建立)的时候触发.输出消息源的远程地址。并返回欢迎消息。
*
* 在3.x版本中此处有很大区别。在3.x版本中write()方法是自动flush的。
* 在4.x版本的前面几个版本也是一样的。但是在4.0.9之后修改为WriteAndFlush。
* 普通的write方法将不会发送消息。需要手动在write之后flush()一次
*
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("RamoteAddress : " + ctx.channel().remoteAddress() + " active !");
ctx.write( "Welcome to " + InetAddress.getLocalHost().getHostName() + " service!\n");
ctx.flush();
super.channelActive(ctx);
}
}
server端详解
server首先绑定了端口9527，接着在main函数开始的位置定义了两个工作线程group，一个命名为WorkerGroup，另一个命名为BossGroup，Worker线程用于管理线程为Boss线程服务，都是实例化NioEventLoopGroup。
接下来实例化ServerBootstrap、设置group、设置channel为NioServerSocketChannel，也就是使用nio的传输方式。
接着设置childHandler，在这里使用实例化一个HelloServerInitializer类来实现，继承ChannelInitializer。主要作用是设置相关的字节解码编码器和代码处理逻辑。Handler的具体分析会在下文列出。
设置好Handler绑定端口7878，并调用函数sync()，监听端口等待客户端连接和发送消息，监听端口关闭。
在main函数的结尾用到了EventLoopGroup提供的便捷的方法shutdownGraceFully，我们可以在DefaultEventExecutorGroup的父类MultithreadEventExecutorGroup中看到它的实现代码。关闭了全部EventExecutor数组child里面子元素。相比于3.x版本这是一个比较重大的改动。可以很轻松的全部关闭，而不需要担心出现内存泄露。
client端代码
public class HelloClient {
public static String host = "127.0.0.1";
public static int port = 9527;
public static void main(String[] args) throws InterruptedException, IOException {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
//使用nio传输方式并绑定自己的处理流程
b.group(group).channel(NioSocketChannel.class).handler(new HelloClientInitializer());
// 连接服务端
Channel ch = b.connect(host, port).sync().channel();
// 控制台输入
BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
for (;;) {
String line = in.readLine();
if (line == null) {
continue;
}
/**
* 向服务端发送在控制台输入的文本 并用"\r\n"结尾
* 之所以用\r\n结尾 是因为我们在handler中添加了 DelimiterBasedFrameDecoder 帧解码。
* 这个解码器是一个根据\n符号位分隔符的解码器。所以每条消息的最后必须加上\n否则无法识别和解码
*/
ch.writeAndFlush(line + "\r\n");
}
} finally {
// The connection is closed automatically on shutdown.
group.shutdownGracefully();
}
}
}
HelloClientInitializer : 我们定义的客户端处理流程
public class HelloClientInitializer extends ChannelInitializer<SocketChannel> {
/**
* 初始化HelloClientInitializer时执行此方法，并定义流程
*/
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline = socketChannel.pipeline();
/**
* 这个地方的 必须和服务端对应上。否则无法正常解码和编码
*/
// 以("\n")为结尾分割的 解码器
pipeline.addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
// 字符串解码 和 编码
pipeline.addLast("decoder", new StringDecoder());
pipeline.addLast("encoder", new StringEncoder());
// 自己的逻辑Handler
pipeline.addLast("handler", new HelloClientHandler());
}
}
HelloClientHandler : 客户端自己的处理逻辑
public class HelloClientHandler extends SimpleChannelInboundHandler<String> {
//服务端发送信息时触发此方法
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception {
System.out.println("Server say : " + s);
}
//连接成功时触发此方法
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("Client active ");
super.channelActive(ctx);
}
//服务器关闭时触发此方法
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
System.out.println("Client close ");
super.channelInactive(ctx);
}
}
client端详解
相比于服务端的代码，客户端要精简一些，客户端仅仅只需要一个worker的EventLoopGroup
唯一不同的可能就是客户端的connect方法。服务端的绑定并监听端口，客户端是连接指定的地址。Sync().channel()是为了返回这个连接服务端的channel，并用于后面代码的调用。
当用户输入一行内容并回车之后。循环的读取每一行内容。然后使用writeAndFlush向服务端发送消息。
netty高性能分析
很早看到一个问题：实现rpc高性能的三个话题，传输、协议、线程模型。下面就先对照代码，从这三方面谈谈netty的设计。
- 传输：IO模型在很大程度上决定了框架的性能，相比于bio，netty建议采用异步通信模式，因为nio一个线程可以并发处理N个客户端连接和读写操作，这从根本上解决了传统同步阻塞IO一连接一线程模型，架构的性能、弹性伸缩能力和可靠性都得到了极大的提升。正如代码中所示，使用的是NioEventLoopGroup和NioSocketChannel来提升传输效率。
- 协议：采用什么样的通信协议，对系统的性能极其重要，netty默认提供了对Google Protobuf的支持，也可以通过扩展Netty的编解码接口，用户可以实现其它的高性能序列化框架。
- 线程：netty使用了Reactor线程模型，但Reactor模型不同，对性能的影响也非常大，下面介绍常用的Reactor线程模型有三种，分别如下：
- Reactor单线程模型
单线程模型的线程即作为NIO服务端接收客户端的TCP连接，又作为NIO客户端向服务端发起TCP连接，即读取通信对端的请求或者应答消息，又向通信对端发送消息请求或者应答消息。理论上一个线程可以独立处理所有IO相关的操作，但一个NIO线程同时处理成百上千的链路，性能上无法支撑，即便NIO线程的CPU负荷达到100%，也无法满足海量消息的编码、解码、读取和发送，又因为当NIO线程负载过重之后，处理速度将变慢，这会导致大量客户端连接超时，超时之后往往会进行重发，这更加重了NIO线程的负载，最终会导致大量消息积压和处理超时，NIO线程会成为系统的性能瓶颈。 - Reactor多线程模型
有专门一个NIO线程用于监听服务端，接收客户端的TCP连接请求；网络IO操作(读写)由一个NIO线程池负责，线程池可以采用标准的JDK线程池实现。但百万客户端并发连接时，一个nio线程用来监听和接受明显不够，因此有了主从多线程模型。 - 主从Reactor多线程模型
利用主从NIO线程模型，可以解决1个服务端监听线程无法有效处理所有客户端连接的性能不足问题，即把监听服务端，接收客户端的TCP连接请求分给一个线程池。因此，在代码中可以看到，我们在server端选择的就是这种方式，并且也推荐使用该线程模型。在启动类中创建不同的EventLoopGroup实例并通过适当的参数配置，就可以支持上述三种Reactor线程模型。
因此netty可以作为现今主流的IO通信框架。
源码分析 -- EventLoopGroup
在netty高性能浅析中我们写了一个小demo，其中服务端和客户端在最开始都要建立一个或多个 EventLoopGroup，用来管理调度线程处理各种数据，但它具体是如何实现的呢？让我们进入源码详细看看。
继承结构
源码分析
EventExecutorGroup
先来看下中间层的EventExecutorGroup
public interface EventExecutorGroup extends ScheduledExecutorService, Iterable<EventExecutor> {
boolean isShuttingDown();
Future<?> shutdownGracefully();
Future<?> shutdownGracefully(long var1, long var3, TimeUnit var5);
Future<?> terminationFuture();
/** @deprecated */
@Deprecated
void shutdown();
/** @deprecated */
@Deprecated
List<Runnable> shutdownNow();
EventExecutor next();
Iterator<EventExecutor> iterator();
Future<?> submit(Runnable var1);
<T> Future<T> submit(Runnable var1, T var2);
<T> Future<T> submit(Callable<T> var1);
ScheduledFuture<?> schedule(Runnable var1, long var2, TimeUnit var4);
<V> ScheduledFuture<V> schedule(Callable<V> var1, long var2, TimeUnit var4);
ScheduledFuture<?> scheduleAtFixedRate(Runnable var1, long var2, long var4, TimeUnit var6);
ScheduledFuture<?> scheduleWithFixedDelay(Runnable var1, long var2, long var4, TimeUnit var6);
}
这个接口定义比较清晰，首先它继承了ScheduleEcecutorService ，因此可以把它看成是一个Task的执行器，另外又实现了Iterable接口，就可以看出它的目的也可以当成EventExecutor的容器方法名也清晰代表了其意思，像是一些提交任务，取出可用的EventExecutor，关闭执行器等
EventExecutor
接着来看EventExecutorGroup的其中一个子类EventExecutor
public interface EventExecutor extends EventExecutorGroup {
EventExecutor next();
EventExecutorGroup parent();
boolean inEventLoop();
boolean inEventLoop(Thread var1);
<V> Promise<V> newPromise();
<V> ProgressivePromise<V> newProgressivePromise();
<V> Future<V> newSucceededFuture(V var1);
<V> Future<V> newFailedFuture(Throwable var1);
}
它继承了EventExecutorGroup类，因此也可以将它看成是一个任务的执行器，不过稍微有点不同的是它的next方法返回的是自己的一个引用，还有就是可以判断某线程任务是否被EventLoop执行了。
EventLoopGroup
再来看EventExecutorGroup的另一个子类EventLoopGroup
public interface EventLoopGroup extends EventExecutorGroup {
EventLoop next();
ChannelFuture register(Channel var1);
ChannelFuture register(Channel var1, ChannelPromise var2);
}
它的定义也比较简单，不过这里的next方法返回的是空闲的EventLoop，也可以看出它的特有功能就是注册某channel
EventLoop
最后来看EventLoop的定义：
public interface EventLoop extends EventExecutor, EventLoopGroup {
EventLoopGroup parent();
}
该接口同时继承了EventExecutor和EventLoopGroup，因此可以看做是容器里的一个线程执行器
NioEventLoopGroup
然后我们看一下在上节demo中用到的EventLoopGroup实现类NioEventLoopGroup，他也是netty推荐使用的EventLoopGroup实现类，其实NioEventLoopGroup和NioEventLoop都可以。但是前者使用的是线程池
group中默认的NioEventLoop个数为cpu核数*2
最终对于用户，一般都以EventLoopGroup作为使用的入口，也就是所谓的事件循环，主要是用于事件的注册
