1. Netty源码研究笔记(1)——开篇
1.1. Netty介绍
Netty是一个老牌的高性能网络框架。在众多开源框架中都有它的身影,比如:grpc、dubbo、seata等。
里面有着非常多值得学的东西:
-
I/O模型
-
内存管理
-
各种网络协议的实现:http、redis、websocket等等
-
各种各样有趣的技巧的实现:异步、时间轮、池化、内存泄露探测等等。
-
代码风格、设计思想、设计原则等。
1.2. 源码分析方法
我一般是这样进行源码分析的:
-
首先是纵向,通过官方提供的demo,进行debug,并记录在一个完整的生命周期下的调用链上,会涉及到哪些组件。
-
然后对涉及到的组件拿出来,找出它们的顶层定义(接口、抽象类)。通过其模块/包的划分、类注释、定义的方法及其注释,来大致知晓每个组件是做什么的,以及它们在整个框架中的位置是怎样的。
-
第二步完成后,就可以对第一步的调用链流程、步骤、涉及到的组件,进行归纳、划分,从而做到心中有数,知道东南西北了。
-
之后就是横向,对这些归纳出来的组件体系,逐个进行分析。
-
在分析每个组件体系的时候,也是按照先纵向,再横向的步骤:
-
首先是纵向:找出该组件体系中的核心顶层接口、类,然后结合其的所有实现类,捋出继承树,然后弄清楚每个类做的是啥,它是怎么定义的,同一层级的不同实现类之间的区别大致是什么,必要的话,可以将这个继承树记下来,在心中推算几遍。
-
然后是横向:将各个类有选择性地拿出来分析。
-
当然,所谓的纵向,横向,这两个过程实际是互相交织的,也就是说整个流程不一定就分为前后两半:前面一半都是纵向,后面一半都是横向。
通过纵向的分析,我们能发现整个框架可以分成大致哪几个部分,以及有
1.3. 分析前的准备
- 首先在本地建一个对应的分析学习用的项目,比如:learn_netty,用maven管理依赖
- 然后在maven仓库,中找到我们需要的依赖,比如这里我用的是最新的:
<!-- https://mvnrepository.com/artifact/io.netty/netty-all --> <dependency> <groupId>io.netty</groupId> <artifactId>netty-all</artifactId> <version>4.1.77.Final</version> </dependency>
- 将官方提供的demo代码,导入到项目中。
- 学习项目搭建好之后,就尝试编译、运行,没问题后,就命令行
mvn dependency:sources
命令(或者通过IDE)来下载依赖的源代码。 - 可选:在github上,将项目同时clone到本地,如果分析中发现问题或者自己有些优化建议,可以尝试为分析的项目贡献代码。
1.4. 分析示例的代码
以一个简单的EchoServer、EchoClient来研究。
public class EchoServer { private final int port; public EchoServer(int port) { this.port = port; } public static void main(String[] args) throws Exception { new EchoServer(8083).start(); } public void start() throws Exception { final EchoServerHandler serverHandler = new EchoServerHandler(); EventLoopGroup group = new NioEventLoopGroup(); try { ServerBootstrap b = new ServerBootstrap(); b.group(group) .channel(NioServerSocketChannel.class) .localAddress(new InetSocketAddress(port)) .childHandler(new ChannelInitializer<SocketChannel>() { @Override public void initChannel(SocketChannel ch) { ch.pipeline().addLast(serverHandler); } }); ChannelFuture f = b.bind().sync(); f.channel().closeFuture().sync(); } finally { group.shutdownGracefully().sync(); } }
public class EchoServerHandler extends ChannelInboundHandlerAdapter { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { ByteBuf in = (ByteBuf) msg; System.out.println("Server received: " + in.toString(CharsetUtil.UTF_8)); ctx.write(in); } @Override public void channelReadComplete(ChannelHandlerContext ctx) { ctx.writeAndFlush(Unpooled.EMPTY_BUFFER) .addListener(ChannelFutureListener.CLOSE); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { cause.printStackTrace(); ctx.close(); }
public class EchoClient { public static void main(String[] args) throws Exception { connect("127.0.0.1", 8083); } public static void connect(String host, int port) throws Exception { NioEventLoopGroup group = new NioEventLoopGroup(); Bootstrap bootstrap = new Bootstrap(); try { bootstrap.group(group) .channel(NioSocketChannel.class).remoteAddress(new InetSocketAddress(host, port)) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) { ch.pipeline().addLast(new EchoClientHandler()); } }); ChannelFuture f = bootstrap.connect(); f.channel().closeFuture().sync(); } finally { group.shutdownGracefully(); } } }
public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> { @Override public void channelRegistered(ChannelHandlerContext ctx) throws Exception { super.channelRegistered(ctx); } @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { ctx.writeAndFlush(Unpooled.copiedBuffer("Netty Sockets!", CharsetUtil.UTF_8)); } @Override protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception { System.out.println(msg.toString(CharsetUtil.UTF_8)); } }
1.5. 开始分析
分别启动EchoServer、EchoClient,在两个ChannelFuture的位置打断点。
1.5.1. EchoServer启动调用链
进入ServerBootstrap
的bind
方法,发现该方法定义在父类AbstractBootstrap
中:
public ChannelFuture bind() { validate(); SocketAddress localAddress = this.localAddress; if (localAddress == null) { throw new IllegalStateException("localAddress not set"); } return doBind(localAddress); }
接着来看doBind
方法,发现也在AbstractBootstrap
中:
private ChannelFuture doBind(final SocketAddress localAddress) { final ChannelFuture regFuture = initAndRegister(); final Channel channel = regFuture.channel(); if (regFuture.cause() != null) { return regFuture; } if (regFuture.isDone()) { // At this point we know that the registration was complete and successful. ChannelPromise promise = channel.newPromise(); doBind0(regFuture, channel, localAddress, promise); return promise; } else { // Registration future is almost always fulfilled already, but just in case it's not. final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel); regFuture.addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) throws Exception { Throwable cause = future.cause(); if (cause != null) { // Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an // IllegalStateException once we try to access the EventLoop of the Channel. promise.setFailure(cause); } else { // Registration was successful, so set the correct executor to use. // See https://github.com/netty/netty/issues/2586 promise.registered(); doBind0(regFuture, channel, localAddress, promise); } } }); return promise; } }
发现doBind
中主要做了两件事:
-
initAndRegister
(初始化Channel并注册到EventLoop中),这个操作是异步操作,立即返回该操作对应的句柄。 -
拿到
initAndRegister
操作的句柄后,对其进行检查。-
如果
initAndRegister
已完成那么立即进行doBind0
操作(实际的bind
操作),并返回doBind0
操作对应的句柄。 -
如果
initAndRegister
还没有完成,那么就将doBind0
操作异步化:initAndRegister
操作完成后再触发doBind0
。
-
然后我们先看initAndRegister
,它同样在AbstractBootstrap
中:
final ChannelFuture initAndRegister() { Channel channel = null; try { channel = channelFactory.newChannel(); init(channel); } catch (Throwable t) { if (channel != null) { // channel can be null if newChannel crashed (eg SocketException("too many open files")) channel.unsafe().closeForcibly(); // as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t); } // as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor return new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE).setFailure(t); } ChannelFuture regFuture = config().group().register(channel); if (regFuture.cause() != null) { if (channel.isRegistered()) { channel.close(); } else { channel.unsafe().closeForcibly(); } } // If we are here and the promise is not failed, it's one of the following cases: // 1) If we attempted registration from the event loop, the registration has been completed at this point. // i.e. It's safe to attempt bind() or connect() now because the channel has been registered. // 2) If we attempted registration from the other thread, the registration request has been successfully // added to the event loop's task queue for later execution. // i.e. It's safe to attempt bind() or connect() now: // because bind() or connect() will be executed *after* the scheduled registration task is executed // because register(), bind(), and connect() are all bound to the same thread. return regFuture; }
忽略对异常的处理,看到有三个步骤:
-
使用工厂创建一个
channel
-
对这个
channel
进行init
:由子类实现。 -
将创建的
channel
注册(register
)到EventLoopGroup
中,异步操作,将该操作对应的句柄返回。
看完了initAndRegister
后,在回来看doBind0
:
private static void doBind0( final ChannelFuture regFuture, final Channel channel, final SocketAddress localAddress, final ChannelPromise promise) { // This method is invoked before channelRegistered() is triggered. Give user handlers a chance to set up // the pipeline in its channelRegistered() implementation. channel.eventLoop().execute(new Runnable() { @Override public void run() { if (regFuture.isSuccess()) { channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE); } else { promise.setFailure(regFuture.cause()); } } }); }
发现在doBind0
中,最终是通过调用channel
的bind
方法来完成的。而这个动作是包裹成了一个任务,提交给了channel
所注册到的eventloop
,由它来执行。
1.5.2. EchoClient启动调用链
首先进入Bootstrap
的connect
方法中:
public ChannelFuture connect() { validate(); SocketAddress remoteAddress = this.remoteAddress; if (remoteAddress == null) { throw new IllegalStateException("remoteAddress not set"); } return doResolveAndConnect(remoteAddress, config.localAddress()); }
同样忽略validate
,直接看doResolveAndConnect
。
private ChannelFuture doResolveAndConnect(final SocketAddress remoteAddress, final SocketAddress localAddress) { final ChannelFuture regFuture = initAndRegister(); final Channel channel = regFuture.channel(); if (regFuture.isDone()) { if (!regFuture.isSuccess()) { return regFuture; } return doResolveAndConnect0(channel, remoteAddress, localAddress, channel.newPromise()); } else { // Registration future is almost always fulfilled already, but just in case it's not. final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel); regFuture.addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) throws Exception { // Directly obtain the cause and do a null check so we only need one volatile read in case of a // failure. Throwable cause = future.cause(); if (cause != null) { // Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an // IllegalStateException once we try to access the EventLoop of the Channel. promise.setFailure(cause); } else { // Registration was successful, so set the correct executor to use. // See https://github.com/netty/netty/issues/2586 promise.registered(); doResolveAndConnect0(channel, remoteAddress, localAddress, promise); } } }); return promise; } }
我们发现Bootstrap::doResolveAndConnect
和AbstractBootstrap::doBind
类似。意思也是说,在initAndRegister
完成channel
的创建、初始化、绑定到EventLoop
之后再进行实际的操作doResolveAndConnect0
。
于是我们来看doResolveAndConnect0
:
private ChannelFuture doResolveAndConnect0(final Channel channel, SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) { try { final EventLoop eventLoop = channel.eventLoop(); AddressResolver<SocketAddress> resolver; try { resolver = this.resolver.getResolver(eventLoop); } catch (Throwable cause) { channel.close(); return promise.setFailure(cause); } if (!resolver.isSupported(remoteAddress) || resolver.isResolved(remoteAddress)) { // Resolver has no idea about what to do with the specified remote address or it's resolved already. doConnect(remoteAddress, localAddress, promise); return promise; } final Future<SocketAddress> resolveFuture = resolver.resolve(remoteAddress); if (resolveFuture.isDone()) { final Throwable resolveFailureCause = resolveFuture.cause(); if (resolveFailureCause != null) { // Failed to resolve immediately channel.close(); promise.setFailure(resolveFailureCause); } else { // Succeeded to resolve immediately; cached? (or did a blocking lookup) doConnect(resolveFuture.getNow(), localAddress, promise); } return promise; } // Wait until the name resolution is finished. resolveFuture.addListener(new FutureListener<SocketAddress>() { @Override public void operationComplete(Future<SocketAddress> future) throws Exception { if (future.cause() != null) { channel.close(); promise.setFailure(future.cause()); } else { doConnect(future.getNow(), localAddress, promise); } } }); } catch (Throwable cause) { promise.tryFailure(cause); } return promise; }
我们可以看出,doResolveAndConnect0
正如其名:
- 首先获取
channel
所绑定的eventloop
所对应的AddressResolver
(从AddressResolverGroup
)中拿。 - 拿到
AddressResolver
之后,如果它不知道该怎么处理给定的需要连接的地址,或者说这个地址已经被其解析过,那么就直接doConnect
。否则使用AddressResolver
来解析需要连接的地址(异步操作),并将doConnect
操作异步化。
先暂时忽略AddressResolver
,我们来看doConnect
:
private static void doConnect( final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise connectPromise) { // This method is invoked before channelRegistered() is triggered. Give user handlers a chance to set up // the pipeline in its channelRegistered() implementation. final Channel channel = connectPromise.channel(); channel.eventLoop().execute(new Runnable() { @Override public void run() { if (localAddress == null) { channel.connect(remoteAddress, connectPromise); } else { channel.connect(remoteAddress, localAddress, connectPromise); } connectPromise.addListener(ChannelFutureListener.CLOSE_ON_FAILURE); } }); }
我们看到doConnect
和之前的doBind0
一样,最终也是调用channel
的方法,并且将实际的执行交给channel
绑定的eventloop
来执行。
1.6. 总结
就目前debug的调用链上,我们发现涉及到的组件有:
- Bootstrap系列:脚手架,提供给开发人员使用,类似Spring的ApplicationContext
- Channel系列:连接通道
- EventLoopGroup、EventLoop系列:执行器与事件驱动循环,IO模型。
- AddressResolverGroup、AddressResolver系列:地址解析器
- netty自定义的Future、Promise相关:异步化的基础
我们发现netty的操作全程是异步化的,并且最终要解开其原理的庐山真面目,关键还在于提及的eventloop、channel。
此阶段的纵向分析,目前只解开一隅,待我们看看eventloop、channel后,再来解开更大的谜题。