編者注：Netty是Java領(lǐng)域有名的開源網(wǎng)絡(luò)庫，特點(diǎn)是高性能和高擴(kuò)展性则披，因此很多流行的框架都是基于它來構(gòu)建的转晰，比如我們熟知的Dubbo、Rocketmq支示、Hadoop等，針對高性能RPC鄙才，一般都是基于Netty來構(gòu)建颂鸿，比如soft-bolt≡茆郑總之一句話嘴纺，Java小伙伴們需要且有必要學(xué)會(huì)使用Netty并理解其實(shí)現(xiàn)原理。
關(guān)于Netty的入門講解可參考：Netty 入門浓冒，這一篇文章就夠了
Netty的連接處理就是IO事件的處理栽渴，IO事件包括讀事件、ACCEPT事件稳懒、寫事件和OP_CONNECT事件闲擦。
IO事件的處理是結(jié)合ChanelPipeline來做的，一個(gè)IO事件到來，首先進(jìn)行數(shù)據(jù)的讀寫操作墅冷，然后交給ChannelPipeline進(jìn)行后續(xù)處理纯路，ChannelPipeline中包含了channelHandler鏈（head + 自定義channelHandler + tail）。
使用channelPipeline和channelHandler機(jī)制俺榆，起到了解耦和可擴(kuò)展的作用感昼。一個(gè)IO事件的處理，包含了多個(gè)處理流程罐脊，這些處理流程正好對應(yīng)channelPipeline中的channelHandler定嗓。如果對數(shù)據(jù)處理有新的需求，那么就新增channelHandler添加到channelPipeline中萍桌，這樣實(shí)現(xiàn)很6宵溅，以后自己寫代碼可以參考。
說到這里上炎，一般為了滿足擴(kuò)展性要求恃逻，常用2種模式：
- 方法模板模式：模板中定義了各個(gè)主流程，并且留下對應(yīng)hook方法藕施，便于擴(kuò)展寇损。
- 責(zé)任鏈模式：串行模式，可以動(dòng)態(tài)添加鏈數(shù)量和對應(yīng)回調(diào)方法裳食。
netty的channelHandler的channelPipeline可以理解成就是責(zé)任鏈模式矛市，通過動(dòng)態(tài)增加channelHandler可達(dá)到復(fù)用和高擴(kuò)展性目的。
了解netty連接處理機(jī)制之前需要了解下NioEventLoop模型诲祸，其中處理連接事件的架構(gòu)圖如下：
對應(yīng)的處理邏輯源碼為：
// 處理各種IO事件
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
try {
int readyOps = k.readyOps();
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
// OP_CONNECT事件浊吏，client連接上客戶端時(shí)觸發(fā)的事件
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) {
// 注意，這里讀事件和ACCEPT事件對應(yīng)的unsafe實(shí)例是不一樣的
// 讀事件 -> NioByteUnsafe, ACCEPT事件 -> NioMessageUnsafe
unsafe.read();
}
} catch (CancelledKeyException ignored) {
unsafe.close(unsafe.voidPromise());
}
}
從上面代碼來看救氯，事件主要分為3種找田，分別是OP_CONNECT事件、寫事件和讀事件（也包括ACCEPT事件）着憨。下面分為3部分展開：
ACCEPT事件
// NioMessageUnsafe
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 {
do {
// 調(diào)用java socket的accept方法墩衙，接收請求
int localRead = doReadMessages(readBuf);
// 增加統(tǒng)計(jì)計(jì)數(shù)
allocHandle.incMessagesRead(localRead);
} while (allocHandle.continueReading());
} catch (Throwable t) {
exception = t;
}
// readBuf中存的是NioChannel
int size = readBuf.size();
for (int i = 0; i < size; i ++) {
readPending = false;
// 觸發(fā)fireChannelRead
pipeline.fireChannelRead(readBuf.get(i));
}
readBuf.clear();
allocHandle.readComplete();
pipeline.fireChannelReadComplete();
}
連接建立好之后就該連接的channel注冊到workGroup中某個(gè)NIOEventLoop的selector中，注冊操作是在fireChannelRead中完成的享扔，這一塊邏輯就在ServerBootstrapAcceptor.channelRead中底桂。
// ServerBootstrapAcceptor
public void channelRead(ChannelHandlerContext ctx, Object msg) {
final Channel child = (Channel) msg;
// 設(shè)置channel的pipeline handler，及channel屬性
child.pipeline().addLast(childHandler);
setChannelOptions(child, childOptions, logger);
for (Entry<AttributeKey<?>, Object> e: childAttrs) {
child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
}
try {
// 將channel注冊到childGroup中的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);
}
}
READ事件
// NioByteUnsafe
public final void read() {
final ChannelConfig config = config();
final ChannelPipeline pipeline = pipeline();
final ByteBufAllocator allocator = config.getAllocator();
final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
allocHandle.reset(config);
ByteBuf byteBuf = null;
boolean close = false;
try {
do {
byteBuf = allocHandle.allocate(allocator);
// 從channel中讀取數(shù)據(jù)惧眠，存放到byteBuf中
allocHandle.lastBytesRead(doReadBytes(byteBuf));
allocHandle.incMessagesRead(1);
readPending = false;
// 觸發(fā)fireChannelRead
pipeline.fireChannelRead(byteBuf);
byteBuf = null;
} while (allocHandle.continueReading());
// 觸發(fā)fireChannelReadComplete，如果在fireChannelReadComplete中執(zhí)行了ChannelHandlerContext.flush于个，則響應(yīng)結(jié)果返回給客戶端
allocHandle.readComplete();
// 觸發(fā)fireChannelReadComplete
pipeline.fireChannelReadComplete();
if (close) {
closeOnRead(pipeline);
}
} catch (Throwable t) {
if (!readPending && !config.isAutoRead()) {
removeReadOp();
}
}
}
寫事件
正常情況下一般是不會(huì)注冊寫事件的氛魁，如果Socket發(fā)送緩沖區(qū)中沒有空閑內(nèi)存時(shí)，再寫入會(huì)導(dǎo)致阻塞，此時(shí)可以注冊寫事件秀存，當(dāng)有空閑內(nèi)存（或者可用字節(jié)數(shù)大于等于其低水位標(biāo)記）時(shí)捶码，再響應(yīng)寫事件，并觸發(fā)對應(yīng)回調(diào)或链。
if ((readyOps & SelectionKey.OP_WRITE) != 0) {
// 寫事件惫恼，從flush操作來看，雖然之前沒有向socket緩沖區(qū)寫數(shù)據(jù)澳盐，但是已經(jīng)寫入到
// 了chnanel的outboundBuffer中祈纯，flush操作是將數(shù)據(jù)從outboundBuffer寫入到
// socket緩沖區(qū)
ch.unsafe().forceFlush();
}
CONNECT事件
該事件是client觸發(fā)的，由主動(dòng)建立連接這一側(cè)觸發(fā)的叼耙。
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
// OP_CONNECT事件腕窥，client連接上客戶端時(shí)觸發(fā)的事件
int ops = k.interestOps();
ops &= ~SelectionKey.OP_CONNECT;
k.interestOps(ops);
// 觸發(fā)finishConnect事件，其中就包括fireChannelActive事件筛婉，如果有自定義的handler有channelActive方法簇爆，則會(huì)觸發(fā)
unsafe.finishConnect();
}
推薦閱讀
- Netty 入門，這一篇文章就夠了
- Java常見幾種動(dòng)態(tài)代理的對比
- 程序員必看| mockito原理淺析
- Eureka 原理分析
- MQ初窺門徑【面試必看的Kafka和RocketMQ存儲(chǔ)區(qū)別】
- java lambda 深入淺出
歡迎小伙伴關(guān)注【TopCoder】閱讀更多精彩好文爽撒。
