Netty源码分析第3章(客户端接入流程)---->第4节: NioSocketChannel注册到selector

Netty源码分析第三章: 客户端接入流程

第四节: NioSocketChannel注册到selector

我们回到最初的NioMessageUnsafe的read()方法:

public void read() {
    //必须是NioEventLoop方法调用的, 不能通过外部线程调用
    assert eventLoop().inEventLoop();
    //服务端channel的config
    final ChannelConfig config = config();
    //服务端channel的pipeline
    final ChannelPipeline pipeline = pipeline();
    //处理服务端接入的速率
    final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
    //设置配置
    allocHandle.reset(config);
    boolean closed = false;
    Throwable exception = null;
    try {
        try {
            do {
                //创建jdk底层的channel
                //readBuf用于临时承载读到链接
                int localRead = doReadMessages(readBuf);
                if (localRead == 0) {
                    break;
                }
                if (localRead < 0) {
                    closed = true;
                    break;
                }
                //分配器将读到的链接进行计数
                allocHandle.incMessagesRead(localRead);
                //连接数是否超过最大值
            } while (allocHandle.continueReading());
        } catch (Throwable t) {
            exception = t;
        }
        int size = readBuf.size();
        //遍历每一条客户端连接
        for (int i = 0; i < size; i ++) {
            readPending = false;
            //传递事件, 将创建NioSokectChannel进行传递
            //最终会调用ServerBootstrap的内部类ServerBootstrapAcceptor的channelRead()方法
            pipeline.fireChannelRead(readBuf.get(i));
        }
        readBuf.clear();
        allocHandle.readComplete();
        pipeline.fireChannelReadComplete();
        //代码省略
    } finally {
        //代码省略
    }
}

在while循环结束之后, 将会通过一个for循环遍历readBuf集合, 并将创建的NioSocketChannel传入fireChannelRead()中, 传播channel的读取事件

有关pipeline的知识, 我们下一章会详细剖析, 并会根据剖析后的内容回顾之前的有关pipeline的操作, 这里我们只需知道, 通过fireChannelRead()我们最终调用了ServerBootstrap的内部类ServerBootstrapAcceptor 中的channelRead()方法

跟到channelRead()方法中:

public void channelRead(ChannelHandlerContext ctx, Object msg) {
    final Channel child = (Channel) msg;
    //代码省略
    try {
        //work线程注册channel
        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);
    }
}

其中参数的msg就是最初传入fireChannelRead()方法的NioSocketChannel

所以这里可以通过 final Channel child = (Channel) msg 这种方式拿到NioSocketChannel

其中childGroup是我们最初初始化的work线程, 这里的register()方法跟boss线程一样, 通过next()方法获选择一个线程进行注册, 这里不再赘述

我们紧跟调用链, 跟到SingleThreadEventLoop的register()方法:

public ChannelFuture register(final ChannelPromise promise) {
    ObjectUtil.checkNotNull(promise, "promise");
    promise.channel().unsafe().register(this, promise);
    return promise;
}

这里的unsafe(), 根据我们之前的剖析, 是NioByteUnsafe, 这里的register最终会调用AbstractUnsafe的register()方法, 并NioSocketChannel

不知道同学们是否记得, 当初NioServerSocketChannel注册的时候也走的这个方法

我们跟到register()这个方法中:

public final void register(EventLoop eventLoop, final ChannelPromise promise) {
    //省略验证代码
    //所有的复制操作, 都交给eventLoop处理
    AbstractChannel.this.eventLoop = eventLoop;

    if (eventLoop.inEventLoop()) {
        //做实际主注册
        register0(promise);
    } else {
        try {
            eventLoop.execute(new Runnable() {
                @Override
                public void run() {
                    register0(promise);
                }
            });
        } catch (Throwable t) {
            //代码省略
        }
    }
}

我们学习过NioEventLoop相关知识之后, 应该对这部分代码不太陌生, 首先判断是不是当前NioEventLoop线程, 如果是, 则直接进行注册操作, 如果不是, 则封装成task在当前NioEventLoop中执行

走到这里不难明白, 这里并不是当前NioEventLoop线程, 这是boss线程执行的, 所以这里会走到else, 如果是第一次的连接操作, work线程的NioEventLoop并没有启动, 所以这里也会启动NioEventLoop, 并开始轮询操作

跟到register0(promise)中看其是如何做实际操作的:

private void register0(ChannelPromise promise) {
    try {
        //省略代码
        //做实际的注册
        doRegister();
        neverRegistered = false;
        registered = true;
        //触发事件
        pipeline.invokeHandlerAddedIfNeeded();
        safeSetSuccess(promise);
        //触发注册成功事件
        pipeline.fireChannelRegistered();
        if (isActive()) {
            if (firstRegistration) {
                //传播active事件(4)
                pipeline.fireChannelActive();
            } else if (config().isAutoRead()) {
                beginRead();
            }
        }
    } catch (Throwable t) {
        //省略代码
    }
}

这段代码我们同样并不陌生, 因为NioServerSokectChannel中也走这一部分, 我们继续关注doRegister()方法:

protected void doRegister() throws Exception {
    boolean selected = false;
    for (;;) {
        try {
            //jdk底层的注册方法
            //第一个参数为selector, 第二个参数表示不关心任何事件
            selectionKey = javaChannel().register(eventLoop().selector, 0, this);
            return;
        } catch (CancelledKeyException e) {
            //省略代码
        }
    }
}

这部分也是我们之前剖析过的jdk底层的注册, 只是不同的是, 这里的javaChannel()是SocketChanel而不是ServerSocketChannel

同样, 这里也是表示不关心任何事件, 只是在当前NioEventLoop绑定的selector上注册

至此, NioSocketChannel完成注册

原文地址:https://www.cnblogs.com/xiangnan6122/p/10204023.html

时间: 12-31

Netty源码分析第3章(客户端接入流程)---->第4节: NioSocketChannel注册到selector的相关文章

netty 源码分析二

以服务端启动,接收客户端连接整个过程为例分析, 简略分为 五个过程: 1.NioServerSocketChannel 管道生成, 2.NioServerSocketChannel 管道完成初始化, 3.NioServerSocketChannel注册至Selector选择器, 4.NioServerSocketChannel管道绑定到指定端口,启动服务 5.NioServerSocketChannel接受客户端的连接,进行相应IO操作 Ps:netty内部过程远比这复杂,简略记录下方便以后回忆

Solr4.8.0源码分析(25)之SolrCloud的Split流程

Solr4.8.0源码分析(25)之SolrCloud的Split流程(一) 题记:昨天有位网友问我SolrCloud的split的机制是如何的,这个还真不知道,所以今天抽空去看了Split的原理,大致也了解split的原理了,所以也就有了这篇文章.本系列有两篇文章,第一篇为core split,第二篇为collection split. 1. 简介 这里首先需要介绍一个比较容易混淆的概念,其实Solr的HTTP API 和 SolrCloud的HTTP API是不一样,如果接受到的是Solr的

Netty源码分析之客户端启动过程

一.先来看一下客户端示例代码. 1 public class NettyClientTest { 2 public void connect(int port, String host) throws Exception { 3 EventLoopGroup group = new NioEventLoopGroup();//与服务端不同,客户端只需要一个IO线程组 4 5 try { 6 Bootstrap b = new Bootstrap(); 7 b.group(group) 8 .op

[转载] Netty源码分析

转载自http://blog.csdn.net/kobejayandy/article/details/11836813 Netty提供异步的.事件驱动的网络应用程序框架和工具,用以快速开发高性能.高可靠性的网络服务器和客户端程序[官方定义],整体来看其包含了以下内容: 1.提供了丰富的协议编解码支持 2.实现自有的buffer系统,减少复制所带来的消耗 3.整套channel 的实现 4.基于事件的过程流转以及完整的网络事件响应与扩展 5.丰富的example. 本文并不对Netty实际使用中

netty源码分析之一:server的启动

nio server启动的第一步,都是要创建一个serverSocketChannel,我截取一段启动代码,一步步分析: public void afterPropertiesSet() throws Exception { // 创建rpc工厂 ThreadFactory threadRpcFactory = new NamedThreadFactory("NettyRPC ThreadFactory"); //执行worker线程池数量 int parallel = Runtime

ZooKeeper源码分析:Quorum请求的整个流程(转)

Quorum请求是转发给Leader处理,并且需要得一个Follower Quorum确认的请求.这些请求包括: 1)znode的写操作(OpCode.create,OpCode.delete,OpCode.setData,OpCode.setACL) 2)Session的创建和关闭操作(OpCode.createSession和OpCode.closeSession) 3)OpCode.multi操作. 本文分析了Client, Follower和Leader协同完成Quorum请求的过程.另

angular源码分析:图解angular的启动流程

今天做了一些图来说明angular,由于angular实在太复杂了,不知道用什么图表示比较好,所以就胡乱画了一些,希望有人能看得懂. 一.源码文件编译合并顺序图 二.angular.module函数功能图 三.publishExternalAPI函数功能图 四.注入器工厂函数createInjector内部的providerCache和instanceCache

Android7.0 Phone应用源码分析(一) phone拨号流程分析

1.1 dialer拨号 拨号盘点击拨号DialpadFragment的onClick方法会被调用 public void onClick(View view) { int resId = view.getId(); if (resId == R.id.dialpad_floating_action_button) { view.performHapticFeedback(HapticFeedbackConstants.VIRTUAL_KEY); handleDialButtonPressed(

netty 5 alph1源码分析(服务端创建过程)

参照<Netty系列之Netty 服务端创建>,研究了netty的服务端创建过程.至于netty的优势,可以参照网络其他文章.<Netty系列之Netty 服务端创建>是 李林锋撰写的netty源码分析的一篇好文,绝对是技术干货.但抛开技术来说,也存在一些瑕疵. 缺点如下 代码衔接不连贯,上下不连贯. 代码片段是截图,对阅读代理不便(可能和阅读习惯有关) 本篇主要内容,参照<Netty系列之Netty 服务端创建>,梳理出自己喜欢的阅读风格. 1.整体逻辑图 整体将服务

【React源码分析】组件通信、refs、key和ReactDOM

React源码系列文章,请多支持:React源码分析1 - 组件和对象的创建(createClass,createElement)React源码分析2 - React组件插入DOM流程React源码分析3 - React生命周期详解React源码分析4 - setState机制React源码分析5 -- 组件通信,refs,key,ReactDOMReact源码分析6 - React合成事件系统 1 组件间通信 父组件向子组件通信 React规定了明确的单向数据流,利用props将数据从父组件传