转自: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作为使用的入口,也就是所谓的事件循环,主要是用于事件的注册