`

源码级强力分析hadoop的RPC机制

 
阅读更多

转:http://weixiaolu.iteye.com/blog/1504898

分析对象: 
hadoop版本:hadoop 0.20.203.0

必备技术点: 
1. 动态代理(参考
 :http://weixiaolu.iteye.com/blog/1477774 
2. Java NIO(参考
 :http://weixiaolu.iteye.com/blog/1479656 
3. Java网络编程

目录: 
一.RPC协议
二.ipc.RPC源码分析
三.ipc.Client源码分析
四.ipc.Server源码分析
 

分析: 

一.RPC协议 

在分析协议之前,我觉得我们很有必要先搞清楚协议是什么。下面我就谈一点自己的认识吧。如果你学过java的网络编程,你一定知道:当客户端发送一个字节给服务端时,服务端必须也要有一个读字节的方法在阻塞等待;反之亦然。 这种我把它称为底层的通信协议。可是对于一个大型的网络通信系统来说,很显然这种说法的协议粒度太小,不方便我们理解整个网络通信的流程及架构,所以我造了个说法:架构层次的协议。通俗一点说,就是我把某些接口和接口中的方法称为协议,客户端和服务端只要实现这些接口中的方法就可以进行通信了,从这个角度来说,架构层次协议的说法就可以成立了(注:如果从架构层次的协议来分析系统,我们就先不要太在意方法的具体实现,呵呵,我相信你懂得~)。

Hadoop的RPC机制正是采用了这种“架构层次的协议”,有一整套作为协议的接口。如图:

 

 

下面就几个重点的协议介绍一下吧:

 

VersionedProtocol :它是所有RPC协议接口的父接口,其中只有一个方法:getProtocolVersion()

(1)HDFS相关 
ClientDatanodeProtocol :一个客户端和datanode之间的协议接口,用于数据块恢复
ClientProtocol :client与Namenode交互的接口,所有控制流的请求均在这里,如:创建文件、删除文件等;
DatanodeProtocol : Datanode与Namenode交互的接口,如心跳、blockreport等;
NamenodeProtocol :SecondaryNode与Namenode交互的接口。

(2)Mapreduce相关 
InterDatanodeProtocol :Datanode内部交互的接口,用来更新block的元数据;
InnerTrackerProtocol :TaskTracker与JobTracker交互的接口,功能与DatanodeProtocol相似;
JobSubmissionProtocol :JobClient与JobTracker交互的接口,用来提交Job、获得Job等与Job相关的操作;
TaskUmbilicalProtocol :Task中子进程与母进程交互的接口,子进程即map、reduce等操作,母进程即TaskTracker,该接口可以回报子进程的运行状态(词汇扫盲: umbilical 脐带的, 关系亲密的) 。

 

一下子罗列了这么多的协议,有些人可能要问了,hadoop是怎么使用它们的呢?呵呵,不要着急哦,其实本篇博客所分析的是hadoop的RPC机制底层的具体实现,而这些协议却是应用层上的东西,比如hadoop是怎么样保持“心跳”的啊。所以在我的下一篇博客:源码级分析hadoop的心跳机制中会详细说明以上协议是怎样被使用的。尽请期待哦~。现在就开始我们的RPC源码之旅吧•••

二.ipc.RPC源码分析 

ipc.RPC类中有一些内部类,为了大家对RPC类有个初步的印象,就先罗列几个我们感兴趣的分析一下吧:

 

Invocation :用于封装方法名和参数,作为数据传输层,相当于VO吧。
ClientCache :用于存储client对象,用socket factory作为hash key,存储结构为hashMap <SocketFactory, Client>。
Invoker :是动态代理中的调用实现类,继承了InvocationHandler.
Server :是ipc.Server的实现类。

 

从以上的分析可以知道,Invocation类仅作为VO,ClientCache类只是作为缓存,而Server类用于服务端的处理,他们都和客户端的数据流和业务逻辑没有关系。现在就只剩下Invoker类了。如果你对动态代理(参考:http://weixiaolu.iteye.com/blog/1477774 )比较了解的话,你一下就会想到,我们接下来去研究的就是RPC.Invoker类中的invoke()方法了。代码如下:

代码一:

Java代码   收藏代码
  1. public Object invoke(Object proxy, Method method, Object[] args)  
  2.   throws Throwable {  
  3.   •••  
  4.   ObjectWritable value = (ObjectWritable)  
  5.     client.call(new Invocation(method, args), remoteId);  
  6.   •••  
  7.   return value.get();  
  8. }  

 

呵呵,如果你发现这个invoke()方法实现的有些奇怪的话,那你就对了。一般我们看到的动态代理的invoke()方法中总会有 method.invoke(ac, arg);  这句代码。而上面代码中却没有,这是为什么呢?其实使用 method.invoke(ac, arg); 是在本地JVM中调用;而在hadoop中,是将数据发送给服务端,服务端将处理的结果再返回给客户端,所以这里的invoke()方法必然需要进行网络通信。而网络通信就是下面的这段代码实现的:

代码二:

Java代码   收藏代码
  1. ObjectWritable value = (ObjectWritable)  
  2. client.call(new Invocation(method, args), remoteId);  

 

Invocation类在这里封装了方法名和参数,充当VO。其实这里网络通信只是调用了Client类的call()方法。那我们接下来分析一下ipc.Client源码吧。不过在分析ipc.Client源码之前,为了不让我们像盲目的苍蝇一样乱撞,我想先确定一下我们分析的目的是什么,我总结出了三点需要解决的问题:


1. 客户端和服务端的连接是怎样建立的?
2. 客户端是怎样给服务端发送数据的?
3. 客户端是怎样获取服务端的返回数据的?


基于以上三个问题,我们开始吧!!!

三.ipc.Client源码分析 

同样,为了对Client类有个初步的了解,我们也先罗列几个我们感兴趣的内部类:

 

Call :用于封装Invocation对象,作为VO,写到服务端,同时也用于存储从服务端返回的数据
Connection :用以处理远程连接对象。继承了Thread
ConnectionId :唯一确定一个连接

 

问题1:客户端和服务端的连接是怎样建立的? 

下面我们来看看Client类中的cal()方法吧:

代码三:

Java代码   收藏代码
  1. public Writable call(Writable param, ConnectionId remoteId)    
  2.                        throws InterruptedException, IOException {  
  3.     Call call = new Call(param);       //将传入的数据封装成call对象  
  4.     Connection connection = getConnection(remoteId, call);   //获得一个连接  
  5.     connection.sendParam(call);     // 向服务端发送call对象  
  6.     boolean interrupted = false;  
  7.     synchronized (call) {  
  8.       while (!call.done) {  
  9.         try {  
  10.           call.wait(); // 等待结果的返回,在Call类的callComplete()方法里有notify()方法用于唤醒线程  
  11.         } catch (InterruptedException ie) {  
  12.           // 因中断异常而终止,设置标志interrupted为true  
  13.           interrupted = true;  
  14.         }  
  15.       }  
  16.       if (interrupted) {  
  17.         Thread.currentThread().interrupt();  
  18.       }  
  19.   
  20.       if (call.error != null) {  
  21.         if (call.error instanceof RemoteException) {  
  22.           call.error.fillInStackTrace();  
  23.           throw call.error;  
  24.         } else { // 本地异常  
  25.           throw wrapException(remoteId.getAddress(), call.error);  
  26.         }  
  27.       } else {  
  28.         return call.value; //返回结果数据  
  29.       }  
  30.     }  
  31.   }  

 

具体代码的作用我已做了注释,所以这里不再赘述。但到目前为止,你依然不知道RPC机制底层的网络连接是怎么建立的。呵呵,那我们只好再去深究了,分析代码后,我们会发现和网络通信有关的代码只会是下面的两句了:

代码四:

Java代码   收藏代码
  1. Connection connection = getConnection(remoteId, call);   //获得一个连接  
  2. connection.sendParam(call);      // 向服务端发送call对象  

 

先看看是怎么获得一个到服务端的连接吧,下面贴出ipc.Client类中的getConnection()方法。

代码五:

Java代码   收藏代码
  1. private Connection getConnection(ConnectionId remoteId,  
  2.                                    Call call)  
  3.                                    throws IOException, InterruptedException {  
  4.     if (!running.get()) {  
  5.       // 如果client关闭了  
  6.       throw new IOException("The client is stopped");  
  7.     }  
  8.     Connection connection;  
  9. //如果connections连接池中有对应的连接对象,就不需重新创建了;如果没有就需重新创建一个连接对象。  
  10. //但请注意,该//连接对象只是存储了remoteId的信息,其实还并没有和服务端建立连接。  
  11.     do {  
  12.       synchronized (connections) {  
  13.         connection = connections.get(remoteId);  
  14.         if (connection == null) {  
  15.           connection = new Connection(remoteId);  
  16.           connections.put(remoteId, connection);  
  17.         }  
  18.       }  
  19.     } while (!connection.addCall(call)); //将call对象放入对应连接中的calls池,就不贴出源码了  
  20.    //这句代码才是真正的完成了和服务端建立连接哦~  
  21.     connection.setupIOstreams();  
  22.     return connection;  
  23.   }  

 

如果你还有兴趣继续分析下去,那我们就一探建立连接的过程吧,下面贴出Client.Connection类中的setupIOstreams()方法:

代码六:

Java代码   收藏代码
  1. private synchronized void setupIOstreams() throws InterruptedException {  
  2.  •••  
  3.     try {  
  4.      •••  
  5.       while (true) {  
  6.         setupConnection();  //建立连接  
  7.         InputStream inStream = NetUtils.getInputStream(socket);     //获得输入流  
  8.         OutputStream outStream = NetUtils.getOutputStream(socket);  //获得输出流  
  9.         writeRpcHeader(outStream);  
  10.         •••  
  11.         this.in = new DataInputStream(new BufferedInputStream  
  12.             (new PingInputStream(inStream)));   //将输入流装饰成DataInputStream  
  13.         this.out = new DataOutputStream  
  14.         (new BufferedOutputStream(outStream));   //将输出流装饰成DataOutputStream  
  15.         writeHeader();  
  16.         // 跟新活动时间  
  17.         touch();  
  18.         //当连接建立时,启动接受线程等待服务端传回数据,注意:Connection继承了Tread  
  19.         start();  
  20.         return;  
  21.       }  
  22.     } catch (IOException e) {  
  23.       markClosed(e);  
  24.       close();  
  25.     }  
  26.   }  

 

再有一步我们就知道客户端的连接是怎么建立的啦,下面贴出Client.Connection类中的setupConnection()方法:

代码七:

Java代码   收藏代码
  1. private synchronized void setupConnection() throws IOException {  
  2.     short ioFailures = 0;  
  3.     short timeoutFailures = 0;  
  4.     while (true) {  
  5.       try {  
  6.         this.socket = socketFactory.createSocket(); //终于看到创建socket的方法了  
  7.         this.socket.setTcpNoDelay(tcpNoDelay);  
  8.        •••  
  9.         // 设置连接超时为20s  
  10.         NetUtils.connect(this.socket, remoteId.getAddress(), 20000);  
  11.         this.socket.setSoTimeout(pingInterval);  
  12.         return;  
  13.       } catch (SocketTimeoutException toe) {  
  14.         /* 设置最多连接重试为45次。 
  15.          * 总共有20s*45 = 15 分钟的重试时间。 
  16.          */  
  17.         handleConnectionFailure(timeoutFailures++, 45, toe);  
  18.       } catch (IOException ie) {  
  19.         handleConnectionFailure(ioFailures++, maxRetries, ie);  
  20.       }  
  21.     }  
  22.   }  

 

终于,我们知道了客户端的连接是怎样建立的了,其实就是创建一个普通的socket进行通信。呵呵,那服务端是不是也是创建一个ServerSocket进行通信的呢?呵呵,先不要急,到这里我们只解决了客户端的第一个问题,下面还有两个问题没有解决呢,我们一个一个地来解决吧。

问题2:客户端是怎样给服务端发送数据的? 

我们回顾一下代码四吧。第一句为了完成连接的建立,我们已经分析完毕;而第二句是为了发送数据,呵呵,分析下去,看能不能解决我们的问题呢。下面贴出Client.Connection类的sendParam()方法吧:

代码八:

Java代码   收藏代码
  1. public void sendParam(Call call) {  
  2.       if (shouldCloseConnection.get()) {  
  3.         return;  
  4.       }  
  5.       DataOutputBuffer d=null;  
  6.       try {  
  7.         synchronized (this.out) {  
  8.           if (LOG.isDebugEnabled())  
  9.             LOG.debug(getName() + " sending #" + call.id);  
  10.           //创建一个缓冲区  
  11.           d = new DataOutputBuffer();  
  12.           d.writeInt(call.id);  
  13.           call.param.write(d);  
  14.           byte[] data = d.getData();  
  15.           int dataLength = d.getLength();  
  16.           out.writeInt(dataLength);        //首先写出数据的长度  
  17.           out.write(data, 0, dataLength); //向服务端写数据  
  18.           out.flush();  
  19.         }  
  20.       } catch(IOException e) {  
  21.         markClosed(e);  
  22.       } finally {  
  23.         IOUtils.closeStream(d);  
  24.       }  
  25.     }    

 

其实这就是java io的socket发送数据的一般过程哦,没有什么特别之处。到这里问题二也解决了,来看看问题三吧。

问题3:客户端是怎样获取服务端的返回数据的? 

我们再回顾一下代码六吧。代码六中,当连接建立时会启动一个线程用于处理服务端返回的数据,我们看看这个处理线程是怎么实现的吧,下面贴出Client.Connection类和Client.Call类中的相关方法吧:

代码九:

Java代码   收藏代码
  1. 方法一:    
  2.   public void run() {  
  3.       •••  
  4.       while (waitForWork()) {  
  5.         receiveResponse();  //具体的处理方法  
  6.       }  
  7.       close();  
  8.      •••  
  9. }  
  10.   
  11. 方法二:  
  12. private void receiveResponse() {  
  13.       if (shouldCloseConnection.get()) {  
  14.         return;  
  15.       }  
  16.       touch();  
  17.       try {  
  18.         int id = in.readInt();                    // 阻塞读取id  
  19.         if (LOG.isDebugEnabled())  
  20.           LOG.debug(getName() + " got value #" + id);  
  21.           Call call = calls.get(id);    //在calls池中找到发送时的那个对象  
  22.         int state = in.readInt();     // 阻塞读取call对象的状态  
  23.         if (state == Status.SUCCESS.state) {  
  24.           Writable value = ReflectionUtils.newInstance(valueClass, conf);  
  25.           value.readFields(in);           // 读取数据  
  26.         //将读取到的值赋给call对象,同时唤醒Client等待线程,贴出setValue()代码方法三  
  27.           call.setValue(value);                
  28.           calls.remove(id);               //删除已处理的call      
  29.         } else if (state == Status.ERROR.state) {  
  30.         •••  
  31.         } else if (state == Status.FATAL.state) {  
  32.         •••  
  33.         }  
  34.       } catch (IOException e) {  
  35.         markClosed(e);  
  36.       }  
  37. }  
  38.   
  39. 方法三:  
  40. public synchronized void setValue(Writable value) {  
  41.       this.value = value;  
  42.       callComplete();   //具体实现  
  43. }  
  44. protected synchronized void callComplete() {  
  45.       this.done = true;  
  46.       notify();         // 唤醒client等待线程  
  47.     }  

 

代码九完成的功能主要是:启动一个处理线程,读取从服务端传来的call对象,将call对象读取完毕后,唤醒client处理线程。就这么简单,客户端就获取了服务端返回的数据了哦~。客户端的源码分析就到这里了哦,下面我们来分析Server端的源码吧。

四.ipc.Server源码分析 

同样,为了让大家对ipc.Server有个初步的了解,我们先分析一下它的几个内部类吧:

 

Call :用于存储客户端发来的请求
Listener : 监听类,用于监听客户端发来的请求,同时Listener内部还有一个静态类,Listener.Reader,当监听器监听到用户请求,便让Reader读取用户请求。
Responder :响应RPC请求类,请求处理完毕,由Responder发送给请求客户端。
Connection :连接类,真正的客户端请求读取逻辑在这个类中。
Handler :请求处理类,会循环阻塞读取callQueue中的call对象,并对其进行操作。

 

如果你看过ipc.Server的源码,你会发现其实ipc.Server是一个abstract修饰的抽象类。那随之而来的问题就是:hadoop是怎样初始化RPC的Server端的呢?这个问题着实也让我想了好长时间。不过后来我想到Namenode初始化时一定初始化了RPC的Sever端,那我们去看看Namenode的初始化源码吧:

1. 初始化Server


代码十:

Java代码   收藏代码
  1. private void initialize(Configuration conf) throws IOException {  
  2.    •••  
  3.     // 创建 rpc server  
  4.     InetSocketAddress dnSocketAddr = getServiceRpcServerAddress(conf);  
  5.     if (dnSocketAddr != null) {  
  6.       int serviceHandlerCount =  
  7.         conf.getInt(DFSConfigKeys.DFS_NAMENODE_SERVICE_HANDLER_COUNT_KEY,  
  8.                     DFSConfigKeys.DFS_NAMENODE_SERVICE_HANDLER_COUNT_DEFAULT);  
  9.       //获得serviceRpcServer  
  10.       this.serviceRpcServer = RPC.getServer(this, dnSocketAddr.getHostName(),   
  11.           dnSocketAddr.getPort(), serviceHandlerCount,  
  12.           false, conf, namesystem.getDelegationTokenSecretManager());  
  13.       this.serviceRPCAddress = this.serviceRpcServer.getListenerAddress();  
  14.       setRpcServiceServerAddress(conf);  
  15. }  
  16. //获得server  
  17.     this.server = RPC.getServer(this, socAddr.getHostName(),  
  18.         socAddr.getPort(), handlerCount, false, conf, namesystem  
  19.         .getDelegationTokenSecretManager());  
  20.   
  21.    •••  
  22.     this.server.start();  //启动 RPC server   Clients只允许连接该server  
  23.     if (serviceRpcServer != null) {  
  24.       serviceRpcServer.start();  //启动 RPC serviceRpcServer 为HDFS服务的server  
  25.     }  
  26.     startTrashEmptier(conf);  
  27.   }  

 

查看Namenode初始化源码得知:RPC的server对象是通过ipc.RPC类的getServer()方法获得的。下面咱们去看看ipc.RPC类中的getServer()源码吧:

代码十一:

Java代码   收藏代码
  1. public static Server getServer(final Object instance, final String bindAddress, final int port,  
  2.                                  final int numHandlers,  
  3.                                  final boolean verbose, Configuration conf,  
  4.                                  SecretManager<? extends TokenIdentifier> secretManager)   
  5.     throws IOException {  
  6.     return new Server(instance, conf, bindAddress, port, numHandlers, verbose, secretManager);  
  7.   }  

 

这时我们发现getServer()是一个创建Server对象的工厂方法,但创建的却是RPC.Server类的对象。哈哈,现在你明白了我前面说的“RPC.Server是ipc.Server的实现类”了吧。不过RPC.Server的构造函数还是调用了ipc.Server类的构造函数的,因篇幅所限,就不贴出相关源码了。

2. 运行Server 
如代码十所示,初始化Server后,Server端就运行起来了,看看ipc.Server的start()源码吧:

代码十二:

Java代码   收藏代码
  1. /** 启动服务 */  
  2. public synchronized void start() {  
  3.   responder.start();  //启动responder  
  4.   listener.start();   //启动listener  
  5.   handlers = new Handler[handlerCount];  
  6.     
  7.   for (int i = 0; i < handlerCount; i++) {  
  8.     handlers[i] = new Handler(i);  
  9.     handlers[i].start();   //逐个启动Handler  
  10.   }  
  11. }  

 

3. Server处理请求 

1)建立连接 
分析过ipc.Client源码后,我们知道Client端的底层通信直接采用了阻塞式IO编程,当时我们曾做出猜测:Server端是不是也采用了阻塞式IO。现在我们仔细地分析一下吧,如果Server端也采用阻塞式IO,当连接进来的Client端很多时,势必会影响Server端的性能。hadoop的实现者们考虑到了这点,所以他们采用了java  NIO来实现Server端,java NIO可参考博客:
http://weixiaolu.iteye.com/blog/1479656  。那Server端采用java NIO是怎么建立连接的呢?分析源码得知,Server端采用Listener监听客户端的连接,下面先分析一下Listener的构造函数吧:

代码十三:

Java代码   收藏代码
  1. public Listener() throws IOException {  
  2.   address = new InetSocketAddress(bindAddress, port);  
  3.   // 创建ServerSocketChannel,并设置成非阻塞式  
  4.   acceptChannel = ServerSocketChannel.open();  
  5.   acceptChannel.configureBlocking(false);  
  6.   
  7.   // 将server socket绑定到本地端口  
  8.   bind(acceptChannel.socket(), address, backlogLength);  
  9.   port = acceptChannel.socket().getLocalPort();   
  10.   // 获得一个selector  
  11.   selector= Selector.open();  
  12.   readers = new Reader[readThreads];  
  13.   readPool = Executors.newFixedThreadPool(readThreads);  
  14.   //启动多个reader线程,为了防止请求多时服务端响应延时的问题  
  15.   for (int i = 0; i < readThreads; i++) {         
  16.     Selector readSelector = Selector.open();  
  17.     Reader reader = new Reader(readSelector);  
  18.     readers[i] = reader;  
  19.     readPool.execute(reader);  
  20.   }  
  21.   // 注册连接事件  
  22.   acceptChannel.register(selector, SelectionKey.OP_ACCEPT);  
  23.   this.setName("IPC Server listener on " + port);  
  24.   this.setDaemon(true);  
  25. }  

 

在启动Listener线程时,服务端会一直等待客户端的连接,下面贴出Server.Listener类的run()方法:

代码十四:

Java代码   收藏代码
  1. public void run() {  
  2.    •••  
  3.     while (running) {  
  4.       SelectionKey key = null;  
  5.       try {  
  6.         selector.select();  
  7.         Iterator<SelectionKey> iter = selector.selectedKeys().iterator();  
  8.         while (iter.hasNext()) {  
  9.           key = iter.next();  
  10.           iter.remove();  
  11.           try {  
  12.             if (key.isValid()) {  
  13.               if (key.isAcceptable())  
  14.                 doAccept(key);     //具体的连接方法  
  15.             }  
  16.           } catch (IOException e) {  
  17.           }  
  18.           key = null;  
  19.         }  
  20.       } catch (OutOfMemoryError e) {  
  21.      •••           
  22.   }  

 

下面贴出Server.Listener类中doAccept ()方法中的关键源码吧:

代码十五:

Java代码   收藏代码
  1.     void doAccept(SelectionKey key) throws IOException,  OutOfMemoryError {  
  2.       Connection c = null;  
  3.       ServerSocketChannel server = (ServerSocketChannel) key.channel();  
  4.       SocketChannel channel;  
  5.       while ((channel = server.accept()) != null) { //建立连接  
  6.         channel.configureBlocking(false);  
  7.         channel.socket().setTcpNoDelay(tcpNoDelay);  
  8.         Reader reader = getReader();  //从readers池中获得一个reader  
  9.         try {  
  10.           reader.startAdd(); // 激活readSelector,设置adding为true  
  11.           SelectionKey readKey = reader.registerChannel(channel);//将读事件设置成兴趣事件  
  12.           c = new Connection(readKey, channel, System.currentTimeMillis());//创建一个连接对象  
  13.           readKey.attach(c);   //将connection对象注入readKey  
  14.           synchronized (connectionList) {  
  15.             connectionList.add(numConnections, c);  
  16.             numConnections++;  
  17.           }  
  18.         •••   
  19.         } finally {  
  20. //设置adding为false,采用notify()唤醒一个reader,其实代码十三中启动的每个reader都使  
  21. //用了wait()方法等待。因篇幅有限,就不贴出源码了。  
  22.           reader.finishAdd();  
  23.         }  
  24.       }  
  25.     }  

 

当reader被唤醒,reader接着执行doRead()方法。

2)接收请求 
下面贴出Server.Listener.Reader类中的doRead()方法和Server.Connection类中的readAndProcess()方法源码:

代码十六:

Java代码   收藏代码
  1. 方法一:     
  2.  void doRead(SelectionKey key) throws InterruptedException {  
  3.       int count = 0;  
  4.       Connection c = (Connection)key.attachment();  //获得connection对象  
  5.       if (c == null) {  
  6.         return;    
  7.       }  
  8.       c.setLastContact(System.currentTimeMillis());  
  9.       try {  
  10.         count = c.readAndProcess();    // 接受并处理请求    
  11.       } catch (InterruptedException ieo) {  
  12.        •••  
  13.       }  
  14.      •••      
  15. }  
  16.   
  17. 方法二:  
  18. public int readAndProcess() throws IOException, InterruptedException {  
  19.       while (true) {  
  20.         •••  
  21.         if (!rpcHeaderRead) {  
  22.           if (rpcHeaderBuffer == null) {  
  23.             rpcHeaderBuffer = ByteBuffer.allocate(2);  
  24.           }  
  25.          //读取请求头  
  26.           count = channelRead(channel, rpcHeaderBuffer);  
  27.           if (count < 0 || rpcHeaderBuffer.remaining() > 0) {  
  28.             return count;  
  29.           }  
  30.         // 读取请求版本号    
  31.           int version = rpcHeaderBuffer.get(0);  
  32.           byte[] method = new byte[] {rpcHeaderBuffer.get(1)};  
  33.         •••    
  34.          
  35.           data = ByteBuffer.allocate(dataLength);  
  36.         }  
  37.         // 读取请求    
  38.         count = channelRead(channel, data);  
  39.           
  40.         if (data.remaining() == 0) {  
  41.          •••  
  42.           if (useSasl) {  
  43.          •••  
  44.           } else {  
  45.             processOneRpc(data.array());//处理请求  
  46.           }  
  47.         •••  
  48.           }  
  49.         }   
  50.         return count;  
  51.       }  
  52.     }  

 

3)获得call对象 
下面贴出Server.Connection类中的processOneRpc()方法和processData()方法的源码。

代码十七:

Java代码   收藏代码
  1. 方法一:     
  2.  private void processOneRpc(byte[] buf) throws IOException,  
  3.         InterruptedException {  
  4.       if (headerRead) {  
  5.         processData(buf);  
  6.       } else {  
  7.         processHeader(buf);  
  8.         headerRead = true;  
  9.         if (!authorizeConnection()) {  
  10.           throw new AccessControlException("Connection from " + this  
  11.               + " for protocol " + header.getProtocol()  
  12.               + " is unauthorized for user " + user);  
  13.         }  
  14.       }  
  15. }  
  16. 方法二:  
  17.     private void processData(byte[] buf) throws  IOException, InterruptedException {  
  18.       DataInputStream dis =  
  19.         new DataInputStream(new ByteArrayInputStream(buf));  
  20.       int id = dis.readInt();      // 尝试读取id  
  21.       Writable param = ReflectionUtils.newInstance(paramClass, conf);//读取参数  
  22.       param.readFields(dis);          
  23.           
  24.       Call call = new Call(id, param, this);  //封装成call  
  25.       callQueue.put(call);   // 将call存入callQueue  
  26.       incRpcCount();  // 增加rpc请求的计数  
  27.     }  

 

4)处理call对象 
你还记得Server类中还有个Handler内部类吗?呵呵,对call对象的处理就是它干的。下面贴出Server.Handler类中run()方法中的关键代码:

代码十八:

Java代码   收藏代码
  1. while (running) {  
  2.       try {  
  3.         final Call call = callQueue.take(); //弹出call,可能会阻塞  
  4.         •••  
  5.         //调用ipc.Server类中的call()方法,但该call()方法是抽象方法,具体实现在RPC.Server类中  
  6.         value = call(call.connection.protocol, call.param, call.timestamp);  
  7.         synchronized (call.connection.responseQueue) {  
  8.           setupResponse(buf, call,   
  9.                       (error == null) ? Status.SUCCESS : Status.ERROR,   
  10.                       value, errorClass, error);  
  11.            •••  
  12.           //给客户端响应请求  
  13.           responder.doRespond(call);  
  14.         }  
  15. }  

 

 

5)返回请求 
下面贴出Server.Responder类中的doRespond()方法源码:
 

代码十九:

Java代码   收藏代码
  1. 方法一:     
  2.  void doRespond(Call call) throws IOException {  
  3.       synchronized (call.connection.responseQueue) {  
  4.         call.connection.responseQueue.addLast(call);  
  5.         if (call.connection.responseQueue.size() == 1) {  
  6.            // 返回响应结果,并激活writeSelector  
  7.           processResponse(call.connection.responseQueue, true);  
  8.         }  
  9.       }  
  10. }  

 

小结:


到这里,hadoop RPC机制的源码分析就结束了,请继续关注我的后续博客:hadoop心跳机制的源码分析。在这里需要感谢我所参考的iteye上相关博主的文章,因太多了,就不一一列举了,不过最感谢的是wikieno的博客,博客地址为:
http://www.wikieno.com/2012/02/hadoop-ipc-server/ 。

分享到:
评论

相关推荐

    Hadoop RPC机制分析

    本文将深入探讨Hadoop的RPC机制,解析其工作原理,并结合源码分析其内部实现。 一、RPC简介 RPC是一种让程序能够调用运行在其他地址空间(通常在另一台机器上)的程序的方法。在Hadoop中,RPC被广泛用于NameNode、...

    [HBase]源码级强力分析hadoop的RPC机制

    这些天一直奔波于长沙和武汉之间,忙着腾讯的笔试、面试,以至于对hadoopRPC(RemoteProcedureCallProtocol,远程过程调用协议,它是一种通过网络从远程计算机程序上请求服务,而不需要了解底层网络技术的协议。...

    Hadoop_RPC详细分析.doc

    Hadoop RPC 详细分析 Hadoop RPC(Remote Procedure Call,远程过程调用)是 Hadoop 项目中的一个重要组件,用于实现分布式系统中的通信和数据交换。下面是对 Hadoop RPC 的详细分析。 RPCInterface Hadoop RPC ...

    Hadoop rpc源码

    Hadoop rpc源码是从Hadoop分离出的ipc,去掉了认证部分,附录使用文档.使用前请add lib包commons-logging-*.*.*.jar(我用的是1.0.4)和log4j-*.*.*.jar(我的1.2.13) 相关blog post: ...

    hadooprpc机制&&将avro引入hadooprpc机制初探

    RPC(RemoteProcedureCall)——远程过程调用,它是一种通过网络从远程计算机程序上请求服务,而不需要了解底层网络技术的协议。RPCServer实现了一种抽象的RPC服务,同时提供Call队列。RPCServer作为服务提供者由两个...

    hadoop rpc实例

    Hadoop RPC是Hadoop框架中用于进程间通信的一种机制,它允许一个进程调用另一个远程进程中定义的方法,仿佛这个方法是在本地执行一样。这种透明性使得开发者可以专注于业务逻辑,而无需关心底层通信细节。 ### 工作...

    Hadoop源码分析(完整版)

    Hadoop源码分析是深入理解Hadoop分布式计算平台原理的起点,通过源码分析,可以更好地掌握Hadoop的工作机制、关键组件的实现方式和内部通信流程。Hadoop项目包括了多个子项目,其中最核心的是HDFS和MapReduce,这两...

    java操作hadoop的RPC,源码

    Hadoop作为一个开源的大数据处理框架,其RPC机制是实现各个组件如NameNode、DataNode、TaskTracker等之间通信的核心。以下是对这个主题的详细讲解: 1. **RPC基础概念**: - RPC是一种进程间通信(IPC)方式,它...

    Hadoop里的RPC机制过程

    在Hadoop中,远程过程调用(Remote Procedure Call, RPC)是一种重要的通信机制,它允许分布式系统中的组件之间进行高效且便捷的交互。Hadoop的RPC机制基于Java的客户端-服务器模型,允许客户端调用服务器上的方法,...

    Hadoop源代码分析(完整版).pdf

    Hadoop 的源代码分析可以帮助开发者更好地理解 Hadoop 的架构和实现机制,从而更好地使用 Hadoop 实现大数据处理和分析。 Hadoop 的关键部分集中在图中的蓝色部分,这也是我们考察的重点。Hadoop 的包的功能分析...

    hadoop-2.8.1源码

    本文将围绕"Hadoop 2.8.1源码"这一主题,带你深入理解这个强大系统的核心机制,揭示其背后的分布式存储和计算原理。 Hadoop的核心由两个主要组件构成:HDFS(Hadoop Distributed File System)和MapReduce。HDFS是...

    Hadoop源码分析视频下载

    分析Hadoop源码可以帮助我们深入了解其实现细节,例如数据块的分配策略、心跳机制、故障检测和恢复等。这不仅可以帮助我们优化Hadoop集群的性能,还能使我们有能力解决实际运行中遇到的问题,提高开发效率。 4. ...

    2022毕业设计,基于 Hadoop 的游戏数据分析系统源码.zip

    【标题】:“2022毕业设计,基于 Hadoop 的游戏数据分析系统源码” 这个毕业设计项目主要聚焦于使用Hadoop框架开发一个游戏数据分析系统。Hadoop是Apache软件基金会的一个开源分布式计算平台,专为处理和存储大规模...

    基于Hadoop的网站流量日志数据分析系统项目源码+教程.zip

    基于Hadoop网站流量日志数据分析系统项目源码+教程.zip网站流量日志数据分析系统 典型的离线流数据分析系统 技术分析 hadoop nginx flume hive sqoop mysql springboot+mybatisplus+vcharts 基于Hadoop网站流量日志...

    基于Hadoop框架的豆瓣大数据分析与搜索系统源码.zip

    基于Hadoop框架的豆瓣大数据分析与搜索系统源码.zip基于Hadoop框架的豆瓣大数据分析与搜索系统源码.zip基于Hadoop框架的豆瓣大数据分析与搜索系统源码.zip基于Hadoop框架的豆瓣大数据分析与搜索系统源码.zip基于...

    Hadoop豆瓣电影数据分析(Hadoop)操作源码

    Hadoop豆瓣电影数据分析(Hadoop)操作源码

    Hadoop Java接口+RPC代码实现

    1.java接口操作Hadoop文件系统(文件上传下载删除创建......2.RPC远程过程调用的java代码实现,便于理解Hadoop的RPC协议,具体使用方法可参考我的博客https://blog.csdn.net/qq_34233510/article/details/88142507

Global site tag (gtag.js) - Google Analytics