简单Echo案例
代码是netty的源码，我添加了自己理解的中文注释。
了解了Netty的线程模型和组件之后，我们先看看如何写一个简单的Echo案例，后续的源码讲解都基于此案例。以下是服务端的代码：
public final class MyEchoServer {
static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));
public static void main(String[] args) throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
final MyEchoServerHandler serverHandler = new MyEchoServerHandler();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
// 说明服务器端通道的实现类（便于 Netty 做反射处理）
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 100)
// 对服务端的 NioServerSocketChannel 添加 Handler
// LoggingHandler 是 netty 内置的一种 ChannelDuplexHandler，
// 既可以处理出站事件，又可以处理入站事件，即 LoggingHandler
// 既记录出站日志又记录入站日志。
.handler(new LoggingHandler(LogLevel.INFO))
// 对服务端接收到的、与客户端之间建立的 SocketChannel 添加 Handler
.childHandler(new ChannelInitializer() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
p.addLast(serverHandler);
}
});
// Start the server.
ChannelFuture f = b.bind(PORT).sync();
// Wait until the server socket is closed.
f.channel().closeFuture().sync();
} finally {
// Shut down all event loops to terminate all threads.
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
EventLoopGroup的创建与初始化
对应代码
EventLoopGroup bossGroup = new NioEventLoopGroup();
跟踪 NioEventLoopGroup的无参构造
NioEventLoopGroup()
-->
NioEventLoopGroup(int nThreads)
-->
NioEventLoopGroup(int nThreads, Executor executor)
-->
NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider)
-->
NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider,final SelectStrategyFactory selectStrategyFactory)
-->
protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
}
这里能看到，如果构造传入的线程数为0，则使用DEFAULT_EVENT_LOOP_THREADS
值为系统变量 io.netty.eventLoopThreads，没有环境变量就取cpu逻辑线程数*2
例如我的电脑为8核16线程，nThreads = 16 * 2
继续跟踪代码，以下代码有部分省略
protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
EventExecutorChooserFactory chooserFactory, Object... args) {
// 检查线程数量不能小于1
checkPositive(nThreads, "nThreads");
// 这里的 ThreadPerTaskExecutor 实例是下文用于创建 EventExecutor 实例的参数
if (executor == null) {
executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
}
children = new EventExecutor[nThreads];
for (int i = 0; i < nThreads; i ++) {
boolean success = false;
try {
// 创建EventLoop（重点）
children[i] = newChild(executor, args);
success = true;
} catch (Exception e) {
// TODO: Think about if this is a good exception type
throw new IllegalStateException("failed to create a child event loop", e);
} finally {
。。。。。。
}
}
// chooser 的作用是为了实现 next()方法，即从 EventLoopGroup 中挑选
// 一个 NioEventLoop 来处理连接上 IO 事件的方法
chooser = chooserFactory.newChooser(children);
。。。。。。
}
ThreadPerTaskExecutor很简单，实现了Executor接口
public final class ThreadPerTaskExecutor implements Executor {
。。。。。。
@Override
public void execute(Runnable command) {
threadFactory.newThread(command).start();
}
}
这意味着每次执行executor.execute方法，都会开启一个线程。
EventLoop的创建是在newChild中
// 类NioEventLoopGroup
protected EventLoop newChild(Executor executor, Object... args) throws Exception {
// selector工厂
SelectorProvider selectorProvider = (SelectorProvider) args[0];
// 选择策略工厂
SelectStrategyFactory selectStrategyFactory = (SelectStrategyFactory) args[1];
// 拒绝执行处理器（任务添加到队列中失败时调用）
RejectedExecutionHandler rejectedExecutionHandler = (RejectedExecutionHandler) args[2];
EventLoopTaskQueueFactory taskQueueFactory = null;
EventLoopTaskQueueFactory tailTaskQueueFactory = null;
int argsLength = args.length;
if (argsLength > 3) {
taskQueueFactory = (EventLoopTaskQueueFactory) args[3];
}
if (argsLength > 4) {
tailTaskQueueFactory = (EventLoopTaskQueueFactory) args[4];
}
// 创建NioEventLoop并返回
return new NioEventLoop(this, executor, selectorProvider,
selectStrategyFactory.newSelectStrategy(),
rejectedExecutionHandler, taskQueueFactory, tailTaskQueueFactory);
}
NioEventLoopGroup的创建，初始化了selector工厂，选择策略，拒绝执行处理器等。
并创建了同样线程数的NioEventLoop
服务端引导类 ServerBootstrap的创建与设置
对应代码
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)......
public ServerBootstrap group(EventLoopGroup group) {
return group(group, group);
}
public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {
super.group(parentGroup);
if (this.childGroup != null) {
throw new IllegalStateException("childGroup set already");
}
this.childGroup = ObjectUtil.checkNotNull(childGroup, "childGroup");
return this;
}
提供了两个设置EventLoopGroup的方法，也就是parentGroup和childGroup可以是同一个group，
而parentGroup对应线程图中的bossGroup，childGroup对应线程图中的workerGroup
public B channel(Class channelClass) {
return channelFactory(new (
ObjectUtil.checkNotNull(channelClass, "channelClass")
));
}
这里设置的是channel反射工厂，该工厂会使用反射生成 NioServerSocketChannel对象。
创建并绑定channel
对应代码
ChannelFuture f = b.bind(PORT).sync();
准确点说，是负责创建连接（ACCEPT）的channel的创建
AbstractBootstrap#bind(int inetPort) -->
AbstractBootstrap#bind(SocketAddress localAddress) -->
private ChannelFuture doBind(final SocketAddress localAddress) {
// 初始化 NioServerSocketChannel 的实例，并且将其注册到
// bossGroup 中的 EvenLoop 中的 Selector 中，initAndRegister()
// 实例的初始化和注册（此方法是异步的）：
// (1) 初始化：将handler注册进通道，并执行handler的handlerAdded、channelRegistered方法
// (2) 将channel注册进selector
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.
// 若异步过程 initAndRegister()已经执行完毕，则进入该分支
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.
// 若异步过程 initAndRegister()还未执行完毕，则进入该分支
final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
regFuture.addListener(new ChannelFutureListener() {
// 监听 regFuture 的完成事件，完成之后再调用
// doBind0(regFuture, channel, localAddress, promise);
@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;
}
}
initAndRegister方法中主要做了3个操作，channel的创建、初始化以及将channel注册到EventLoop中
final ChannelFuture initAndRegister() {
Channel channel = null;
try {
// 无参构造会创建pipelile
// NioServerSocketChannel
channel = channelFactory.newChannel();
// 初始化相关属性
// 如果是ServerBoottrap，还会设置bossGroup的handler，
// 其中包括ServerBootstrap.handler设置的handler，以及最后添加ServerBootstrapAcceptor
// ServerBootstrapAcceptor就是将channel注册到workerGroup的类
init(channel);
} catch (Throwable t) {
。。。。。。
}
// 将channel注册进selector(监听ACCEPT事件)
// 依然是通过开启eventLoop线程的方式进行注册
// MultithreadEventLoopGroup
ChannelFuture regFuture = config().group().register(channel);
if (regFuture.cause() != null) {
if (channel.isRegistered()) {
channel.close();
} else {
channel.unsafe().closeForcibly();
}
}
return regFuture;
}
重新回到AbstractBootstrap#doBind方法中
// 如果上面的initAndRegister方法执行完毕（异步执行的）,则执行doBind0
if (regFuture.isDone()) {
// At this point we know that the registration was complete and successful.
// 若异步过程 initAndRegister()已经执行完毕，则进入该分支
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.
// 若异步过程 initAndRegister()还未执行完毕，则进入该分支
final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
regFuture.addListener(new ChannelFutureListener() {
// 监听 regFuture 的完成事件，完成之后再调用
// doBind0(regFuture, channel, localAddress, promise);
@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;
}
上面这段if/esle做了同一件是，就是自行doBind0方法，区别在于如果initAndRegister执行完毕，则执行调用doBind0，否则添加监听器，等执行完成触发调用doBind0
继续看doBind0
// 类AbstractBootstrap
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.
// execute方法会将这个Runnable加入到taskQueue中，并开线程执行EventLoop的run方法(死循环)
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());
}
}
});
}
channel.eventLoop().execute这个后面再说，可以看到，里面的逻辑是调用channel.bind在实现绑定的，继续跟踪
AbstractChannel#bind(SocketAddress localAddress, ChannelPromise promise)
-->
// 类AbstractChannel
public final ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
// tail就是TailContext
return tail.bind(localAddress, promise);
}
-->
// 类AbstractChannel
public ChannelFuture bind(final SocketAddress localAddress, final ChannelPromise promise) {
ObjectUtil.checkNotNull(localAddress, "localAddress");
if (isNotValidPromise(promise, false)) {
// cancelled
return promise;
}
// 在管道中从当前handlerContext往前查找实现了bind方法的handlerContext
final AbstractChannelHandlerContext next = findContextOutbound(MASK_BIND);
EventExecutor executor = next.executor();
if (executor.inEventLoop()) {
// 执行handlerContext的bind方法
next.invokeBind(localAddress, promise);
} else {
safeExecute(executor, new Runnable() {
@Override
public void run() {
next.invokeBind(localAddress, promise);
}
}, promise, null, false);
}
return promise;
}
-->
// 类AbstractChannel
private void invokeBind(SocketAddress localAddress, ChannelPromise promise) {
if (invokeHandler()) {
try {
((ChannelOutboundHandler) handler()).bind(this, localAddress, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
}
} else {
bind(localAddress, promise);
}
}
从上面可以看到，最终会执行handler的bind方法，拿 LoggingHandler的bind方法举例
// 类LoggingHandler
public void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) throws Exception {
if (logger.isEnabled(internalLevel)) {
logger.log(internalLevel, format(ctx, "BIND", localAddress));
}
ctx.bind(localAddress, promise);
}
ctx.bind(localAddress, promise)是不是很眼熟，没错，就是AbstractChannel#bind(final SocketAddress localAddress, final ChannelPromise promise)
就像一个循环，每一次都在当前handlerContext往前找有实现了bind方法的handlerContext，执行bind，然后继续往前找。
最终找到管道中的第一个handler，也就是 HeadContext，看看它实现的bind方法
// 类HeadContext
public void bind(
ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) {
unsafe.bind(localAddress, promise);
}
-->
AbstractChannel#bind(final SocketAddress localAddress, final ChannelPromise promise)
-->
NioServerSocketChannel#doBind(SocketAddress localAddress)
最后，还是Java NIO的API来绑定
参考资料：
《Netty in Action》，Norman Maurer
《Scalable IO in Java》，Doug Lea
Original: https://www.cnblogs.com/konghuanxi/p/16381317.html
Author: 王谷雨
Title: Netty源码解读（二）-服务端源码讲解
原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/583264/
转载文章受原作者版权保护。转载请注明原作者出处！