目标
- 服务器启动后，客户端进行连接，服务器端此时要接受客户端请求，并且返回给客户端想要的请求，下面我们的目标就是分析Netty 服务器端启动后是怎么接受到客户端请求的。
- 我们的代码依然与上一篇中用同一个demo， 用io.netty.example下的echo包下的代码
- 我们直接debug模式启动Server端，让后在浏览器输入Http://localhost:8007，接着以下代码分析
源码剖析
- 在上一篇文章Netty启动过程源码分析中，我们知道了服务器最终注册 一个Accept事件等待客户端的连接，同时将NioServerSocketChannel注册到boss单例线程池中，也就是EventLoop如上图左边黄色区域部分
- 因此我们想要分析接受client连接的代码，先找到对应的EventLoop源码，如上图中NioEventLoop 循环，找到如下源码
//代码位置 NioEventLoop --- > run()
@Override
protected void run() {
int selectCnt = 0;
for (;;) {
try {
int strategy;
try {
strategy = selectStrategy.calculateStrategy(selectNowSupplier, hasTasks());
switch (strategy) {
.......
// 处理各种strategy类型
default:
}
} catch (IOException e) {
// If we receive an IOException here its because the Selector is messed up. Let's rebuild
// the selector and retry. https://github.com/netty/netty/issues/8566
rebuildSelector0();
selectCnt = 0;
handleLoopException(e);
continue;
}
selectCnt++;
cancelledKeys = 0;
needsToSelectAgain = false;
final int ioRatio = this.ioRatio;
boolean ranTasks;
if (ioRatio == 100) {
try {
if (strategy > 0) {
//对strategy事件进行处理
processSelectedKeys();
}
} finally {
ranTasks = runAllTasks();
}
} else if (strategy > 0) {
final long ioStartTime = System.nanoTime();
try {
processSelectedKeys();
} finally {
.......
}
} else {
ranTasks = runAllTasks(0); // This will run the minimum number of tasks
}
.......
} catch (CancelledKeyException e) {
.......
} catch (Throwable t) {
handleLoopException(t);
}
.......
}
}
- 如上代码中 strategy 更具请求的类型走不同的策略，最后处理策略的方法是 processSelectedKeys();，我们继续根核心方法 processSelectedKeys();，如下源码
//进入processSelectedKeys ---》processSelectedKeysOptimized(); ---〉processSelectedKey
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
if (!k.isValid()) {
final EventLoop eventLoop;
try {
eventLoop = ch.eventLoop();
} catch (Throwable ignored) {
return;
}
if (eventLoop == this) {
unsafe.close(unsafe.voidPromise());
}
return;
}
try {
int readyOps = k.readyOps();
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
int ops = k.interestOps();
ops &= ~SelectionKey.OP_CONNECT;
k.interestOps(ops);
unsafe.finishConnect();
}
if ((readyOps & SelectionKey.OP_WRITE) != 0) {
ch.unsafe().forceFlush();
}
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
unsafe.read();
}
} catch (CancelledKeyException ignored) {
unsafe.close(unsafe.voidPromise());
}
}
- 第一个if中最事件合法性验证，接着获取readyOps，我们debug得到是16，如下图
- 找到SelectionKey中16 代码的意义
/**
* Operation-set bit for socket-accept operations.
*
* <p> Suppose that a selection key's interest set contains
* <tt>OP_ACCEPT</tt> at the start of a <a
* href="Selector.html#selop">selection operation</a>. If the selector
* detects that the corresponding server-socket channel is ready to accept
* another connection, or has an error pending, then it will add
* <tt>OP_ACCEPT</tt> to the key's ready set and add the key to its
* selected-key set. </p>
*/
public static final int OP_ACCEPT = 1 << 4;
- 术语连接请求，这就是我们拿到了之前用Http://localhost:8007 请求的连接，接着继续跟进代码 EventLoopGroup —> processSelectedKey —> unsafe.read(); 其中unsafe是NioMessageUnsafed，上一篇中有过分析用来处理消息接收
- 继续跟进AbstractNioMessageChannel —> read() ，得到如下源码，删了一些对本次无关的一些代码，如下
public void read() {
assert eventLoop().inEventLoop();
final ChannelConfig config = config();
final ChannelPipeline pipeline = pipeline();
final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
allocHandle.reset(config);
boolean closed = false;
Throwable exception = null;
try {
try {
do {
int localRead = doReadMessages(readBuf);
......
allocHandle.incMessagesRead(localRead);
} while (allocHandle.continueReading());
} catch (Throwable t) {
exception = t;
}
int size = readBuf.size();
for (int i = 0; i < size; i ++) {
readPending = false;
pipeline.fireChannelRead(readBuf.get(i));
}
......
if (exception != null) {
......
}
if (closed) {
......
}
} finally {
......
}
}
}
-
assert eventLoop().inEventLoop(); 判断改eventLoop线程是否当前线程
-
ChannelConfig config = config(); 获取NioServerSocketChannelConfig
-
ChannelPipeline pipeline = pipeline(); 获取DefaultChannelPipeline。他是一个双向链表，可以看到内部包含 LoggingHandler，ServerBootStraptHandler
-
继续跟进 NioServersocketChannel —> doMessage（buf），可以进入到NioServerSocketChannel，找到doMessage方法
protected int doReadMessages(List<Object> buf) throws Exception {
SocketChannel ch = SocketUtils.accept(javaChannel());
try {
if (ch != null) {
buf.add(new NioSocketChannel(this, ch));
return 1;
}
} catch (Throwable t) {
logger.warn("Failed to create a new channel from an accepted socket.", t);
try {
ch.close();
} catch (Throwable t2) {
logger.warn("Failed to close a socket.", t2);
}
}
return 0;
}
-
参数buf是一个静态队列。private final List readBuf = new ArrayList(); 读取boss线程中的NioServerSocketChannel接受到的请求，并且将请求放到buf容器中
-
SocketChannel ch = SocketUtils.accept(javaChannel()); 通过Nio中工具类的建立连接，其实底层是调用了ServerSocketChannelImpl —> accept()方法建立TCP连接，并返回一个Nio中的SocketChannel
-
buf.add(new NioSocketChannel(this, ch)); 将获取到的Nio中SocketCHannel包装成Netty中的NioSocketChannel 并且添加到buf队列中（list）
-
doReadMessages到这分析完。
-
我们回到回到EventLoopGroup —> ProcessSelectedKey
-
循环遍历之前doReadMessage中获取的buf中的所有请求，调用Pipeline的firstChannelRead方法，用于处理这些接受的请求或者其他事件，在read方法中，循环调用ServerSocket的Pipeline的fireChannelRead方法，开始执行管道中的handler的ChannelRead方法，如下
- 继续跟进，进入 pipeline.fireChannelRead(readBuf.get(i)); 一直跟到AbstracChannelHandlerContext —> invokeChannelRead
private void invokeChannelRead(Object msg) {
if (invokeHandler()) {
try {
((ChannelInboundHandler) handler()).channelRead(this, msg);
} catch (Throwable t) {
notifyHandlerException(t);
}
} else {
fireChannelRead(msg);
}
}
- 进入 handler() 中，DefaultChannelPipeline —> handler()
- debug源码可以看到，在管道中添加了多个Handler，分别是：HeadContext，LoggingContext，ServerBootStrapAcceptor，TailContext 因此debug时候会依次进入每一个Handler中。我们重点看ServerBootStrapAcceptor中的channelRead方法
@Override
@SuppressWarnings("unchecked")
public void channelRead(ChannelHandlerContext ctx, Object msg) {
final Channel child = (Channel) msg;
child.pipeline().addLast(childHandler);
setChannelOptions(child, childOptions, logger);
setAttributes(child, childAttrs);
try {
childGroup.register(child).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (!future.isSuccess()) {
forceClose(child, future.cause());
}
}
});
} catch (Throwable t) {
forceClose(child, t);
}
}
- 因为参数msg是NioSocketChannel，此处强制转成channel，
- child.pipeline().addLast(childHandler); 将我们在main方法中设置的EchoServerHandler添加到pipeline的handler链表中
- setChannelOptions 对TCP参数赋值
- setAttributes 设置各种属性
- childGroup.register(child).addListener(…) 将NioSocketChannel注册到 NioEventLoopGroup中的一个EventLoop中，并且添加一个监听器
- 以上NioEventLoopGroup就是我们main方法创建的数组workerGroup
- 进入register方法, MultithreadEventLoopGroup —>register , SingleThreadEventLoop —>register , AbstractChannel —> register，如下
- 首先看MultithreadEventLoopGroup中的register
@Override
public ChannelFuture register(Channel channel) {
return next().register(channel);
}
- 进入next()方法中，最终我们可以追到 DefaultEventExecutorChooserFactory — > PowerOfTwoEventExecutorChooser — > next() 内部类中的next
private static final class PowerOfTwoEventExecutorChooser implements EventExecutorChooser {
private final AtomicInteger idx = new AtomicInteger();
private final EventExecutor[] executors;
PowerOfTwoEventExecutorChooser(EventExecutor[] executors) {
this.executors = executors;
}
@Override
public EventExecutor next() {
return executors[idx.getAndIncrement() & executors.length - 1];
}
}
-
入上我们通过debug可以看到，next返回的就是我们在workerGroup中创建的线程数组中的某一个子线程EventExecutor
-
接下来我们在回到register方法： AbstractChannel —> register 方法，如下：
@Override
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
......
AbstractChannel.this.eventLoop = eventLoop;
if (eventLoop.inEventLoop()) {
register0(promise);
} else {
try {
eventLoop.execute(new Runnable() {
@Override
public void run() {
register0(promise);
}
});
} catch (Throwable t) {
......
}
}
}
- 关键方法register0
private void register0(ChannelPromise promise) {
try {
// check if the channel is still open as it could be closed in the mean time when the register
// call was outside of the eventLoop
if (!promise.setUncancellable() || !ensureOpen(promise)) {
return;
}
boolean firstRegistration = neverRegistered;
doRegister();
neverRegistered = false;
registered = true;
// Ensure we call handlerAdded(...) before we actually notify the promise. This is needed as the
// user may already fire events through the pipeline in the ChannelFutureListener.
pipeline.invokeHandlerAddedIfNeeded();
safeSetSuccess(promise);
pipeline.fireChannelRegistered();
// Only fire a channelActive if the channel has never been registered. This prevents firing
// multiple channel actives if the channel is deregistered and re-registered.
if (isActive()) {
if (firstRegistration) {
pipeline.fireChannelActive();
} else if (config().isAutoRead()) {
// This channel was registered before and autoRead() is set. This means we need to begin read
// again so that we process inbound data.
//
// See https://github.com/netty/netty/issues/4805
beginRead();
}
}
} catch (Throwable t) {
// Close the channel directly to avoid FD leak.
closeForcibly();
closeFuture.setClosed();
safeSetFailure(promise, t);
}
}
- 进入 doRegister(); 方法：AbstractNioChannel —> doRegister
@Override
protected void doRegister() throws Exception {
boolean selected = false;
for (;;) {
try {
selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
return;
} catch (CancelledKeyException e) {
if (!selected) {
// Force the Selector to select now as the "canceled" SelectionKey may still be
// cached and not removed because no Select.select(..) operation was called yet.
eventLoop().selectNow();
selected = true;
} else {
// We forced a select operation on the selector before but the SelectionKey is still cached
// for whatever reason. JDK bug ?
throw e;
}
}
}
}
- 上代码，selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);此处我们将bossGroup中的EventLoop的channel 注册到workerGroup中的EventLoop中的 select中，方法中会得到一个selectionKey
- 我们可以看register方法的注视，如下：
Registers this channel with the given selector, returning a selectionkey.
使用给定的选择器注册此通道，并返回选择键。
- 接着debug，最终会到 AbstractNioChannel 中的doBeginRead方法
@Override
protected void doBeginRead() throws Exception {
// Channel.read() or ChannelHandlerContext.read() was called
final SelectionKey selectionKey = this.selectionKey;
if (!selectionKey.isValid()) {
return;
}
readPending = true;
final int interestOps = selectionKey.interestOps();
if ((interestOps & readInterestOp) == 0) {
selectionKey.interestOps(interestOps | readInterestOp);
}
}
- 此方法比较难进入，包含了几个异步，将之前的断电去掉，再次http请求，可以到这个方法中
- 追到这里，针对客户的连接已经完成，接下来是读取监听事件，也就是bossGroup的连接建立，注册步骤已近完成了，接下来就是workerGroup中的事件处理了
Netty接收请求过程梳理
-
总流程：接收连接 — 》创建一个新的NioSocketChannel —〉 注册到一个WorkerEventLoop上 —》 注册selecotRead事件
- 服务器沦陷Accept事件（文中最开始的那个for循环），获取事件后调用unsafe的read方法，这个unsafe是ServerSocket的内部类，改方法内部由2部分组成
- doReadMessage 用于创建NioSocketChannel对象，改对象包装JDK的NioChannel客户端，该方法创建一个ServerSocketChannel
- 之后执行pipeline.firstChannelRead方法，并且将自己绑定到一个chooser选择器选择的workerGroup中的某个EventLoop上，并且注册一个0（连接），表示注册成功，但是并没有注册1 （读取）
-
上一篇：Netty启动流程源码剖析
