java重试机制代码设计 RPC原理及实例分析(4)

zookeeper提供了“心跳检测”功能，它会定时向各个服务提供者发送一个请求（实际上建立的是一个 Socket 长连接），如果长期没有响应，服务中心就认为该服务提供者已经“挂了”，并将其剔除，比如100.19.20.02这台机器如果宕机了，那么zookeeper上的路径就会只剩/HelloWorldService/1.0.0/100.19.20.01:16888。

服务消费者会去相应路径（/HelloWorldService/1.0.0），一旦路径上的数据有任务变化（增加或减少），zookeeper都会通知服务消费方服务提供者地址列表已经发生改变，从而进行更新。

更为重要的是zookeeper与生俱来的容错容灾能力（比如leader选举），可以确保服务注册表的高可用性。

3.Hadoop中RPC实例分析

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

Invocation ：用于封装方法名和参数，作为数据传输层。

ClientCache ：用于存储client对象，用socket factory作为hash key,存储结构为hashMap

    public Object invoke(Object proxy, Method method, Object[] args)
      throws Throwable {
      ??????
      ObjectWritable value = (ObjectWritable)
        client.call(new Invocation(method, args), remoteId);
      ??????
      return value.get();
    }

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

ObjectWritable value = (ObjectWritable)
client.call(new Invocation(method, args), remoteId);

Invocation类在这里封装了方法名和参数。其实这里网络通信只是调用了Client类的call()方法。那我们接下来分析一下ipc.Client源码吧。和第一章一样，同样是3个问题

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

3.1 客户端和服务端的连接是怎样建立的？

public Writable call(Writable param, ConnectionId remoteId)  
                       throws InterruptedException, IOException {
    Call call = new Call(param);       //将传入的数据封装成call对象
    Connection connection = getConnection(remoteId, call);   //获得一个连接
    connection.sendParam(call);     // 向服务端发送call对象
    boolean interrupted = false;
    synchronized (call) {
      while (!call.done) {
        try {
          call.wait(); // 等待结果的返回，在Call类的callComplete()方法里有notify()方法用于唤醒线程
        } catch (InterruptedException ie) {
          // 因中断异常而终止，设置标志interrupted为true
          interrupted = true;
        }
      }
      if (interrupted) {
        Thread.currentThread().interrupt();
      }

      if (call.error != null) {
        if (call.error instanceof RemoteException) {
          call.error.fillInStackTrace();
          throw call.error;
        } else { // 本地异常
          throw wrapException(remoteId.getAddress(), call.error);
        }
      } else {
        return call.value; //返回结果数据
      }
    }
  }

本文来自电脑杂谈，转载请注明本文网址：
http://www.pc-fly.com/a/jisuanjixue/article-77937-4.html