通过Hadoop RPC框架学习NIO

今儿个我们通过HDFS NameNode学习Java NIO框架。读本文之前假定各位大佬有NIO的基础(Selector、Channel、Buffer、以及NIO编程的模板框架)。

Hadoop RPC的Server类用到了NIO、Reactor模式等技术来尽可能的提高服务器端的并发度。可以说,学习Server类的源码是学习NIO非常好的实例,好,话不多说,让我们看看NIO是如何在HDFS这种广泛使用的大数据组件上使用的。

首先说一些前置知识:Hadoop RPC框架的服务端代码实现是在Server类。Server类下面有很多内部类。如下图所示:

Server#Listener

Listener继承了Thread类,所以它本质上也是个线程类。
Listener类主要是用来处理客户端的Socket连接,类似于Reactor模式中的Reactor角色。

Listener的成员变量如下图所示:

Listener的构造函数中进行这些成员变量的初始化,是NIO编程框架的固定写法:

接着在Listener的run方法中调用select()方法,阻塞的做select操作,只有当至少一个channel被selected,或者当前线程被interrupted,或者此selector的wakeup方法被调用时才会返回。select()最终会选择一些key的结合,这些key对应的channel是I/O就绪的。

当Listener这侧判断key是OP_ACCEPT,就知道是客户端要与服务器进行连接了,于是调用doAccept方法,并把key作为参数传入(便于后续调用attach方法分配handler)。

这里我们接着看doAccept方法,先用文字分析一下:
通过ServerSocketChannel的accept方法,拿到代表客户端Socket的SocketChannel,将客户端的SocketChannel注册到connectionManager中,ConnectionManager类用于连接的管理,包括最长空闲时间,最多连接数,Connetion对象集合等等。然后将包含客户端SocketChannel的Connection对象attach到SelectionKey上,方便后面从SelectionKey上调用attachment方法得到这个包含客户端SocketChannel的Connection对象。最后把连接添加到Reader的pendingConnections这个阻塞队列中,同时让Reader类的Selector对象readSelector立即返回,readSelector在进行下一次select之前,会把阻塞队列pendingConnections里的Connection处理完。代码如下:

    void doAccept(SelectionKey key) throws InterruptedException, IOException,  OutOfMemoryError {
      ServerSocketChannel server = (ServerSocketChannel) key.channel();
      SocketChannel channel;
      // 获取到代表客户端Socket的channel
      while ((channel = server.accept()) != null) {
        // NIO常用配置
        channel.configureBlocking(false);
        channel.socket().setTcpNoDelay(tcpNoDelay);
        channel.socket().setKeepAlive(true);
        // Reader相当于Reactor模式中的handler,用来从channel中读取客户端Socket发送过来的数据
        Reader reader = getReader();
        // 创建一个Connection对象,在register里把这个对象添加到ConnectionManager#connections集合里
        Connection c = connectionManager.register(channel);
        // If the connectionManager can't take it, close the connection.
        // 如果超过了设置的最大连接数c == null
        if (c == null) {
          if (channel.isOpen()) {
            IOUtils.cleanup(null, channel);
          }
          connectionManager.droppedConnections.getAndIncrement();
          continue;
        }
        // 将Connection对象attach到此key上,这样后续通过key.attachment()即可拿到c这个对象
        key.attach(c);  // so closeCurrentConnection can get the object
        // 把连接添加到Reader的Connection阻塞队列中
        reader.addConnection(c);
      }
    }

接着我们来看Reader的run方法,前面我们看到了,在Listener类构造方法执行的时候,会初始化它的Reader[]成员变量readers,然后启动Reader线程,如下图:

所以接下来我们要看Reader的run方法:
很简单调用了doRunLoop方法,直接去看doRunLoop方法:

看下doRead方法,此方法中有两个地方需要深入研究:
① 通过key.attachment()拿到之前attach在此key的object,这个object是什么?又是什么时候attach到key上的的?
② readAndProcess函数的处理逻辑,看名字此函数就是读取客户端RPC请求然后进行处理的地方。

    void doRead(SelectionKey key) throws InterruptedException {
      int count;
      // 从SelectionKey中拿出之前附着的Connection对象,一会我们再讲什么时候附着的
      Connection c = (Connection)key.attachment();
      if (c == null) {
        return;  
      }
      // 设置最近一次连接时间
      c.setLastContact(Time.now());
      
      try {
        // 读客户端的RPC请求数据,并处理。这个函数后面分析
        count = c.readAndProcess();
      } catch (InterruptedException ieo) {
        LOG.info(Thread.currentThread().getName() + ": readAndProcess caught InterruptedException", ieo);
        throw ieo;
      } catch (Exception e) {
        // Any exceptions that reach here are fatal unexpected internal errors
        // that could not be sent to the client.
        LOG.info(Thread.currentThread().getName() +
            ": readAndProcess from client " + c +
            " threw exception [" + e + "]", e);
        count = -1; //so that the (count < 0) block is executed
      }
      // setupResponse will signal the connection should be closed when a
      // fatal response is sent.
      if (count < 0 || c.shouldClose()) {
        closeConnection(c);
        c = null;
      }
      else {
        c.setLastContact(Time.now());
      }
    }

首先来回答第一个问题:通过key.attachment()拿到之前attach在此key的object,这个object是什么?又是什么时候attach到key上的的?

请看doRead方法前面的register方法,如下图所示:

看下register方法:
register(Selector sel, int ops,Object att)
这个方法是NIO库的方法。用来将channel注册到给定的selector上,并设置监听事件类型为ops,同时在key上attach一个对象att。

接着看第二个问题:readAndProcess的逻辑:
文字描述一下:
主要是通过channel的read方法将流中数据读到ByteBuffer类型的成员变量中,ByteBuffer成员变量是在Connection类中,Connection类有几个不同作用的ByteBuffer,分别是data、dataLengthBuffer、connectionHeaderBuf、unwrappedData、unwrappedDataLengthBuffer。 根据Hadoop的协议头,读取相应位置的数据到对应的ByteBuffer中,有关请求体的数据被读到data这个ByteBuffer中,然后调用processOneRpc处理这个请求,processOneRpc方法不是本文关注的重点,但是可以稍微描述一下他的作用:将RPC请求封装成Call,加入到Call Queue中,等待Server#Handler内部类去消费。

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 220,884评论 6 513
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 94,212评论 3 395
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 167,351评论 0 360
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 59,412评论 1 294
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 68,438评论 6 397
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 52,127评论 1 308
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 40,714评论 3 420
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 39,636评论 0 276
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 46,173评论 1 319
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 38,264评论 3 339
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 40,402评论 1 352
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 36,073评论 5 347
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 41,763评论 3 332
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 32,253评论 0 23
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 33,382评论 1 271
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 48,749评论 3 375
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 45,403评论 2 358

推荐阅读更多精彩内容