作者： 一字馬胡
轉(zhuǎn)載標(biāo)志 【2017-11-03】
更新日志
| 日期 | 更新內(nèi)容 | 備注 |
|---|---|---|
| 2017-11-03 | 添加轉(zhuǎn)載標(biāo)志 | 持續(xù)更新 |
NI/O C/S通信過(guò)程
下面分別展示了NI/O模式下的客戶端/服務(wù)端編程模型：
Netty是一種基于NI/O的網(wǎng)絡(luò)框架袋倔，網(wǎng)絡(luò)層面的操作只是對(duì)NI/O提供的API的封裝，所以生真，它的服務(wù)端流程和客戶端流程是和NI/O的一致的区拳，對(duì)于客戶端而言溃列，它要做的事情就是連接到服務(wù)端，然后發(fā)送/接收消息。對(duì)于服務(wù)端而言涕蜂，它要bind一個(gè)端口馒稍，然后等待客戶端的連接皿哨，接收客戶端的連接使用的是Accept操作，一個(gè)服務(wù)端需要為多個(gè)客戶端提供服務(wù)纽谒，而每次accept都會(huì)生成一個(gè)新的Channel证膨，Channel是一個(gè)服務(wù)端和客戶端之間數(shù)據(jù)傳輸?shù)耐ǖ溃梢韵蚱鋡rite數(shù)據(jù)鼓黔，或者從中read數(shù)據(jù)央勒。本文將分析Netty框架的Accept流程。
Netty的Accept流程
Netty是一個(gè)較為復(fù)雜的網(wǎng)絡(luò)框架请祖，想要理解它的設(shè)計(jì)需要首先了解NI/O的相關(guān)知識(shí)订歪，為了對(duì)Netty框架有一個(gè)大概的了解，你可以參考Netty線程模型及EventLoop詳解肆捕，該文章詳解解析了Netty中重要的事件循環(huán)處理流程，包含EventLoop的初始化和啟動(dòng)等相關(guān)內(nèi)容盖高。下面首先展示了Netty中的EventLoop的分配模型慎陵，Netty服務(wù)端會(huì)為每一個(gè)新建立的Channel分配一個(gè)EventLoop，并且這個(gè)EventLoop將服務(wù)于這個(gè)Channel得整個(gè)生命周期不會(huì)改變喻奥，而一個(gè)EventLoop可能會(huì)被分配給多個(gè)Channel席纽，也就是一個(gè)EventLoop可能會(huì)服務(wù)于多個(gè)Channel的讀寫事件，這對(duì)于想要使用ThreadLocal的場(chǎng)景需要認(rèn)真考慮撞蚕。
在文章Netty線程模型及EventLoop詳解中已經(jīng)分析了EventLoop的流程润梯，現(xiàn)在從事件循環(huán)的起點(diǎn)開(kāi)始看起，也就是NioEventLoop的run方法甥厦，本文關(guān)心的是Netty的Accept事件纺铭，當(dāng)在Channel上發(fā)生了事件之后，會(huì)執(zhí)行processSelectedKeysPlain方法刀疙，看一下這個(gè)方法：
private void processSelectedKeysPlain(Set<SelectionKey> selectedKeys) {
// check if the set is empty and if so just return to not create garbage by
// creating a new Iterator every time even if there is nothing to process.
// See https://github.com/netty/netty/issues/597
if (selectedKeys.isEmpty()) {
return;
}
Iterator<SelectionKey> i = selectedKeys.iterator();
for (;;) {
final SelectionKey k = i.next();
final Object a = k.attachment();
i.remove();
if (a instanceof AbstractNioChannel) {
processSelectedKey(k, (AbstractNioChannel) a);
} else {
@SuppressWarnings("unchecked")
NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
processSelectedKey(k, task);
}
if (!i.hasNext()) {
break;
}
if (needsToSelectAgain) {
selectAgain();
selectedKeys = selector.selectedKeys();
// Create the iterator again to avoid ConcurrentModificationException
if (selectedKeys.isEmpty()) {
break;
} else {
i = selectedKeys.iterator();
}
}
}
}
接著會(huì)執(zhí)行processSelectedKey這個(gè)方法舶赔，下面是它的細(xì)節(jié)：
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
try {
int readyOps = k.readyOps();
// We first need to call finishConnect() before try to trigger a read(...) or write(...) as otherwise
// the NIO JDK channel implementation may throw a NotYetConnectedException.
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
// remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
// See https://github.com/netty/netty/issues/924
int ops = k.interestOps();
ops &= ~SelectionKey.OP_CONNECT;
k.interestOps(ops);
unsafe.finishConnect();
}
// Process OP_WRITE first as we may be able to write some queued buffers and so free memory.
if ((readyOps & SelectionKey.OP_WRITE) != 0) {
// Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write
ch.unsafe().forceFlush();
}
// Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead
// to a spin loop
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
unsafe.read();
}
} catch (CancelledKeyException ignored) {
unsafe.close(unsafe.voidPromise());
}
}
現(xiàn)在我們可以看到和NI/O一樣的類似OP_READ和OP_ACCEPT之類的東西了，OP_ACCEPT表示的是有Accept事件發(fā)生了谦秧，需要我們處理竟纳，但是發(fā)現(xiàn)好像OP_READ 事件和OP_ACCEPT事件的處理都是通過(guò)一個(gè)read方法進(jìn)行的，我們先來(lái)找到這個(gè)read方法：
-------------------------------------------------
AbstractNioMessageChannel.NioMessageUnsafe.read
-------------------------------------------------
public void read() {
try {
try {
do {
int localRead = doReadMessages(readBuf);
if (localRead == 0) {
break;
}
if (localRead < 0) {
closed = true;
break;
}
allocHandle.incMessagesRead(localRead);
} while (allocHandle.continueReading());
}
int size = readBuf.size();
for (int i = 0; i < size; i ++) {
readPending = false;
pipeline.fireChannelRead(readBuf.get(i));
}
readBuf.clear();
allocHandle.readComplete();
pipeline.fireChannelReadComplete();
}
}
}
本文中的所有代碼都是已經(jīng)處理過(guò)的疚鲤，完整的代碼參考源代碼锥累，本文為了控制篇幅去除了一些不影響閱讀（影響邏輯）的代碼，上面的read方法中有一個(gè)關(guān)鍵的方法doReadMessages集歇，下面是它的實(shí)現(xiàn)：
--------------------------------------------
NioServerSocketChannel. doReadMessages
--------------------------------------------
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;
}
這個(gè)方法就是處理accept類型的事件的桶略，為了更好的理解上面的代碼，下面展示一段在NI/O中服務(wù)端的代碼：
int port = 8676;
ServerSocketChannel serverChannel = ServerSocketChannel.open();
serverChannel.configureBlocking (false);
ServerSocket serverSocket = serverChannel.socket();
serverSocket.bind (new InetSocketAddres(port));
Selector selector = Selector.open();
serverChannel.register (selector, SelectionKey.OP_ACCEPT);
while (true) {
int n = selector.select();
Iterator it = selector.selectedKeys().iterator();
while (it.hasNext()) {
SelectionKey key = (SelectionKey) it.next();
if (key.isAcceptable()) {
ServerSocketChannel server = (ServerSocketChannel) key.channel();
SocketChannel channel = server.accept();
channel.configureBlocking (false);
channel.register (selector, SelectionKey.OP_READ);
}
if (key.isReadable( )) {
processReadEvent(key);
}
it.remove( );
}
}
可以看到，服務(wù)端accept一次就會(huì)產(chǎn)生一個(gè)新的Channel删性，Netty也是亏娜，每次Accept都會(huì)new一個(gè)新的NioSocketChannel，當(dāng)然蹬挺，這個(gè)Channel需要分配一個(gè)EventLoop給他才能開(kāi)始事件循環(huán)维贺，但是Netty服務(wù)端的Accept事件到此應(yīng)該可以清楚流程了，下面分析這個(gè)新的Channel是怎么開(kāi)始事件循環(huán)的巴帮。繼續(xù)看AbstractNioMessageChannel.NioMessageUnsafe.read這個(gè)方法溯泣，其中有一個(gè)句話：
pipeline.fireChannelRead(readBuf.get(i));
現(xiàn)在來(lái)跟蹤一下這個(gè)方法的調(diào)用鏈：
-> ChannelPipeline.fireChannelRead
-> DefaultChannelPipeline.fireChannelRead
-> AbstractChannelHandlerContext.invokeChannelRead
-> ChannelInboundHandler.channelRead
->ServerBootstrapAcceptor.channelRead
上面列出的是主要的調(diào)用鏈路，只是為了分析Accept的過(guò)程榕茧，到ServerBootstrapAcceptor.channelRead這個(gè)方法就可以看到是怎么分配EventLoop給Channel的了垃沦，下面展示了ServerBootstrapAcceptor.channelRead這個(gè)方法的細(xì)節(jié)：
public void channelRead(ChannelHandlerContext ctx, Object msg) {
final Channel child = (Channel) msg;
child.pipeline().addLast(childHandler); 【1】
setChannelOptions(child, childOptions, logger);
for (Entry<AttributeKey<?>, Object> e: childAttrs) {
child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue()); 【2】
}
try {
childGroup.register(child).addListener(new ChannelFutureListener() { 【3】
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (!future.isSuccess()) {
forceClose(child, future.cause());
}
}
});
} catch (Throwable t) {
forceClose(child, t);
}
}
- 【1】首先將服務(wù)端的處理邏輯賦值給新建立的這個(gè)Channel得pipeline，使得這個(gè)新建立的Channel可以得到服務(wù)端提供的服務(wù)
- 【2】屬性賦值
- 【3】將這個(gè)新建立的Channel添加到EventLoopGroup中去用押，這里進(jìn)行EventLoop的分配
下面來(lái)仔細(xì)看一下【3】這個(gè)register方法：
========================================
MultithreadEventLoopGroup.register
========================================
public ChannelFuture register(Channel channel) {
return next().register(channel);
}
========================================
SingleThreadEventLoop.register
========================================
public ChannelFuture register(Channel channel) {
return register(new DefaultChannelPromise(channel, this));
}
========================================
AbstractUnsafe.register
========================================
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);
}
}
}
上面的流程展示了這個(gè)新的Channel是怎么獲得一個(gè)EventLoop的肢簿，而EventLoopGroup分配EventLoop的關(guān)鍵在于MultithreadEventLoopGroup.register這個(gè)方法中的next方法，而這部分的分析已經(jīng)在Netty線程模型及EventLoop詳解中做過(guò)了蜻拨，不再贅述池充。當(dāng)一個(gè)新的連接被服務(wù)端Accept之后，會(huì)創(chuàng)建一個(gè)新的Channel來(lái)維持服務(wù)端與客戶端之間的通信缎讼，而每個(gè)新建立的Channel都會(huì)被分配一個(gè)EventLoop來(lái)實(shí)現(xiàn)事件循環(huán)收夸，本文分析了Netty服務(wù)端Accept的詳細(xì)過(guò)程，至此血崭，對(duì)于Netty的EventLoop卧惜、EventLoopGroup以及EventLoop是如何被運(yùn)行起來(lái)的，以及服務(wù)端是如何Accept新的連接的這些問(wèn)題應(yīng)該都已經(jīng)有答案了夹纫。
