Hadoop RPC源码分析(一)

ipc.Server 类分析

Hadoop采用了Master/Slave结构。其中,Master通过ipc.Server接收并处理所有Slave发送的请求,这就要求ipc.Server将高并发和可扩展性作为设计目标。为此,ipc.Server采用了很多具有提高并发处理能力的技术,主要包括线程池、事件驱动Reactor设计模式等。这些技术采用了JDK自带的库实现。我们先重点分析它是如何利用Reactor设计模式提高整体性能的。

01 Reactor设计模式

Reactor是并发编程中的一种基于事件驱动的设计模式。它具有以下2个特点:

  • 通过派发/分离IO操作时间提高系统的并发性能
  • 提供了粗粒度的并发控制,使用单线程实现,避免了复杂的同步处理

一个典型的Reactor模式中主要包括以下几个角色:

  • Reactor:IO事件的派发者
  • Acceptor:接收来自Client的连接,建立与Client对应的Handler,并向Reactor注册此Handler
  • Handler:与一个Client通信的实体,并按一定的过程实现业务的处理。Handler内部往往会有更进一步的层次划分,用来抽象诸如read,decode,compute,encode和send等过程。在Reactor模式中,业务逻辑被分散的IO事件所打破,所以Handler需要有适当的机制在所需的信息还不全(读到一半)的时候保存上下文,并在下一次IO事件到来的时候(另一半可读了)能继续上次中断的处理。
  • Reader/Sender:为了加速处理速度,Reactor模式往往构建一个存放数据处理线程的线程池,这样数据读出后,立即扔到线程池中等待后续处理即可。为此,Reactor模式一般分离Handler中的读和写两个过程,分别注册成单独的读事件和写事件,并由对应的Reader和Sender线程处理。

ipc.Server实现了一个典型的Reactor设计模式,其整体架构与上述完全一致。了解了Reactor的架构之后,能够帮助理解和学习ipc.Server的设计思路及实现。下面就分析Ipc.Server的实现细节。

02 ipc.Server实现细节

用eclipse打开已经编译好的源码,找到ipc.Server,使用eclipse的quick outline查看一下该类的大致结构。


ipc.Server outline

源码内容很多,要先找到下手的地方,通过outline可以捕获到Server有几个内部类,这几个类是什么作用这是需要关心的,然后就是程序的入口,这个start()方法。先看看start()做了什么。

启动服务
Server.start()

  public synchronized void start() {
    responder.start();
    listener.start();
    handlers = new Handler[handlerCount];
    
    for (int i = 0; i < handlerCount; i++) {
      handlers[i] = new Handler(i);
      handlers[i].start();
    }
  }

start()方法启动了几个对象,通过名称可以知道他们是几个内部类的实例,那下一步就应该分析一下每个类的作用。大致查看一下这几个类,发现都继承自Thread类,也就是说每个类都启动了一个新的线程,那么重点就是去考察这个几个线程主体干了什么。

Responder、Listener和Handler
1.Responder

private class Responder extends Thread {
     // 代码.... 此处省略
    @Override
    public void run() {
     // 代码.... 此处省略
      while (running) {
        try {
          writeSelector.select(PURGE_INTERVAL);
          Iterator<SelectionKey> iter = writeSelector.selectedKeys().iterator();
          while (iter.hasNext()) {
            SelectionKey key = iter.next();
            iter.remove();
            try {
              if (key.isValid() && key.isWritable()) {
                  doAsyncWrite(key);
              }
            } catch (IOException e) {
             // 代码.... 此处省略
            }
          }  
        } 
     // 代码.... 此处省略
    }
}

  1. Listener
  private class Listener extends Thread {
     // 代码.... 此处省略
    @Override
    public void run() {
      // 代码.... 此处省略
      while (running) {
        SelectionKey key = null;
        try {
          selector.select();
          Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
          while (iter.hasNext()) {
            key = iter.next();
            iter.remove();
            try {
              if (key.isValid()) {
                if (key.isAcceptable())
                  doAccept(key);
              }
            } catch (IOException e) {
            }
            key = null;
          }
        } catch (OutOfMemoryError e) {
          // 代码.... 此处省略
        } catch (Exception e) {
          // 代码.... 此处省略
        }
        // 代码.... 此处省略
      }
     // 代码.... 此处省略
  }
  1. Handler
  private class Handler extends Thread {
// 代码.... 此处省略
    @Override
    public void run() {
     // 代码.... 此处省略
      while (running) {
        try {
          final Call call = callQueue.take(); // pop the queue; maybe blocked here
           // 代码.... 此处省略
          try { 
            if (call.connection.user == null) {
              value = call(call.connection.protocol, call.param, 
                           call.timestamp);
            } else {
            }
          } catch (Throwable e) {
            // 代码.... 此处省略
          }
          // 代码.... 此处省略
          synchronized (call.connection.responseQueue) {           
            // 代码.... 此处省略
            setupResponse(buf, call, 
                        (error == null) ? Status.SUCCESS : Status.ERROR, 
                        value, errorClass, error);
            responder.doRespond(call);
          }
        } catch (InterruptedException e) {
          // 代码.... 此处省略
        } catch (Exception e) {
          // 代码.... 此处省略
        }
      }
      // 代码.... 此处省略
    }

  }

纵观这几个线程的主体,发现Responder和Listener的代码很熟悉,NIO里的知识。Listener负责监听op_accept事件,然后调用doAccept()方法处理连接;Responder负责监听op_write事件,然后调用doAsyncWrite()方法;Handler里只能大致知道调用了Server.call()这个抽象方法(应该会在某个地方实现)得到了value,然后setupResponse()把处理结果关联到Call,再用responder.doRespond()向客户端做出回应,至于Call,Connection,这也正是我们还没有弄清楚的几个类;还有Call是从一个叫做callQueue的变量里拿到的,这个变量也成为了我们进一步需要关心的地方。

目前能知道的就是:Listener是监听连接的,但对连接是如何处理的还需要解读doRead()方法;Handler是处理业务逻辑的,起点是存放在callQueue中的Call,Call又与Connection联系密切,但这2个类的作用还未知,处理完之后调用responder.doRespond()做出回应,不过Responder功能不仅仅如此,还负责doAsyncWrite()。

所以,接下来的任务是分析一下Call类、Connection类、callQueue变量、doAccept()方法、doRespond()方法和doAsyncWrite()方法。

Call、Connection、callQueue、doAccept()、doRespond()、doAsyncWrite()

  1. Call类
  /** A call queued for handling. */
  private static class Call {
    private int id;                               // the client's call id
    private Writable param;                       // the parameter passed
    private Connection connection;                // connection to client
    private long timestamp;     // the time received when response is null
                                   // the time served when response is not null
    private ByteBuffer response;                      // the response for this call

    public Call(int id, Writable param, Connection connection) { 
      this.id = id;
      this.param = param;
      this.connection = connection;
      this.timestamp = System.currentTimeMillis();
      this.response = null;
    }
    
    @Override
    public String toString() {
      return param.toString() + " from " + connection.toString();
    }

    public void setResponse(ByteBuffer response) {
      this.response = response;
    }
  }
  1. Connection类
  /** Reads calls from a connection and queues them for handling. */
  public class Connection {
      // 代码.... 此处省略
    public Connection(SelectionKey key, SocketChannel channel, 
                      long lastContact) {
      this.channel = channel;
      this.lastContact = lastContact;
      this.data = null;
      this.dataLengthBuffer = ByteBuffer.allocate(4);
      this.unwrappedData = null;
      this.unwrappedDataLengthBuffer = ByteBuffer.allocate(4);
      this.socket = channel.socket();
      this.addr = socket.getInetAddress();
      if (addr == null) {
        this.hostAddress = "*Unknown*";
      } else {
        this.hostAddress = addr.getHostAddress();
      }
      this.remotePort = socket.getPort();
      this.responseQueue = new LinkedList<Call>();
      if (socketSendBufferSize != 0) {
        try {
          socket.setSendBufferSize(socketSendBufferSize);
        } catch (IOException e) {
        }
      }
    } 
  }

Call类的代码比较少,联系RPC的目的,可以分析出这个类是对RPC请求的封装,有传递的参数param,还有连接客户端的Connection,以及处理的结果response。而Connection类的成员变量多,方法也多,所以观察一下构造器,留意到变量responseQueue,应该是用来存放经过handle之后的Call。

  1. callQueue变量
public abstract class Server {
//省略代码
  private BlockingQueue<Call> callQueue;
//省略代码
}

callQueue是一个全局变量,专门用来存放封装请求的Call。call从哪生产,又是被谁消费呢。使用eclipse的Call Hierarchy查看一下调用层次。


callQueue

依次查看可以发现在Connection的processData()方法里面出现了

    private void processData(byte[] buf) throws  IOException, InterruptedException {
     //省略代码
      Call call = new Call(id, param, this);
      callQueue.put(call);   
     //省略代码
    }

Call将一些参数封装,并放入队列callQueue中。这些参数是从字节数组buf里读到的,所以继续往上找:

call

终于找到了我们认识的Listener,点开doRead()方法。

    void doRead(SelectionKey key) throws InterruptedException {
      int count = 0;
      Connection c = (Connection)key.attachment();
      //代码....此处省略
      try {
        count = c.readAndProcess();
      } catch (InterruptedException ieo) {
        //代码....此处省略
      }
       //代码....此处省略
    }

NIO里的SelectionKey对象,doRead()方法中将Connection从SelectionKey中取出,然后通过Connection的readAndRrocess()方法封装call,也就是doRead()中生产了Call,并存放在callQueue中。

    public int readAndProcess() throws IOException, InterruptedException {
        //代码....此处省略
        count = channelRead(channel, data);
        //代码....此处省略
          if (useSasl) {
            saslReadAndProcess(data.array());
          } else {
            processOneRpc(data.array());
          }
}

readAndProcess()是从channel中读取传递过来的字节,然后从里拿到封装Call需要的那些参数,至于具体的细节就不再钻了。

  1. doAccept()
   void doAccept(SelectionKey key) throws IOException,  OutOfMemoryError {
      Connection c = null;
      ServerSocketChannel server = (ServerSocketChannel) key.channel();
      SocketChannel channel;
      while ((channel = server.accept()) != null) {
        channel.configureBlocking(false);
        channel.socket().setTcpNoDelay(tcpNoDelay);
        Reader reader = getReader();
        try {
          reader.startAdd();
          SelectionKey readKey = reader.registerChannel(channel);
          c = new Connection(readKey, channel, System.currentTimeMillis());
          readKey.attach(c);
          synchronized (connectionList) {
            connectionList.add(numConnections, c);
            numConnections++;
          }
         //代码....此处省略       
        } finally {
          reader.finishAdd(); 
        }

      }
    }

doAccept()中生产了Connection并attach到SelectionKey对象中。 这里涉及到一个新的类Reader,我们看看Reader是干什么用的。

    private class Reader implements Runnable {
      private volatile boolean adding = false;
      private Selector readSelector = null;

      Reader(Selector readSelector) {
        this.readSelector = readSelector;
      }
      public void run() {
        LOG.info("Starting SocketReader");
        synchronized (this) {
          while (running) {
            SelectionKey key = null;
            try {
              readSelector.select();
              while (adding) {
                this.wait(1000);
              }              

              Iterator<SelectionKey> iter = readSelector.selectedKeys().iterator();
              while (iter.hasNext()) {
                key = iter.next();
                iter.remove();
                if (key.isValid()) {
                  if (key.isReadable()) {
                    doRead(key);
                  }
                }
                key = null;
              }
            } catch (InterruptedException e) {
              //代码....此处省略
            } catch (IOException ex) {
              //代码....此处省略
            }
          }
        }
      }
    }

Reader继承自Thread,那么就要搞清楚是在哪里启动的线程。调用Call Hierarchy查看,发现是在Listener初始化的时候启动的,代码如下:

    public Listener() throws IOException {
      //代码....此处省略
      for (int i = 0; i < readThreads; i++) {
        Selector readSelector = Selector.open();
        Reader reader = new Reader(readSelector);
        readers[i] = reader;
        readPool.execute(reader);
      }
      //代码....此处省略
    }

Reader的线程体主要是通过doRead()在解析请求,从上面我们知道了doRead()内部是使用Connection.readAndProcess()来解析的。

  1. doRespond()
    void doRespond(Call call) throws IOException {
      synchronized (call.connection.responseQueue) {
        call.connection.responseQueue.addLast(call);
        if (call.connection.responseQueue.size() == 1) {
          processResponse(call.connection.responseQueue, true);
        }
      }
    }

doRespond()调用了processResponse():

    private boolean processResponse(LinkedList<Call> responseQueue,
                                    boolean inHandler) throws IOException {
          //代码....此处省略
      try {
        synchronized (responseQueue) {
          //代码....此处省略
          call = responseQueue.removeFirst();
          SocketChannel channel = call.connection.channel;
          //代码....此处省略
          //
          // Send as much data as we can in the non-blocking fashion
          //
          int numBytes = channelWrite(channel, call.response);
          if (!call.response.hasRemaining()) {
            if (inHandler) {
              //代码....此处省略
              try {
                // Wakeup the thread blocked on select, only then can the call 
                // to channel.register() complete.
                writeSelector.wakeup();
                channel.register(writeSelector, SelectionKey.OP_WRITE, call);
              } catch (ClosedChannelException e) {
                //Its ok. channel might be closed else where.
                done = true;
              } finally {
                 //代码....此处省略
              }
            }
      } finally {
        //代码....此处省略
        }
      }
      return done;
    }

channelWrite(channel, call.response)把处理的结果返回给客户端,“Send as much data as we can in the non-blocking fashion”,如果有剩余的data就会注册写事件:
channel.register(writeSelector, SelectionKey.OP_WRITE, call),也就会调用doAysnWrite()去处理剩下的数据。

到这里大概的原理就清楚了,画一个不科学的示意图(但比较直观哈),如下:

Server

总结一下就是:

(1)接收请求
该阶段主要任务是接收来自各个客户端的RPC请求,并将它们封装成固定的格式(Call类)放到一个共享队列(callQueue)中,以便进行后续处理。该阶段内部又分为建立连接和接收请求两个子阶段,分别由Listener和Reader两种线程完成。整个Server只有一个Listener线程,统一负责监听来自客户端的连接请求,一旦有新的请求到达,它会采用轮询的方式从线程池中选择一个Reader线程进行处理,而Reader线程可同时存在多个,它们分别负责接收一部分客户端连接的RPC请求,至于每个Reader线程负责哪些客户端连接,完全由Listener决定,当前Listener只是采用了简单的轮询分配机制。

(2)处理请求
该阶段主要任务是从共享队列callQueue中获取Call对象,执行对应的函数调用,并将结果返回给客户端,这全部由Handler线程完成。

Server端可同时存在多个Handler线程,它们并行从共享队列中读取Call对象,经执行对应的函数调用后,将尝试着直接将结果返回给对应的客户端。但考虑到某些函数调用返回结果很大或者网络速度过慢,可能难以将结果一次性发送到客户端,此时Handler将尝试着将后续发送任务交给Responder线程。

(3)返回结果
前面提到,每个Handler线程执行完函数调用后,会尝试着将执行结果返回给客户端,但对于特殊情况,比如函数调用返回结果过大或者网络异常情况(网速过慢),会将发送任务交给Responder线程。

Server端仅存在一个Responder线程,它的内部包含一个Selector对象,用于监听SelectionKey.OP_WRITE事件。当Handler没能将结果一次性发送到客户端时,会向该Selector对象注册SelectionKey.OP_WRITE事件,进而由Responder线程采用异步方式继续发送未发送完成的结果。

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

推荐阅读更多精彩内容

  • 2018-02-04 意志力训练营心得体会 1.本周意志力相比较前两周明显不稳定,原因主要是考试报名扰乱了心绪,心...
    cj2018阅读 163评论 0 0
  • 如果让你选择未来。 1,快乐的。 2,凄凉的。 3,幸福的。 4,悲惨的。 选择快乐的你,现在就不要再迷茫了,随你...
    单个儿酒窝阅读 143评论 0 0
  • 谢尔顿对艾米表白了,那个被人们称为怪胎、天才,脑回路似乎永远不跟正常人在一个频道的人,发誓要为科学献身,永远以自...
    绒团阅读 252评论 0 0
  • 工作: (๑• . •๑)唉?我还木有工作...暑假可能找一份暑假工作,减轻爸妈负担啦! 学习: 希望坚持住自己...
    李腿腿阅读 143评论 0 0