Netty 源码解析系列-客户端连接接入及读I/O解析

前言

    
上一章节《Netty 源码解析系列-服务端启动流程解析》
我们完成了服务端启动，那么服务端启动完成后，客户端接入以及读I/O 事件是怎么哪里开始的？以及 netty 的 boss 线程接收到客户端 TCP 连接请求后如何将链路注册到 worker 线程池？带着这些疑问，我们开始客户端连接接入及读写 I/O 解析。

1.NioEventLoop run()开始

    processSelectedKeys();

private void processSelectedKeys() {
    if (selectedKeys != null) {
        processSelectedKeysOptimized(selectedKeys.flip());
    } else {
        processSelectedKeysPlain(selector.selectedKeys());
    }
}

    
根据 selectedKeys 是否为空，判断是否采用优化后的 selectedKeys ，进到 processSelectedKeysOptimized。

private void processSelectedKeysOptimized(SelectionKey[] selectedKeys) {
    for (int i = 0;; i ++) {
        final SelectionKey k = selectedKeys[i];
        if (k == null) {
            break;
        }
        selectedKeys[i] = null;

        final Object a = k.attachment();

        if (a instanceof AbstractNioChannel) {
               processSelectedKey(k, (AbstractNioChannel) a);
        } else {
               @SuppressWarnings("unchecked")
               NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
               processSelectedKey(k, task);
        }
            ...
}
}

    
k.attachment() 获取附加的对象，那我们是在哪里附加上去的呢？上一篇《Netty 源码解析-服务端启动流程解析》注册时 attach 上去的对象,其实就是 NioServerSocketChannel 自身。

@Override
protected void doRegister() throws Exception {
    boolean selected = false;
    for (;;) {
    ...
    selectionKey = javaChannel().register(eventLoop().selector, 0, this);
    ...        
    }
}

    
我们再回到 k.attachment() ，在取出附加对象后，判断类型是否为 AbstractNioChannel ，从这里我们可以看到，不是附加 AbstractNioChannel 类型，那么就是附加的 NioTask 对象，在这里我们只看关于 AbstractNioChannel 的，进到 processSelectedKey() 方法。

private static void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
    final NioUnsafe unsafe = ch.unsafe();
    ...
    int readyOps = k.readyOps();
    if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
         unsafe.read();
    if (!ch.isOpen()) {
         return;
    }
    if ((readyOps & SelectionKey.OP_WRITE) != 0) {
          ch.unsafe().forceFlush();
    }
    if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
          int ops = k.interestOps();
          ops &= ~SelectionKey.OP_CONNECT;
          k.interestOps(ops);
          unsafe.finishConnect();
    }
    ...
}

    
当操作类型是读操作或者连接操作，进入 unsafe.read() ,有两个类实现了这个方法，一个是 AbstractNioByteChannel 的内部类 NioByteUnsafe ，一个是 AbstractNioMessageChannel 的内部类 NioMessageUnsafe ，这两个类都是 NioUnsafe 实现类 AbstractNioChannel 的子类，那到底是哪一个子类？我们看看 NioServerSocketChannel 创建时是创建的 NioByteUnsafe 还是 NioMessageUnsafe。

public class NioServerSocketChannel extends AbstractNioMessageChannel
                             implements io.netty.channel.socket.ServerSocketChannel {
        public NioServerSocketChannel() {
                this(newSocket(DEFAULT_SELECTOR_PROVIDER));
        }
}

public NioServerSocketChannel(ServerSocketChannel channel) {
        super(null, channel, SelectionKey.OP_ACCEPT);
        config = new NioServerSocketChannelConfig(this, javaChannel().socket());
}

public abstract class AbstractNioMessageChannel extends AbstractNioChannel {
    protected AbstractNioMessageChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
          super(parent, ch, readInterestOp);
      }
}

public abstract class AbstractNioChannel extends AbstractChannel {
    protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
            super(parent);
    }
}


public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {
        protected AbstractChannel(Channel parent) {
                this.parent = parent;
                unsafe = newUnsafe();
                pipeline = new DefaultChannelPipeline(this);
        }
}

    
NioServerSocketChannel 是 AbstractNioMessageChannel 的子类，AbstractNioMessageChannel 是 AbstractNioChannel 的子类，newUnsafe() 是 AbstractChannel 的抽象方法，那么我们从这里就知道，AbstractNioMessageChannel 实现了 AbstractChannel的newUnsafe() 抽象方法，由此判断，我们选择 AbstractNioMessageChannel 的内部类 NioMessageUnsafe 的 read()。

private final class NioMessageUnsafe extends AbstractNioUnsafe {
    private final List<Object> readBuf = new ArrayList<Object>();
    @Override
    public void read() {
        ...
        for (;;) {
           int localRead = doReadMessages(readBuf);
           ...
    }
    setReadPending(false);
    int size = readBuf.size();
    for (int i = 0; i < size; i ++) {
            pipeline.fireChannelRead(readBuf.get(i));
    }
    readBuf.clear();
    pipeline.fireChannelReadComplete();
    ...
}

    
这里分两部分，一个是处理消息，一个是处理事件。
         
1.处理消息

@Override
protected int doReadMessages(List<Object> buf) throws Exception {
    SocketChannel ch = javaChannel().accept();
    ...
    buf.add(new NioSocketChannel(this, ch));
    return 1;
    ...
}

    
接受了一个客户端 SocketChannel，封装到NioSocketChannel，添加到list集合中，我们看看new NioSocketChannel()。

public class NioSocketChannel extends AbstractNioByteChannel implements io.netty.channel.socket.SocketChannel {
    public NioSocketChannel(Channel parent, SocketChannel socket) {
            super(parent, socket);
            config = new NioSocketChannelConfig(this, socket.socket());
    }
}

public abstract class AbstractNioByteChannel extends AbstractNioChannel {
    protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) {
            super(parent, ch, SelectionKey.OP_READ);
    }

    @Override
    protected AbstractNioUnsafe newUnsafe() {
        return new NioByteUnsafe();
    }

    protected class NioByteUnsafe extends AbstractNioUnsafe {
        @Override
        public final void read() {
            ...
        }
    }
}

    
AbstractNioByteChannel 也继承了 AbstractNioChannel ，并实现了 newUnsafe() 方法,由此我们可以推断出当客户端第一次连接时，走的是 AbstractNioMessageChannel 的子类 NioMessageUnsafe的read() ，当客户端发送数据时，走的是 AbstractNioByteChannel 的内部类 AbstractNioUnsafe 的 read() 方法。
        
2.处理事件

   for (int i = 0; i < size; i ++) {
           pipeline.fireChannelRead(readBuf.get(i));
     }


@Override
public ChannelPipeline fireChannelRead(Object msg) {
    head.fireChannelRead(msg);
    return this;
}

@Override
public ChannelHandlerContext fireChannelRead(final Object msg) {
    final AbstractChannelHandlerContext next = findContextInbound();
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        next.invokeChannelRead(msg);
    } else {
        executor.execute(new OneTimeTask() {
            @Override
            public void run() {
                next.invokeChannelRead(msg);
            }
        });
    }
    return this;
}

    

image

    
从next的debug可以看出，当前handler是ServerBootstrapAcceptor这个处理器来处理ChannelRead() 方法，如果看了上一篇《Netty 源码解析-服务端启动流程解析》就会知道，这是在init() 方法中pipeline.addLast(new ServerBootstrapAcceptor())。为什么不是p.addLast(new ChannelInitializer<Channel>())? 因为在ChannelInitializer.channelRegistered() 会删除当前initChannel 处理器。

public final void channelRegistered(ChannelHandlerContext ctx) throws Exception {
    initChannel((C) ctx.channel());
    ctx.pipeline().remove(this);
    ctx.fireChannelRegistered();
}

    
我们继续看ServerBootstrapAcceptor的ChannelRead() 方法。

@Override
@SuppressWarnings("unchecked")
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    final Channel child = (Channel) msg;
    child.pipeline().addLast(childHandler);
    for (Entry<ChannelOption<?>, Object> e: childOptions) {
       try {
          if (!child.config().setOption((ChannelOption<Object>) e.getKey(), e.getValue())) {
              logger.warn("Unknown channel option: " + e);
          }
        } catch (Throwable t) {
              logger.warn("Failed to set a channel option: " + child, t);
        }
    }
    for (Entry<AttributeKey<?>, Object> e: childAttrs) {
         child.attr((AttributeKey<Object>) 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);
     }
}

    
这里分三个步骤
        
(1) 将childHandler添加到处理器上，这个从哪里来？就是从最开始设置serverBootstrap.childHandler(new IOChannelInitialize())。
        
(2) 设置一些参数。
        
(3) work线程池register客户端的channel。

@Override
public ChannelFuture register(Channel channel) {
    return next().register(channel);
}


@Override
public EventLoop next() {
    return (EventLoop) super.next();
}

@Override
public EventExecutor next() {
    return chooser.next();
}

private final class GenericEventExecutorChooser implements EventExecutorChooser {
    @Override
    public EventExecutor next() {
        return children[Math.abs(childIndex.getAndIncrement() % children.length)];
    }
}

    
从work线程池选一个线程来执行register。

@Override
public ChannelFuture register(Channel channel) {
    return register(channel, new DefaultChannelPromise(channel, this));
}

@Override
public ChannelFuture register(final Channel channel, final ChannelPromise promise) {
     ...
        channel.unsafe().register(this, promise);
        return promise;
}

@Override
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
     ...
     AbstractChannel.this.eventLoop = eventLoop;
     if (eventLoop.inEventLoop()) {
     register0(promise);
     } else {
          try {
              eventLoop.execute(new OneTimeTask() {
              @Override
              public void run() {
                 register0(promise);
              }
              });
           } catch (Throwable t) {
                ...
           }
     }
}

@Override
protected void doRegister() throws Exception {
    ...
    selectionKey = javaChannel().register(eventLoop().selector, 0, this);
    ...
}

    
后面的流程和上一篇《Netty 源码解析-服务端启动流程解析》
的注册流程是一样的，区别在于服务启动时注册是在boss线程池任务队列中执行注册，客户端新接入注册是在work线程池任务队列中执行register0() 方法，并将work线程池的selector注册到Java NIO
到这里，我们就可以回答开篇的的几个问题：客户端是如何接入？netty的boss线程接收到客户端TCP连接请求后如何将链路注册到worker线程池？
现在我们还剩下一个问题：读写I/O事件是怎么哪里开始的？
    
我们回到文章开头

private void processSelectedKeysOptimized(SelectionKey[] selectedKeys) {
    for (int i = 0;; i ++) {
        final SelectionKey k = selectedKeys[i];
        if (k == null) {
            break;
        }
        selectedKeys[i] = null;

        final Object a = k.attachment();

        if (a instanceof AbstractNioChannel) {
               processSelectedKey(k, (AbstractNioChannel) a);
        } else {
               @SuppressWarnings("unchecked")
               NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
               processSelectedKey(k, task);
        }
             ...
    } 
}

    
前面boss线程池在这里完成了客户端连接接入，并将链路注册到worker线程池任务队列，添加了read事件的监听，那么现在work线程不停循环selectedKeys中有没有待处理的事件，当有待处理事件，那么会执行processSelectedKey() 方法。

private static void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
    ...
    int readyOps = k.readyOps();
    if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
            unsafe.read();
            ...
    }
    ...
}

    
在这里unsafe.read() 选择AbstractNioByteChannel的read()。

@Override
public final void read() {
    final ChannelConfig config = config();
    if (!config.isAutoRead() && !isReadPending()) {
        // ChannelConfig.setAutoRead(false) was called in the meantime
        removeReadOp();
        return;
    }
    final ChannelPipeline pipeline = pipeline();
    final ByteBufAllocator allocator = config.getAllocator();
    final int maxMessagesPerRead = config.getMaxMessagesPerRead();
    RecvByteBufAllocator.Handle allocHandle = this.allocHandle;
    if (allocHandle == null) {
       this.allocHandle = allocHandle = config.getRecvByteBufAllocator().newHandle();
    }
    ByteBuf byteBuf = null;
    int messages = 0;
    boolean close = false;
    try {
       int totalReadAmount = 0;
       boolean readPendingReset = false;
       do {
          byteBuf = allocHandle.allocate(allocator);
          int writable = byteBuf.writableBytes();
          int localReadAmount = doReadBytes(byteBuf);
          if (localReadAmount <= 0) {
           // not was read release the buffer
              byteBuf.release();
              byteBuf = null;
              close = localReadAmount < 0;
              break;
           }
          if (!readPendingReset) {
               readPendingReset = true;
               setReadPending(false);
          }
          pipeline.fireChannelRead(byteBuf);
          byteBuf = null;

          if (totalReadAmount >= Integer.MAX_VALUE - localReadAmount) {
               totalReadAmount = Integer.MAX_VALUE;
               break;
          }
          totalReadAmount += localReadAmount;

          if (!config.isAutoRead()) {
               break;
          }

          if (localReadAmount < writable) {
              break;
          }
       } while (++ messages < maxMessagesPerRead);
         pipeline.fireChannelReadComplete();
         allocHandle.record(totalReadAmount);

        if (close) {
            closeOnRead(pipeline);
            close = false;
        }
     } catch (Throwable t) {
          handleReadException(pipeline, byteBuf, t, close);
     } finally {
         if (!config.isAutoRead() && !isReadPending()) {
                removeReadOp();
          }
     }
    }
}

    
把这一大段代码分解成几部分
        
1.设置循环读,16次，未读完则会等到下一轮select 继续读取，maxMessagesPerRead默认等于16。
        
2.获取缓存操作handler，config.getRecvByteBufAllocator().newHandle()。
        
3.申请缓存空间，allocHandle.allocate(allocator)。
        
4.从socket中读取数据到byteBuf中。
        
5.传递读事件到下一个handler处理器。
        
6.读完之后发送读完时间到下一个handler处理器
我们只看读事件，其他细节后面的文章再详细解析。

@Override
public ChannelPipeline fireChannelRead(Object msg) {
    head.fireChannelRead(msg);
    return this;
}

@Override
public ChannelHandlerContext fireChannelRead(final Object msg) {
    if (msg == null) {
        throw new NullPointerException("msg");
    }

    final AbstractChannelHandlerContext next = findContextInbound();
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        next.invokeChannelRead(msg);
 } else {
        executor.execute(new OneTimeTask() {
            @Override
            public void run() {
                next.invokeChannelRead(msg);
            }
        });
    }
    return this;
}

image

    
Handler事件顺序是HeadContextHandler --> IdleStateHandler -->IOHandler --> TailContext

private void invokeChannelRead(Object msg) {
    try {
        ((ChannelInboundHandler) handler()).channelRead(this, msg);
    } catch (Throwable t) {
        notifyHandlerException(t);
    }
}

    
进到IdleStateHandler

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    if (readerIdleTimeNanos > 0 || allIdleTimeNanos > 0) {
        reading = true;
        firstReaderIdleEvent = firstAllIdleEvent = true;
    }
    ctx.fireChannelRead(msg);
}

    
设置读事件为true，为后面状态检测做准备，继续向下传递读事件，这次是IOHandler的读事件。

image
public class IOHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        super.channelRead(ctx, msg);
        System.out.println(msg.toString());
    }
    ...
}

    
交给用户自定义handler处理读事件，自此读I/O事件是怎么哪里开始，如何交给用户handler处理已解析完毕。

总结:
    
1.boss线程处理NioServerSocketChannel的accept事件，并将客户端添加到work任务队列，任务队列执行redister0()方法， 将read事件注册到work线程的selector。
    
2.work线程轮询selectkeys，当有事件上来时，将缓存数据发送到用户handler 。

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 214,504评论 6赞 496
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 91,434评论 3赞 389
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 160,089评论 0赞 349
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 57,378评论 1赞 288
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 66,472评论 6赞 386
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 50,506评论 1赞 292
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 39,519评论 3赞 413
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 38,292评论 0赞 270
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 44,738评论 1赞 307
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 37,022评论 2赞 329
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 39,194评论 1赞 342
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 34,873评论 5赞 338
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 40,536评论 3赞 322
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 31,162评论 0赞 21
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,413评论 1赞 268
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 47,075评论 2赞 365
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 44,080评论 2赞 352