Hive on Spark源码分析（三）—— SparkClilent与SparkClientImpl（上）

  
  
// 提交一个异步执行的job，返回一个用于监控job的JobHandle
<T extends Serializable> JobHandle<T> submit(Job<T> job);
// 请求远程context执行job。该方法忽视job队列，建议仅用于执行快速结束的任务。
// 返回一个用于监控job的Future结果
<T extends Serializable> Future<T> run(Job<T> job);
/**
   * Stops the remote context.
   *
   * Any pending jobs will be cancelled, and the remote context will be torn down.
   */
  void stop();
  /**
   * Adds a jar file to the running remote context.
   *
   * Note that the URL should be reachable by the Spark driver process. If running the driver
   * in cluster mode, it may reside on a different host, meaning "file:" URLs have to exist
   * on that node (and not on the client machine).
   *
   * @param uri The location of the jar file.
   * @return A future that can be used to monitor the operation.
   */
  Future<?> addJar(URI uri);
  /**
   * Adds a file to the running remote context.
   *
   * Note that the URL should be reachable by the Spark driver process. If running the driver
   * in cluster mode, it may reside on a different host, meaning "file:" URLs have to exist
   * on that node (and not on the client machine).
   *
   * @param uri The location of the file.
   * @return A future that can be used to monitor the operation.
   */
  Future<?> addFile(URI uri);
  /**
   * Get the count of executors.
   */
  Future<Integer> getExecutorCount();
  /**
   * Get default parallelism. For standalone mode, this can be used to get total number of cores.
   */
  Future<Integer> getDefaultParallelism();
  /**
   * Check if remote context is still active.
   */
  boolean isActive();

  
  
    this.conf = conf;
    this.hiveConf = hiveConf;
    this.childIdGenerator = new AtomicInteger();
    this.jobs = Maps.newConcurrentMap();
    String clientId = UUID.randomUUID().toString();
    //产生secret用于与remoteDriver建立连接时的身份认证
    String secret = rpcServer.createSecret();
    //startDriver用来在新的进程中启动RemoteDriver,并返回一个接受执行结果的线程
    this.driverThread = startDriver(rpcServer, clientId, secret);
    //创建ClientProtocol用于rpc通信,发送消息（提交任务）,以及响应远端发来的消息
    this.protocol = new ClientProtocol();

  
  
    try {
      // The RPC server will take care of #rpc client connection# timeouts here.
      //向远端RPCServer注册rpc客户端
      this.driverRpc = rpcServer.registerClient(clientId, secret, protocol).get();
    } catch (Throwable e) {
      if (e.getCause() instanceof TimeoutException) {
        LOG.error("Timed out waiting for client to connect.\nPossible reasons include network " +
            "issues, errors in remote driver or the cluster has no available resources, etc." +
            "\nPlease check YARN or Spark driver's logs for further information.\nReason2 from SparkClientImpl", e);
      } else {
        LOG.error("Error while waiting for client to connect.", e);
      }
      //终端driverThread线程
      driverThread.interrupt();
      try {
        //等待driverThread挂掉
        driverThread.join();
      } catch (InterruptedException ie) {
        // Give up.
        LOG.debug("Interrupted before driver thread was finished.");
      }
      throw Throwables.propagate(e);
    }

  
  
    driverRpc.addListener(new Rpc.Listener() {
        @Override
        public void rpcClosed(Rpc rpc) {
          //如果rpc通信关闭时,当前SparkClient仍然是alive的,则打印warn信息
          if (isAlive) {
            LOG.warn("Client RPC channel closed unexpectedly.");
            isAlive = false;
          }
        }
    });
    //实例化后标记为isAlive
    isAlive = true;

  
  
  private Thread startDriver(final RpcServer rpcServer, final String clientId, final String secret)
      throws IOException {
    Runnable runnable;
    final String serverAddress = rpcServer.getAddress();
    final String serverPort = String.valueOf(rpcServer.getPort());