大数据框架hadoop的IPC应用场景之getNewJobId

43fdw

                      

作为典型的分布式系统，Hadoop中各个实体间存在着大量的交互，远程过程调用让用户可以像调用本地方法一样调用另外一个应用程序提供的服务，而不必设计和开发相关的信息发送、处理和接收等具体代码，是一种重要的分布式计算技术，它提高了程序的互操作性，在Hadoop的实现中得到广泛的应用。

Hadoop没有使用Java RMI，而是实现了一套自己独有的节点间通信机制，理由是有效的IPC（Inter-Process Communication，进程间通信）对于Hadoop来说是至关重要的，Hadoop需要精确控制进程间通信中比如连接、超时、缓存等通信细节。

如下是Hadoop 框架中实现了客户端调用JobTracker服务器端方法的一个具体应用，功能是返回分配给新作业的ID串。接下来从客户端调用、服务器端接口及功能实现和服务器端的监听来展现这一功能。

1      客户端调用类JobClientpublicclass JobClient extends Configured implements MRConstants, Tool  {
  private JobSubmissionProtocol jobSubmitClient;

  // 客户端通过代理对象直接进行方法调用。
  public RunningJob submitJobInternal(final JobConf job) {

......

    JobID jobId = jobSubmitClient.getNewJobId();

... ...

  }
  // 如下是代理对象的构造方法，首先判断当前hadoop环境是否是本地的，如果不是本地的则构造一个要请求服务器的网络地址（IP和端口），网络地址的信息来自于配置文件中属性mapred.job.tracker的配置，如果配置文件没有该项配置，则取默认值local。
  publicvoid init(JobConf conf) throws IOException {
    String tracker = conf.get("mapred.job.tracker", "local");
    if ("local".equals(tracker)) {
      conf.setNumMapTasks(1);
      this.jobSubmitClient = new LocalJobRunner(conf);
    } else {
      this.jobSubmitClient = createRPCProxy(JobTracker.getAddress(conf), conf);
    }        
  }
    在IPC中，使用RPC.getProxy()方法获取到一个JobSubmissionProtocol代理对象。
  privatestatic JobSubmissionProtocol createRPCProxy(InetSocketAddressaddr,Configuration conf) throws IOException {
    return (JobSubmissionProtocol) RPC.getProxy(JobSubmissionProtocol.class,
       JobSubmissionProtocol.versionID, addr,
       UserGroupInformation.getCurrentUser(),conf,
        NetUtils.getSocketFactory(conf, JobSubmissionProtocol.class));
  }
2      接口JobSubmissionProtocol    如下是服务器端服务接口定义：
interface JobSubmissionProtocolextends VersionedProtocol {
  publicstaticfinallongversionID = 28L;
  /**
   * Allocate a name forthe job.
   * @return a unique job name for submitting jobs.
   * @throws IOException
   */
  public JobID getNewJobId() throws IOException;
}
3      实现类JobTracker    如下是服务器端服务实现类定义：
publicclass JobTracker implements MRConstants, InterTrackerProtocol,
    JobSubmissionProtocol,TaskTrackerManager, RefreshUserMappingsProtocol,
   RefreshAuthorizationPolicyProtocol, AdminOperationsProtocol,
    JobTrackerMXBean {
  /**
   * Allocates a new JobIdstring.
   */
  publicsynchronized JobID getNewJobId() throws IOException {
    returnnew JobID(getTrackerIdentifier(),nextJobId++);
  }
}
4      服务器监听类

在服务器端先构造一个要提供服务的网络地址addr，传递到RPC.getServer()方法中。服务器端使用RPC.getServer()方法创建服务器端对象interTrackerServer。

Server端的Handler实例（线程）的个数为由配置文件的属性mapred.job.tracker.handler.count指定。

publicclass JobTracker implements MRConstants, InterTrackerProtocol,
    JobSubmissionProtocol,TaskTrackerManager, RefreshUserMappingsProtocol,
   RefreshAuthorizationPolicyProtocol, AdminOperationsProtocol,
   JobTrackerMXBean {
JobTracker(final JobConf conf, String identifier, Clock clock, QueueManagerqm) {

... ...

    InetSocketAddress addr = getAddress(conf);

... ...

inthandlerCount = conf.getInt("mapred.job.tracker.handler.count", 10);

    this.interTrackerServer =
      RPC.getServer(this, addr.getHostName(), addr.getPort(), handlerCount,
          false, conf, secretManager);
    ... ...
  }

  publicstatic InetSocketAddress getAddress(Configuration conf) {
    String jobTrackerStr =
      conf.get("mapred.job.tracker", "localhost:8012");
    return NetUtils.createSocketAddr(jobTrackerStr);
  }
}