1. protocol在rpc中的作用
通过对org.apache.hadoop.ipc包分析中，Hadoop实现了基于IPC模型的RPC机制，可以不需要像Java中实现的RMI机制一样，在RPC调用的C/S两端分别创建Stub和Skeleton，而是通过一组协议来进行RPC调用就可以实现通信。这主要是由于Hadoop所采用的序列化机制简化了RPC调用的复杂性。Hadoop定义了自己的通信协议，这些协议都是建立在TCP/IP协议之上的，规范了通信两端的约定。
Hadoop集群中，不同进程之间通信需要使用合适的协议才能够进行交互。在协议接口中约定了通信双方的特定行为，那么，在实现这些通信协议的实现类中，就能看到指定进程是如何实现协议接口中约定的行为的。
2. hadoop实现了那些protocol
所有要使用RPC服务的类都要实现该接口VersionedProtocol，我们可以来看一下有哪些接口继承了该接口。
VersionedProtocol协议是Hadoop的最顶层协议接口的抽象；
1）HDFS相关
- ClientDatanodeProtocol：client与datanode交互的接口，操作不多，只有一个block恢复的方法。那么，其它数据请求的方法呢？client与datanode主要交互是通过流式的socket实现，源码在DataXceiver，在这里先不说了；
- 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 脐带的, 关系亲密的） 。
3）其它
- AdminOperationProtocol：不用用户操作的接口，提供一些管理操作，如刷新JobTracker的node列表；
- RefreshAuthorizationPolicyProtocol，RefreshUserMappingsProtocol：暂不明白。
3. 心跳机制分析
hadoop的集群是基于master/slave模式，namenode和jobtracker属于master，而datanode/tasktracker属于slaves。master只有一个，而slaves有多个。 namenode与datanode之间的通信，jobtracker与tasktracker直接的通信，都是通过“心跳”完成的。
心跳的机制大概是这样的：
1) master启动的时候，会开一个ipc server在那里。
2) slave启动时，会连接master，并每隔3秒钟主动向master发送一个“心跳”，将自己的状态信息告诉master，然后master也是通过这个心跳的返回值，向slave节点传达指令。
3.1 namenode和datanode的心跳
在datanode.java代码里面offerService方法中，每隔3秒向namenode发送心跳的代码
public void offerService() throws Exception { while (shouldRun) { try { long startTime = now(); // 如果到了3秒钟，就向namenode发心跳 if (startTime - lastHeartbeat > heartBeatInterval) { // // All heartbeat messages include following info: // -- Datanode name // -- data transfer port // -- Total capacity // -- Bytes remaining // lastHeartbeat = startTime; DatanodeCommand[] cmds = namenode.sendHeartbeat(dnRegistration, data.getCapacity(), data.getDfsUsed(), data.getRemaining(), xmitsInProgress.get(), getXceiverCount() ); // 注意上面这行代码，“发送心跳”竟然就是调用namenode的一个方法？？ // 处理对心跳的返回值（namenode传给datanode的指令） if ( ! processCommand(cmds)) continue ; } } ... } // while (shouldRun) } // offerService
现在的问题是datanode怎么获得namenode对象的？继续往下看
public class DataNode extends Configured implements InterDatanodeProtocol, ClientDatanodeProtocol, FSConstants, Runnable { ... public DatanodeProtocol namenode = null ; // DatanodeProtocol ... void startDataNode(Configuration conf, AbstractList < File > dataDirs, SecureResources resources) throws IOException { this .namenode = (DatanodeProtocol) RPC.waitForProxy(DatanodeProtocol. class , DatanodeProtocol.versionID, nameNodeAddr, conf); // DatanodeProtocol } } public class NameNode implements ClientProtocol, DatanodeProtocol, NamenodeProtocol, FSConstants, RefreshAuthorizationPolicyProtocol, RefreshUserMappingsProtocol // NameNode实现了DatanodeProtocol接口，DatanodeProtocol接口定义了namenode和datanode之间通信的方法。 public interface DatanodeProtocol extends VersionedProtocol { ... public DatanodeRegistration register(DatanodeRegistration registration) throws IOException; /** * sendHeartbeat() tells the NameNode that the DataNode is still alive and well. Includes some status info, too. * It also gives the NameNode a chance to return an array of "DatanodeCommand" objects. * A DatanodeCommand tells the DataNode to invalidate local block(s), or to copy them to other DataNodes, etc. */ public DatanodeCommand[] sendHeartbeat(DatanodeRegistration registration, long capacity, long dfsUsed, long remaining, int xmitsInProgress, int xceiverCount) throws IOException; public DatanodeCommand blockReport(DatanodeRegistration registration, long [] blocks) throws IOException; ... }
- 1) 对namenode的赋值，并不是真正的new了一个实现了DatanodeProtocol接口的对象，而是获得了一个动态代理！！
- 2) 上面这段代码中，protocol的类型是DatanodeProtocol.class
- 3) 对namenode的所有调用，都被委托(delegate)给了Invoker
剩下的就是和之前分析的client端和server端的流程一样了,总结一下流程就是:
datanode向namenode发送heartbeat过程是这样的：
a) 在datanode初始化获得namenode的proxy
b) 在datanode上，调用namenode proxy的heartbeat方法： namenode.sendHeartbeat(dnRegistration, data.getCapacity(), data.getDfsUsed(), data.getRemaining(), xmitsInProgress.get(), getXceiverCount()); c) 在datanode上的namenode动态代理类将这个调用包装成(或者叫“序列化成”)一个Invocation对象，并调用client.call方法 d) client call方法将Invocation转化为Call对象 e) client 将call发送到真正的namenode服务器 f) namenode接收后，转化成namenode端的Call，并process后，通过Responder发回来！ g) datanode接收结果，并将结果转化为DatanodeCommand[]
3.2 jobtracker和tasktracker的心跳
org.apache.hadoop.mapred.TaskTracker
// 代码一： State offerService() throws Exception { long lastHeartbeat = System.currentTimeMillis(); while (running && ! shuttingDown) { // 发送心跳，调用代码二 HeartbeatResponse heartbeatResponse = transmitHeartBeat(now); return State.NORMAL; } } // 代码二： HeartbeatResponse transmitHeartBeat( long now) throws IOException { HeartbeatResponse heartbeatResponse = jobClient.heartbeat(status, justStarted, justInited, askForNewTask, heartbeatResponseId); return heartbeatResponse; }
剩下的基本上和上面流程差不多....
4. 用户自定义的protocol使用hadoop rpc
MyProtocol.java
import org.apache.hadoop.io.Text; import org.apache.hadoop.ipc.VersionedProtocol; public interface MyProtocol extends VersionedProtocol { public Text println(Text t); }
MyServer.java
import java.io.IOException; import java.net.UnknownHostException; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.io.Text; import org.apache.hadoop.ipc.RPC; import org.apache.hadoop.ipc.RPC.Server; public class MyServer implements MyProtocol { private Server server; public MyServer() { try { server = RPC.getServer( this , " localhost " , 8888 , new Configuration()); server.start(); server.join(); } catch (UnknownHostException e) { e.printStackTrace(); } catch (IOException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } } @Override public Text println(Text t) { System.out.println(t); return new Text( " finish " ); } @Override public long getProtocolVersion(String protocol, long clientVersion) throws IOException { return 1 ; } public static void main(String[] args) { new MyServer(); } }
MyClient.java
import java.io.IOException; import java.net.InetSocketAddress; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.io.Text; import org.apache.hadoop.ipc.RPC; public class MyClient { private MyProtocol proxy; public MyClient() { InetSocketAddress addr = new InetSocketAddress( " localhost " , 8888 ); try { proxy = (MyProtocol) RPC.waitForProxy(MyProtocol. class , 1 , addr , new Configuration()); } catch (IOException e) { e.printStackTrace(); } } public void println(String s) { System.out.println(proxy.println( new Text(s))); } public void close() { RPC.stopProxy(proxy); } public static void main(String[] args) { MyClient c = new MyClient(); c.println( " 123 " ); c.close(); } }
结果输出:
在server端输出
12 / 06 / 23 17 : 11 : 30 INFO ipc.Server: Starting Socket Reader # 1 for port 6789 12 / 06 / 23 17 : 11 : 30 INFO metrics.RpcMetrics: Initializing RPC Metrics with hostName = MyServer, port = 6789 12 / 06 / 23 17 : 11 : 30 INFO metrics.RpcDetailedMetrics: Initializing RPC Metrics with hostName = MyServer, port = 6789 12 / 06 / 23 17 : 11 : 30 INFO ipc.Server: IPC Server Responder: starting 12 / 06 / 23 17 : 11 : 30 INFO ipc.Server: IPC Server listener on 6789 : starting 12 / 06 / 23 17 : 11 : 30 INFO ipc.Server: IPC Server handler 0 on 6789 : starting 123
在client端输出
finish
5. 参考url