rpc原理及rpc实例分析(转)(代码片段)

public class Test 
     public static void main(String[] args) 
         HelloWorldService helloWorldService = new HelloWorldServiceImpl();
         helloWorldService.sayHello("test");
     

public class RPCProxyClient implements java.lang.reflect.InvocationHandler
    private Object obj;

    public RPCProxyClient(Object obj)
        this.obj=obj;
    

    /**
     * 得到被代理对象;
     */
    public static Object getProxy(Object obj)
        return java.lang.reflect.Proxy.newProxyInstance(obj.getClass().getClassLoader(),
                obj.getClass().getInterfaces(), new RPCProxyClient(obj));
    

    /**
     * 调用此方法执行
     */
    public Object invoke(Object proxy, Method method, Object[] args)
            throws Throwable 
        //结果参数;
        Object result = new Object();
        // ...执行通信相关逻辑
        // ...
        return result;
    

public class Test 
     public static void main(String[] args) 
         HelloWorldService helloWorldService = (HelloWorldService)RPCProxyClient.getProxy(HelloWorldService.class);
         helloWorldService.sayHello("test");
     
 

public Object get() 
        synchronized (this)  // 旋锁
            while (!isDone)  // 是否有结果了
                wait(); //没结果是释放锁，让当前线程处于等待状态
            
        

private void setDone(Response res) 
        this.res = res;
        isDone = true;
        synchronized (this)  //获取锁，因为前面wait()已经释放了callback的锁了
            notifyAll(); // 唤醒处于等待的线程
        
    

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

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

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; //返回结果数据
      
    
  

Connection connection = getConnection(remoteId, call);   //获得一个连接
  connection.sendParam(call);      // 向服务端发送call对象

private Connection getConnection(ConnectionId remoteId,
                                   Call call)
                                   throws IOException, InterruptedException 
    if (!running.get()) 
      // 如果client关闭了
      throw new IOException("The client is stopped");
    
    Connection connection;
//如果connections连接池中有对应的连接对象，就不需重新创建了；如果没有就需重新创建一个连接对象。
//但请注意，该//连接对象只是存储了remoteId的信息，其实还并没有和服务端建立连接。
    do 
      synchronized (connections) 
        connection = connections.get(remoteId);
        if (connection == null) 
          connection = new Connection(remoteId);
          connections.put(remoteId, connection);
        
      
     while (!connection.addCall(call)); //将call对象放入对应连接中的calls池，就不贴出源码了
   //这句代码才是真正的完成了和服务端建立连接哦~
    connection.setupIOstreams();
    return connection;
  

private synchronized void setupIOstreams() throws InterruptedException 
   •••
      try 
       •••
        while (true) 
          setupConnection();  //建立连接
          InputStream inStream = NetUtils.getInputStream(socket);     //获得输入流
          OutputStream outStream = NetUtils.getOutputStream(socket);  //获得输出流
          writeRpcHeader(outStream);
          •••
          this.in = new DataInputStream(new BufferedInputStream
              (new PingInputStream(inStream)));   //将输入流装饰成DataInputStream
          this.out = new DataOutputStream
          (new BufferedOutputStream(outStream));   //将输出流装饰成DataOutputStream
          writeHeader();
          // 跟新活动时间
          touch();
          //当连接建立时，启动接受线程等待服务端传回数据，注意：Connection继承了Tread
          start();
          return;
        
       catch (IOException e) 
        markClosed(e);
        close();
      
    

private synchronized void setupConnection() throws IOException 
      short ioFailures = 0;
      short timeoutFailures = 0;
      while (true) 
        try 
          this.socket = socketFactory.createSocket(); //终于看到创建socket的方法了
          this.socket.setTcpNoDelay(tcpNoDelay);
         •••
          // 设置连接超时为20s
          NetUtils.connect(this.socket, remoteId.getAddress(), 20000);
          this.socket.setSoTimeout(pingInterval);
          return;
         catch (SocketTimeoutException toe) 
          /* 设置最多连接重试为45次。
           * 总共有20s*45 = 15 分钟的重试时间。
           */
          handleConnectionFailure(timeoutFailures++, 45, toe);
         catch (IOException ie) 
          handleConnectionFailure(ioFailures++, maxRetries, ie);
        
      
    

public void sendParam(Call call) 
      if (shouldCloseConnection.get()) 
        return;
      
      DataOutputBuffer d=null;
      try 
        synchronized (this.out) 
          if (LOG.isDebugEnabled())
            LOG.debug(getName() + " sending #" + call.id);
          //创建一个缓冲区
          d = new DataOutputBuffer();
          d.writeInt(call.id);
          call.param.write(d);
          byte[] data = d.getData();
          int dataLength = d.getLength();
          out.writeInt(dataLength);        //首先写出数据的长度
          out.write(data, 0, dataLength); //向服务端写数据
          out.flush();
        
       catch(IOException e) 
        markClosed(e);
       finally 
        IOUtils.closeStream(d);
      
      

方法一：  
  public void run() 
      •••
      while (waitForWork()) 
        receiveResponse();  //具体的处理方法
      
      close();
     •••


方法二：
private void receiveResponse() 
      if (shouldCloseConnection.get()) 
        return;
      
      touch();
      try 
        int id = in.readInt();                    // 阻塞读取id
        if (LOG.isDebugEnabled())
          LOG.debug(getName() + " got value #" + id);
          Call call = calls.get(id);    //在calls池中找到发送时的那个对象
        int state = in.readInt();     // 阻塞读取call对象的状态
        if (state == Status.SUCCESS.state) 
          Writable value = ReflectionUtils.newInstance(valueClass, conf);
          value.readFields(in);           // 读取数据
        //将读取到的值赋给call对象，同时唤醒Client等待线程，贴出setValue()代码方法三
          call.setValue(value);              
          calls.remove(id);               //删除已处理的call    
         else if (state == Status.ERROR.state) 
        •••
         else if (state == Status.FATAL.state) 
        •••
        
       catch (IOException e) 
        markClosed(e);
      


方法三：
public synchronized void setValue(Writable value) 
      this.value = value;
      callComplete();   //具体实现

protected synchronized void callComplete() 
      this.done = true;
      notify();         // 唤醒client等待线程
    

private void initialize(Configuration conf) throws IOException 
   •••
    // 创建 rpc server
    InetSocketAddress dnSocketAddr = getServiceRpcServerAddress(conf);
    if (dnSocketAddr != null) 
      int serviceHandlerCount =
        conf.getInt(DFSConfigKeys.DFS_NAMENODE_SERVICE_HANDLER_COUNT_KEY,
                    DFSConfigKeys.DFS_NAMENODE_SERVICE_HANDLER_COUNT_DEFAULT);
      //获得serviceRpcServer
      this.serviceRpcServer = RPC.getServer(this, dnSocketAddr.getHostName(), 
          dnSocketAddr.getPort(), serviceHandlerCount,
          false, conf, namesystem.getDelegationTokenSecretManager());
      this.serviceRPCAddress = this.serviceRpcServer.getListenerAddress();
      setRpcServiceServerAddress(conf);

//获得server
    this.server = RPC.getServer(this, socAddr.getHostName(),
        socAddr.getPort(), handlerCount, false, conf, namesystem
        .getDelegationTokenSecretManager());

   •••
    this.server.start();  //启动 RPC server   Clients只允许连接该server
    if (serviceRpcServer != null) 
      serviceRpcServer.start();  //启动 RPC serviceRpcServer 为HDFS服务的server
    
    startTrashEmptier(conf);
  

public static Server getServer(final Object instance, final String bindAddress, final int port,
                                 final int numHandlers,
                                 final boolean verbose, Configuration conf,
                                 SecretManager<? extends TokenIdentifier> secretManager) 
    throws IOException 
    return new Server(instance, conf, bindAddress, port, numHandlers, verbose, secretManager);
  

  /** 启动服务 */
  public synchronized void start() 
    responder.start();  //启动responder
    listener.start();   //启动listener
    handlers = new Handler[handlerCount];
    
    for (int i = 0; i < handlerCount; i++) 
      handlers[i] = new Handler(i);
      handlers[i].start();   //逐个启动Handler
    
  

 public Listener() throws IOException 
      address = new InetSocketAddress(bindAddress, port);
      // 创建ServerSocketChannel,并设置成非阻塞式
      acceptChannel = ServerSocketChannel.open();
      acceptChannel.configureBlocking(false);

      // 将server socket绑定到本地端口
      bind(acceptChannel.socket(), address, backlogLength);
      port = acceptChannel.socket().getLocalPort(); 
      // 获得一个selector
      selector= Selector.open();
      readers = new Reader[readThreads];
      readPool = Executors.newFixedThreadPool(readThreads);
      //启动多个reader线程，为了防止请求多时服务端响应延时的问题
      for (int i = 0; i < readThreads; i++)        
        Selector readSelector = Selector.open();
        Reader reader = new Reader(readSelector);
        readers[i] = reader;
        readPool.execute(reader);
      
      // 注册连接事件
      acceptChannel.register(selector, SelectionKey.OP_ACCEPT);
      this.setName("IPC Server listener on " + port);
      this.setDaemon(true);
    

public void run() 
     •••
      while (running) 
        SelectionKey key = null;
        try 
          selector.select();
          Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
          while (iter.hasNext()) 
            key = iter.next();
            iter.remove();
            try 
              if (key.isValid()) 
                if (key.isAcceptable())
                  doAccept(key);     //具体的连接方法
              
             catch (IOException e) 
            
            key = null;
          
         catch (OutOfMemoryError e) 
       •••         
    

void doAccept(SelectionKey key) throws IOException,  OutOfMemoryError 
      Connection c = null;
      ServerSocketChannel server = (ServerSocketChannel) key.channel();
      SocketChannel channel;
      while ((channel = server.accept()) != null)  //建立连接
        channel.configureBlocking(false);
        channel.socket().setTcpNoDelay(tcpNoDelay);
        Reader reader = getReader();  //从readers池中获得一个reader
        try 
          reader.startAdd(); // 激活readSelector，设置adding为true
          SelectionKey readKey = reader.registerChannel(channel);//将读事件设置成兴趣事件
          c = new Connection(readKey, channel, System.currentTimeMillis());//创建一个连接对象
          readKey.attach(c);   //将connection对象注入readKey
          synchronized (connectionList) 
            connectionList.add(numConnections, c);
            numConnections++;
          
        ••• 
         finally 
//设置adding为false，采用notify()唤醒一个reader,其实代码十三中启动的每个reader都使
//用了wait()方法等待。因篇幅有限，就不贴出源码了。
          reader.finishAdd();
        
      
    

方法一：   
 void doRead(SelectionKey key) throws InterruptedException 
      int count = 0;
      Connection c = (Connection)key.attachment();  //获得connection对象
      if (c == null) 
        return;  
      
      c.setLastContact(System.currentTimeMillis());
      try 
        count = c.readAndProcess();    // 接受并处理请求  
       catch (InterruptedException ieo) 
       •••
      
     •••    


方法二：
public int readAndProcess() throws IOException, InterruptedException 
      while (true) 
        •••
        if (!rpcHeaderRead) 
          if (rpcHeaderBuffer == null) 
            rpcHeaderBuffer = ByteBuffer.allocate(2);
          
         //读取请求头
          count = channelRead(channel, rpcHeaderBuffer);
          if (count < 0 || rpcHeaderBuffer.remaining() > 0) 
            return count;
          
        // 读取请求版本号  
          int version = rpcHeaderBuffer.get(0);
          byte[] method = new byte[] rpcHeaderBuffer.get(1);
        •••  
       
          data = ByteBuffer.allocate(dataLength);
        
        // 读取请求  
        count = channelRead(channel, data);
        
        if (data.remaining() == 0) 
         •••
          if (useSasl) 
         •••
           else 
            processOneRpc(data.array());//处理请求
          
        •••
          
         
        return count;
      
    

方法一：   
 private void processOneRpc(byte[] buf) throws IOException,
        InterruptedException 
      if (headerRead) 
        processData(buf);
       else 
        processHeader(buf);
        headerRead = true;
        if (!authorizeConnection()) 
          throw new AccessControlException("Connection from " + this
              + " for protocol " + header.getProtocol()
              + " is unauthorized for user " + user);
        
      

方法二：
    private void processData(byte[] buf) throws  IOException, InterruptedException 
      DataInputStream dis =
        new DataInputStream(new ByteArrayInputStream(buf));
      int id = dis.readInt();      // 尝试读取id
      Writable param = ReflectionUtils.newInstance(paramClass, conf);//读取参数
      param.readFields(dis);        
        
      Call call = new Call(id, param, this);  //封装成call
      callQueue.put(call);   // 将call存入callQueue
      incRpcCount();  // 增加rpc请求的计数