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

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;
}

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 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 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请求的计数
}

void doRespond(Call call) throws IOException
{
    synchronized (call.connection.responseQueue)
    {
        call.connection.responseQueue.addLast(call);
        if (call.connection.responseQueue.size() == 1)
        {
            // 返回响应结果，并激活writeSelector
            processResponse(call.connection.responseQueue, true);
        }
    }
}