在上篇文章中，我们最后说worker读取到数据，将数据封装成事件增加到管道中流转、处理。今天我们来分析下整个事件的流转过程。
说明：源码版本 netty 3.10+
看老的版本的原因是早期版本封装性相对新版本比较低，这样好读一些。
首页我们看下 NioWorker类中的read方法：
@Override
protected boolean read(SelectionKey k) {
final SocketChannel ch = (SocketChannel) k.channel();
final NioSocketChannel channel = (NioSocketChannel) k.attachment();
final ReceiveBufferSizePredictor predictor =
channel.getConfig().getReceiveBufferSizePredictor();
final int predictedRecvBufSize = predictor.nextReceiveBufferSize();
final ChannelBufferFactory bufferFactory = channel.getConfig().getBufferFactory();
int ret = 0;
int readBytes = 0;
boolean failure = true;
ByteBuffer bb = recvBufferPool.get(predictedRecvBufSize).order(bufferFactory.getDefaultOrder());
try {
while ((ret = ch.read(bb)) > 0) {
readBytes += ret;
if (!bb.hasRemaining()) {
break;
}
}
failure = false;
} catch (ClosedChannelException e) {
// Can happen, and does not need a user attention.
} catch (Throwable t) {
fireExceptionCaught(channel, t);
}
if (readBytes > 0) {
bb.flip();
final ChannelBuffer buffer = bufferFactory.getBuffer(readBytes);
buffer.setBytes(0, bb);
buffer.writerIndex(readBytes);
// Update the predictor.
predictor.previousReceiveBufferSize(readBytes);
// Fire the event.
fireMessageReceived(channel, buffer);
}
if (ret < 0 || failure) {
k.cancel(); // Some JDK implementations run into an infinite loop without this.
close(channel, succeededFuture(channel));
return false;
}
return true;
}
主要是将 SocketChannel 中的数据读取到netty自己定义的ChannelBuffer 中，然后调用 fireMessageReceived方法，向管道中增加上行事件（UpstreamMessageEvent）：
public static void fireMessageReceived(Channel channel, Object message) {
fireMessageReceived(channel, message, null);
}
public static void fireMessageReceived(Channel channel, Object message, SocketAddress remoteAddress) {
channel.getPipeline().sendUpstream(
new UpstreamMessageEvent(channel, message, remoteAddress));
}
我们看下sendUpstream是如何处理的：
public void sendUpstream(ChannelEvent e) {
DefaultChannelHandlerContext head = getActualUpstreamContext(this.head);
if (head == null) {
if (logger.isWarnEnabled()) {
logger.warn(
"The pipeline contains no upstream handlers; discarding: " + e);
}
return;
}
sendUpstream(head, e);
}
private DefaultChannelHandlerContext getActualUpstreamContext(DefaultChannelHandlerContext ctx) {
if (ctx == null) {
return null;
}
DefaultChannelHandlerContext realCtx = ctx;
while (!realCtx.canHandleUpstream()) {
realCtx = realCtx.next;
if (realCtx == null) {
return null;
}
}
return realCtx;
}
找到第一个属于 Upstream 的handler。DefaultChannelHandlerContext是我们向管道中添加handler对象的一个封装，里面引用了handler的处理对象等信息。在管道流模式中，DefaultChannelHandlerContext就是一个个阀门角色。
private final class DefaultChannelHandlerContext implements ChannelHandlerContext {
volatile DefaultChannelHandlerContext next;
volatile DefaultChannelHandlerContext prev;
private final String name;
private final ChannelHandler handler;
private final boolean canHandleUpstream;
private final boolean canHandleDownstream;
private volatile Object attachment;
....
}
回到sendUpstream方法，这个方法最后调用了重载的sendUpstream方法：
void sendUpstream(DefaultChannelHandlerContext ctx, ChannelEvent e) {
try {
((ChannelUpstreamHandler) ctx.getHandler()).handleUpstream(ctx, e);
} catch (Throwable t) {
notifyHandlerException(e, t);
}
}
重context中获取到实际处理类handler对象，这个对象就是我们增加的handler对象如：decoder、encoder、helloHandler 等。然后调用handleUpstream方法，这里我们查看下 SimpleChannelHandler 的 handleUpstream 方法：
public void handleUpstream(
ChannelHandlerContext ctx, ChannelEvent e) throws Exception {
if (e instanceof MessageEvent) {
messageReceived(ctx, (MessageEvent) e);
} else if (e instanceof WriteCompletionEvent) {
WriteCompletionEvent evt = (WriteCompletionEvent) e;
writeComplete(ctx, evt);
} else if (e instanceof ChildChannelStateEvent) {
ChildChannelStateEvent evt = (ChildChannelStateEvent) e;
if (evt.getChildChannel().isOpen()) {
childChannelOpen(ctx, evt);
} else {
childChannelClosed(ctx, evt);
}
} else if (e instanceof ChannelStateEvent) {
ChannelStateEvent evt = (ChannelStateEvent) e;
switch (evt.getState()) {
case OPEN:
if (Boolean.TRUE.equals(evt.getValue())) {
channelOpen(ctx, evt);
} else {
channelClosed(ctx, evt);
}
break;
case BOUND:
if (evt.getValue() != null) {
channelBound(ctx, evt);
} else {
channelUnbound(ctx, evt);
}
break;
case CONNECTED:
if (evt.getValue() != null) {
channelConnected(ctx, evt);
} else {
channelDisconnected(ctx, evt);
}
break;
case INTEREST_OPS:
channelInterestChanged(ctx, evt);
break;
default:
ctx.sendUpstream(e);
}
} else if (e instanceof ExceptionEvent) {
exceptionCaught(ctx, (ExceptionEvent) e);
} else {
ctx.sendUpstream(e);
}
}
这里进行事件类型的判断，然后根据类型去调用相应的方法，如：messageReceived 这个就是我们helloHandle中接收消息的方法。如果想将事件流转到下一个handler中，调用ctx.sendUpstream（）方法及可。
public void sendUpstream(ChannelEvent e) {
DefaultChannelHandlerContext next = getActualUpstreamContext(this.next);
if (next != null) {
DefaultChannelPipeline.this.sendUpstream(next, e);
}
}