摘要：而這個(gè)正是它是的內(nèi)部類(lèi)，同時(shí)繼承自。獲取最近的并依次執(zhí)行其方法進(jìn)入頭部，并且最終更改了向注冊(cè)了讀事件參考文章總結(jié)如何接入新連接基本流程如上所述，如果有誤，還望各位指正。
歡迎關(guān)注公眾號(hào)：【愛(ài)編程 】2019 贈(zèng)送1T的學(xué)習(xí)資料 哦??！
前文再續(xù)，書(shū)接上一回【NioEventLoop】。
Channel的設(shè)計(jì)
在開(kāi)始分析前，先了解一下Channel的設(shè)計(jì)
頂層Channel接口定義了socket事件如讀、寫(xiě)、連接、綁定等事件，并使用AbstractChannel作為骨架實(shí)現(xiàn)了這些方法。查看器成員變量，發(fā)現(xiàn)大多數(shù)通用的組件，都被定義在這里
第二層AbstractNioChannel定義了以NIO，即Selector的方式進(jìn)行讀寫(xiě)事件的監(jiān)聽(tīng)。其成員變量保存了selector相關(guān)的一些屬性。
第三層內(nèi)容比較多，定義了服務(wù)端channel（左邊繼承了AbstractNioMessageChannel的NioServerSocketChannel）以及客戶(hù)端channel（右邊繼承了AbstractNioByteChannel的NioSocketChannel）。
如何接入新連接?
本文開(kāi)始探索一下Netty是如何接入新連接？主要分為四個(gè)部分
1.檢測(cè)新連接
1.檢測(cè)新連接
Netty服務(wù)端在啟動(dòng)的時(shí)候會(huì)綁定一個(gè)bossGroup，即NioEventLoop,在bind() 綁定端口的時(shí)候注冊(cè)accept（新連接接入）事件。掃描到該事件后，便處理。因此入口從：NioEventLoop#processSelectedKeys() 開(kāi)始。
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
//省略代碼
// Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead
// to a spin loop
//如果當(dāng)前NioEventLoop是workGroup 則可能是OP_READ，bossGroup是OP_ACCEPT
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
//新連接接入以及讀事件處理入口
unsafe.read();
}
}
關(guān)鍵的新連接接入以及讀事件處理入口unsafe.read();
a).這里的unsafe 是在Channel創(chuàng)建過(guò)程的時(shí)候，調(diào)用了父類(lèi)AbstractChannel#AbstractChannel() 的構(gòu)造方法，和pipeline 一起初始化的。
protected AbstractChannel(Channel parent) {
this.parent = parent;
id = newId();
unsafe = newUnsafe();
pipeline = newChannelPipeline();
}
服務(wù)端：NioServerSockeChanne l的父類(lèi)AbstractNioMessageChannel#newUnsafe()創(chuàng)建,可以看到對(duì)應(yīng)的是AbstractNioMessageChannel的內(nèi)部類(lèi)NioMessageUnsafe ;
客戶(hù)端：NioSocketChannel 的的父類(lèi)AbstractNioUnsafe#newUnsafe()創(chuàng)建的話(huà)，它對(duì)應(yīng)的是AbstractNioByteChannel的內(nèi)部類(lèi)NioByteUnsafe
b).unsafe.read()
NioMessageUnsafe.read() 中主要的操作如下：
1.循環(huán)調(diào)用jdk底層的代碼創(chuàng)建channel，并用netty的NioSocketChannel包裝起來(lái)，代表新連接成功接入一個(gè)通道。
private final class NioMessageUnsafe extends AbstractNioUnsafe {
//臨時(shí)存儲(chǔ)讀到的連接
private final List readBuf = new ArrayList();
@Override
public void read() {
assert eventLoop().inEventLoop();
final ChannelConfig config = config();
final ChannelPipeline pipeline = pipeline();
//服務(wù)端接入速率處理器
final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
allocHandle.reset(config);
boolean closed = false;
Throwable exception = null;
try {
try {
//while循環(huán)調(diào)用doReadMessages()創(chuàng)建新連接對(duì)象
do {
//獲取jdk底層的channel,并加入readBuf容器
int localRead = doReadMessages(readBuf);
if (localRead == 0) {
break;
}
if (localRead < 0) {
closed = true;
break;
}
//把讀到的連接做一個(gè)累加totalMessages，默認(rèn)最多累計(jì)讀取16個(gè)連接，結(jié)束循環(huán)
allocHandle.incMessagesRead(localRead);
} while (allocHandle.continueReading());
} catch (Throwable t) {
exception = t;
}
//觸發(fā)readBuf容器內(nèi)所有的傳播事件:ChannelRead 讀事件
int size = readBuf.size();
for (int i = 0; i < size; i ++) {
readPending = false;
pipeline.fireChannelRead(readBuf.get(i));
}
//清空容器
readBuf.clear();
allocHandle.readComplete();
//觸發(fā)傳播事件：ChannelReadComplete，所有的讀事件完成
pipeline.fireChannelReadComplete();
if (exception != null) {
closed = closeOnReadError(exception);
//觸發(fā)傳播事件：exceptionCaught，觸發(fā)異常
pipeline.fireExceptionCaught(exception);
}
if (closed) {
inputShutdown = true;
if (isOpen()) {
close(voidPromise());
}
}
} finally {
// Check if there is a readPending which was not processed yet.
// This could be for two reasons:
// * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
// * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
//
// See https://github.com/netty/netty/issues/2254
if (!readPending && !config.isAutoRead()) {
removeReadOp();
}
}
}
}
而這一段關(guān)鍵代碼邏輯中 int localRead = doReadMessages(readBuf); 它創(chuàng)建jdk底層channel并且用NioSocketChannel包裝起來(lái)，將該channel添加到傳入的容器保存起來(lái)，同時(shí)返回一個(gè)計(jì)數(shù)。
protected int doReadMessages(List buf) throws Exception {
SocketChannel ch = SocketUtils.accept(javaChannel());
try {
if (ch != null) {
//將jdk底層的channel封裝到netty的channel，并存儲(chǔ)到傳入的容器當(dāng)中
//this為服務(wù)端channel
buf.add(new NioSocketChannel(this, ch));
//成功和創(chuàng)建 客戶(hù)端接入的一條通道，并返回
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;
}
2.創(chuàng)建NioSocketChannel
通過(guò)檢測(cè)IO事件輪詢(xún)新連接，當(dāng)前成功檢測(cè)到連接接入事件之后，會(huì)調(diào)用NioServerSocketChannel#doReadMessages() 方法，進(jìn)行創(chuàng)建NioSocketChannel ,即客戶(hù)端channel的過(guò)程。
下面就來(lái)了解一下NioSocketChannel 的主要工作：
public NioSocketChannel(Channel parent, SocketChannel socket) {
super(parent, socket);
//實(shí)例化一個(gè)NioSocketChannelConfig
config = new NioSocketChannelConfig(this, socket.socket());
}
1）、查看NioSocketChannel父類(lèi)構(gòu)造方法,主要是保存客戶(hù)端注冊(cè)的讀事件、channel為成員變量，以及設(shè)置阻塞模式為非阻塞。
public NioSocketChannel(Channel parent, SocketChannel socket) {
super(parent, socket);
//實(shí)例化一個(gè)NioSocketChannelConfig
config = new NioSocketChannelConfig(this, socket.socket());
}
protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) {
//傳入感興趣的讀事件：客戶(hù)端channel的讀事件
super(parent, ch, SelectionKey.OP_READ);
}
protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
super(parent);
//保存客戶(hù)端channel為成員變量
this.ch = ch;
//保存感興趣的讀事件為成員變量
this.readInterestOp = readInterestOp;
try {
//配置阻塞模式為非阻塞
ch.configureBlocking(false);
} catch (IOException e) {
try {
ch.close();
} catch (IOException e2) {
if (logger.isWarnEnabled()) {
logger.warn(
"Failed to close a partially initialized socket.", e2);
}
}
throw new ChannelException("Failed to enter non-blocking mode.", e);
}
}
最后調(diào)用父類(lèi)的構(gòu)造方法，是設(shè)置該客戶(hù)端channel對(duì)應(yīng)的服務(wù)端channel，以及channel的id和兩大組件unsafe和pipeline
protected AbstractChannel(Channel parent) {
//parent為創(chuàng)建次客戶(hù)端channel的服務(wù)端channel（服務(wù)端啟動(dòng)過(guò)程中通過(guò)反射創(chuàng)建的）
this.parent = parent;
id = newId();
unsafe = newUnsafe();
pipeline = newChannelPipeline();
}
2）、再看NioSocketChannelConfig實(shí)例化。主要是保存了javaSocket，并且通過(guò)setTcpNoDelay(true); 禁止了tcp的Nagle算法，目的是為了盡量讓小的數(shù)據(jù)包整合成大的發(fā)送出去,降低延時(shí).
private NioSocketChannelConfig(NioSocketChannel channel, Socket javaSocket) {
super(channel, javaSocket);
calculateMaxBytesPerGatheringWrite();
}
public DefaultSocketChannelConfig(SocketChannel channel, Socket javaSocket) {
super(channel);
if (javaSocket == null) {
throw new NullPointerException("javaSocket");
}
//保存socket
this.javaSocket = javaSocket;
// Enable TCP_NODELAY by default if possible.
if (PlatformDependent.canEnableTcpNoDelayByDefault()) {
try {
//禁止Nagle算法,目的是為了讓小的數(shù)據(jù)包盡量集合成大的數(shù)據(jù)包發(fā)送出去
setTcpNoDelay(true);
} catch (Exception e) {
// Ignore.
}
}
}
3.分配線程和注冊(cè)Selector
服務(wù)端啟動(dòng)初始化的時(shí)候ServerBootstrap#init() ，主要做了一些參數(shù)的配置。其中對(duì)于childGroup,childOptions,childAttrs,childHandler 等參數(shù)被進(jìn)行了多帶帶配置。作為參數(shù)和ServerBootstrapAcceptor 一起，被當(dāng)作一個(gè)特殊的handle，封裝到pipeline中。ServerBootstrapAcceptor 中的eventLoop 為workGroup 。
public class ServerBootstrap extends AbstractBootstrap {
//省略了很多代碼.............
@Override
void init(Channel channel) throws Exception {
//配置AbstractBootstrap.option
final Map, Object> options = options0();
synchronized (options) {
setChannelOptions(channel, options, logger);
}
//配置AbstractBootstrap.attr
final Map, Object> attrs = attrs0();
synchronized (attrs) {
for (Entry, Object> e: attrs.entrySet()) {
@SuppressWarnings("unchecked")
AttributeKey key = (AttributeKey) e.getKey();
channel.attr(key).set(e.getValue());
}
}
//配置pipeline
ChannelPipeline p = channel.pipeline();
//獲取ServerBootstrapAcceptor配置參數(shù)
final EventLoopGroup currentChildGroup = childGroup;
final ChannelHandler currentChildHandler = childHandler;
final Entry, Object>[] currentChildOptions;
final Entry, Object>[] currentChildAttrs;
synchronized (childOptions) {
currentChildOptions = childOptions.entrySet().toArray(newOptionArray(0));
}
synchronized (childAttrs) {
currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(0));
}
p.addLast(new ChannelInitializer() {
@Override
public void initChannel(final Channel ch) throws Exception {
final ChannelPipeline pipeline = ch.pipeline();
//配置AbstractBootstrap.handler
ChannelHandler handler = config.handler();
if (handler != null) {
pipeline.addLast(handler);
}
ch.eventLoop().execute(new Runnable() {
@Override
public void run() {
//配置ServerBootstrapAcceptor,作為Handle緊跟HeadContext
pipeline.addLast(new ServerBootstrapAcceptor(
ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
});
}
});
}
//省略了很多代碼.............
}
可見(jiàn)，整個(gè)服務(wù)端pipeline的結(jié)構(gòu)如下圖所示。bossGroup 控制IO事件的檢測(cè)與處理，整個(gè)bossGroup 對(duì)應(yīng)的pipeline只包括頭（HeadContext ）尾(TailContext )以及中部的ServerBootstrap.ServerBootstrapAcceptor 。
當(dāng)新連接接入的時(shí)候AbstractNioMessageChannel.NioMessageUnsafe#read() 方法被調(diào)用，最終調(diào)用fireChannelRead() ，方法來(lái)觸發(fā)下一個(gè)Handler的channelRead 方法。而這個(gè)Handler正是ServerBootstrapAcceptor
它是ServerBootstrap的內(nèi)部類(lèi)，同時(shí)繼承自ChannelInboundHandlerAdapter 。也是一個(gè)ChannelInboundHandler 。其中channelRead主要做了以下幾件事。
1.為客戶(hù)端channel的pipeline添加childHandler
1）、為客戶(hù)端channel的pipeline添加childHandler
private static class ServerBootstrapAcceptor extends ChannelInboundHandlerAdapter {
private final EventLoopGroup childGroup;
private final ChannelHandler childHandler;
private final Entry, Object>[] childOptions;
private final Entry, Object>[] childAttrs;
private final Runnable enableAutoReadTask;
ServerBootstrapAcceptor(
final Channel channel, EventLoopGroup childGroup, ChannelHandler childHandler,
Entry, Object>[] childOptions, Entry, Object>[] childAttrs) {
this.childGroup = childGroup;
this.childHandler = childHandler;
this.childOptions = childOptions;
this.childAttrs = childAttrs;
//省略了一些代碼。。。。。
@Override
@SuppressWarnings("unchecked")
public void channelRead(ChannelHandlerContext ctx, Object msg) {
//該channel為客戶(hù)端接入時(shí)創(chuàng)建的channel
final Channel child = (Channel) msg;
//添加childHandler
child.pipeline().addLast(childHandler);
//設(shè)置TCP相關(guān)屬性：childOptions
setChannelOptions(child, childOptions, logger);
//設(shè)置自定義屬性:childAttrs
for (Entry, Object> e: childAttrs) {
child.attr((AttributeKey) e.getKey()).set(e.getValue());
}
try {
//選擇NioEventLoop并注冊(cè)Selector
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);
}
}
//省略了一些代碼。。。。。
}
客戶(hù)端channel 的pipeline添加childHandler ，在服務(wù)端EchoServer創(chuàng)建流程中，childHandler的時(shí)候,使用了ChannelInitializer 的一個(gè)自定義實(shí)例。并且覆蓋了其initChannel 方法，改方法獲取到pipeline并添加具體的Handler。查看ChannelInitializer 具體的添加邏輯,handlerAdded 方法。其實(shí)在initChannel 邏輯中，首先是回調(diào)到用戶(hù)代碼執(zhí)行initChannel ,用戶(hù)代碼執(zhí)行添加Handler的添加操作，之后將ChannelInitializer自己從pipeline中刪除 。
public abstract class ChannelInitializer extends ChannelInboundHandlerAdapter {
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
if (ctx.channel().isRegistered()) {
// This should always be true with our current DefaultChannelPipeline implementation.
// The good thing about calling initChannel(...) in handlerAdded(...) is that there will be no ordering
// surprises if a ChannelInitializer will add another ChannelInitializer. This is as all handlers
// will be added in the expected order.
//初始化Channel
if (initChannel(ctx)) {
// We are done with init the Channel, removing the initializer now.
removeState(ctx);
}
}
}
private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
if (initMap.add(ctx)) { // Guard against re-entrance.
try {
//回調(diào)到用戶(hù)代碼
initChannel((C) ctx.channel());
} catch (Throwable cause) {
// Explicitly call exceptionCaught(...) as we removed the handler before calling initChannel(...).
// We do so to prevent multiple calls to initChannel(...).
exceptionCaught(ctx, cause);
} finally {
ChannelPipeline pipeline = ctx.pipeline();
if (pipeline.context(this) != null) {
//刪除本身
pipeline.remove(this);
}
}
return true;
}
return false;
}
}
2）、設(shè)置客戶(hù)端TCP相關(guān)屬性childOptions和自定義屬性childAttrsServerBootstrapAcceptor#init() 方法中已經(jīng)體現(xiàn)
3）、workGroup選擇NioEventLoop并注冊(cè)SelectorAbstractBootstrap#initAndRegister() 方法開(kāi)始，然后跟蹤源碼會(huì)來(lái)到AbstractUnsafe#register() 方法
protected abstract class AbstractUnsafe implements Unsafe {
//省略了一些代碼。。。。。
@Override
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
if (eventLoop == null) {
throw new NullPointerException("eventLoop");
}
if (isRegistered()) {
promise.setFailure(new IllegalStateException("registered to an event loop already"));
return;
}
if (!isCompatible(eventLoop)) {
promise.setFailure(
new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
return;
}
AbstractChannel.this.eventLoop = eventLoop;
if (eventLoop.inEventLoop()) {
register0(promise);
} else {
try {
eventLoop.execute(new Runnable() {
@Override
public void run() {
register0(promise);
}
});
} catch (Throwable t) {
logger.warn(
"Force-closing a channel whose registration task was not accepted by an event loop: {}",
AbstractChannel.this, t);
closeForcibly();
closeFuture.setClosed();
safeSetFailure(promise, t);
}
}
}
//省略了一些代碼。。。。。
}
最后調(diào)用AbstractNioUnsafe#doRegister() 方法通過(guò)jdk的javaChannel().register 完成注冊(cè)功能。
protected abstract class AbstractNioUnsafe extends AbstractUnsafe implements NioUnsafe {
//省略了一些代碼。。。。。
@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;
}
}
}
}
//省略了一些代碼。。。。。
}
4.向Selector注冊(cè)讀事件
a）、入口：ServerBootstrap.ServerBootstrapAcceptor#channelRead()#childGroup.register() ;
public void channelRead(ChannelHandlerContext ctx, Object msg) {
final Channel child = (Channel) msg;
child.pipeline().addLast(childHandler);
setChannelOptions(child, childOptions, logger);
for (Entry, Object> e: childAttrs) {
child.attr((AttributeKey) e.getKey()).set(e.getValue());
}
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);
}
}
b）、實(shí)際上調(diào)用了AbstractChannel.AbstractUnsafe#register0() ，觸發(fā)了通道激活事件;
//觸發(fā)通道激活事件，調(diào)用HeadContent的
pipeline.fireChannelActive();
c）、pipeline 的頭部開(kāi)始，即DefaultChannelPipeline.HeadContext#channelActive() 從而觸發(fā)了readIfIsAutoRead() ;
@Override
public void channelActive(ChannelHandlerContext ctx) {
ctx.fireChannelActive();
readIfIsAutoRead();
}
d）、讀事件將從尾部的TailContent#read()被觸發(fā)，從而依次執(zhí)行ctx.read()，從尾部開(kāi)始，每個(gè)outboundHandler的read()事件都被觸發(fā)。直到頭部。
@Override
public final ChannelPipeline read() {
tail.read();
return this;
}
@Override
public ChannelHandlerContext read() {
//獲取最近的outboundhandler
final AbstractChannelHandlerContext next = findContextOutbound();
EventExecutor executor = next.executor();
//并依次執(zhí)行其read方法
if (executor.inEventLoop()) {
next.invokeRead();
} else {
Tasks tasks = next.invokeTasks;
if (tasks == null) {
next.invokeTasks = tasks = new Tasks(next);
}
executor.execute(tasks.invokeReadTask);
}
return this;
}
e）、進(jìn)入頭部HeadContext#read()，并且最終更改了selectionKey,向selector注冊(cè)了讀事件
HeadContext#read()
@Override
public void read(ChannelHandlerContext ctx) {
unsafe.beginRead();
}
AbstractChannel#beginRead()
@Override
public final void beginRead() {
assertEventLoop();
if (!isActive()) {
return;
}
try {
doBeginRead();
} catch (final Exception e) {
invokeLater(new Runnable() {
@Override
public void run() {
pipeline.fireExceptionCaught(e);
}
});
close(voidPromise());
}
}
AbstractNioMessageChannel#doBeginRead
@Override
protected void doBeginRead() throws Exception {
if (inputShutdown) {
return;
}
super.doBeginRead();
}
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);
}
}
參考文章：
總結(jié)
Netty如何接入新連接基本流程如上所述，如果有誤，還望各位指正。建議先從前兩篇看起比較好理解點(diǎn)。
【Netty】服務(wù)端和客戶(hù)端
最后
如果對(duì) Java、大數(shù)據(jù)感興趣請(qǐng)長(zhǎng)按二維碼關(guān)注一波，我會(huì)努力帶給你們價(jià)值。覺(jué)得對(duì)你哪怕有一丁點(diǎn)幫助的請(qǐng)幫忙點(diǎn)個(gè)贊或者轉(zhuǎn)發(fā)哦。【愛(ài)編碼】 ，回復(fù)2019 有相關(guān)資料哦。
