Hadoop RPC通信Client客户端的流程分析

/** * Superclass of all protocols that use Hadoop RPC. * Subclasses of this interface are also supposed to have * a static final long versionID field. * Hadoop RPC所有协议的基类，返回协议版本号 */ public interface VersionedProtocol { /** * Return protocol version corresponding to protocol interface. * @param protocol The classname of the protocol interface * @param clientVersion The version of the protocol that the client speaks * @return the version that the server will speak */ public long getProtocolVersion(String protocol, long clientVersion) throws IOException; }

/** A client for an IPC service. IPC calls take a single {@link Writable} as a * parameter, and return a {@link Writable} as their value. A service runs on * a port and is defined by a parameter class and a value class. * RPC客户端类 * @see Server */ public class Client { public static final Log LOG = LogFactory.getLog(Client.class); //客户端到服务真个连接 private Hashtable<ConnectionId, Connection> connections = new Hashtable<ConnectionId, Connection>(); //回调值类 private Class<? extends Writable> valueClass; // class of call values //call回调id的计数器 private int counter; // counter for call ids //原子变量判断客户端是不是还在运行 private AtomicBoolean running = new AtomicBoolean(true); // if client runs final private Configuration conf; //socket工厂，用来创建socket private SocketFactory socketFactory; // how to create sockets private int refCount = 1; ......

/** * This class holds the address and the user ticket. The client connections * to servers are uniquely identified by <remoteAddress, protocol, ticket> * 连接的唯1标识，主要通过<远程地址，协议类型，用户组信息> */ static class ConnectionId { //远程的socket地址 InetSocketAddress address; //用户组信息 UserGroupInformation ticket; //协议类型 Class<?> protocol; private static final int PRIME = 16777619; private int rpcTimeout; private String serverPrincipal; private int maxIdleTime; //connections will be culled if it was idle for //maxIdleTime msecs private int maxRetries; //the max. no. of retries for socket connections private boolean tcpNoDelay; // if T then disable Nagle's Algorithm private int pingInterval; // how often sends ping to the server in msecs ....

/** * 作者重写了equal比较方法，只要成员变量都想等也就想到了 */ @Override public boolean equals(Object obj) { if (obj == this) { return true; } if (obj instanceof ConnectionId) { ConnectionId that = (ConnectionId) obj; return isEqual(this.address, that.address) && this.maxIdleTime == that.maxIdleTime && this.maxRetries == that.maxRetries && this.pingInterval == that.pingInterval && isEqual(this.protocol, that.protocol) && this.rpcTimeout == that.rpcTimeout && isEqual(this.serverPrincipal, that.serverPrincipal) && this.tcpNoDelay == that.tcpNoDelay && isEqual(this.ticket, that.ticket); } return false; } /** * 重写了hashCode的生成规则，保证不同的对象产生不同的hashCode值 */ @Override public int hashCode() { int result = 1; result = PRIME * result + ((address == null) ? 0 : address.hashCode()); result = PRIME * result + maxIdleTime; result = PRIME * result + maxRetries; result = PRIME * result + pingInterval; result = PRIME * result + ((protocol == null) ? 0 : protocol.hashCode()); result = PRIME * rpcTimeout; result = PRIME * result + ((serverPrincipal == null) ? 0 : serverPrincipal.hashCode()); result = PRIME * result + (tcpNoDelay ? 1231 : 1237); result = PRIME * result + ((ticket == null) ? 0 : ticket.hashCode()); return result; }

/** Thread that reads responses and notifies callers. Each connection owns a * socket connected to a remote address. Calls are multiplexed through this * socket: responses may be delivered out of order. */ private class Connection extends Thread { //所连接的服务器地址 private InetSocketAddress server; // server ip:port //服务真个krb5的名字，与安全方面相干 private String serverPrincipal; // server's krb5 principal name //连接头部，内部包括了，所用的协议，客户端用户组信息和验证的而方法 private ConnectionHeader header; // connection header //远程连接ID private final ConnectionId remoteId; // connection id //连接验证方法 private AuthMethod authMethod; // authentication method //下面3个变量都是安全方面的 private boolean useSasl; private Token<? extends TokenIdentifier> token; private SaslRpcClient saslRpcClient; //下面是1组socket通讯方面的变量 private Socket socket = null; // connected socket private DataInputStream in; private DataOutputStream out; private int rpcTimeout; private int maxIdleTime; //connections will be culled if it was idle for //maxIdleTime msecs private int maxRetries; //the max. no. of retries for socket connections //tcpNoDelay可设置是不是阻塞模式 private boolean tcpNoDelay; // if T then disable Nagle's Algorithm private int pingInterval; // how often sends ping to the server in msecs // currently active calls 当前活跃的回调，1个连接 可能会有很多个call回调 private Hashtable<Integer, Call> calls = new Hashtable<Integer, Call>(); //最后1次IO活动通讯的时间 private AtomicLong lastActivity = new AtomicLong();// last I/O activity time //连接关闭标记 private AtomicBoolean shouldCloseConnection = new AtomicBoolean(); // indicate if the connection is closed private IOException closeException; // close reason .....

class ConnectionHeader implements Writable { public static final Log LOG = LogFactory.getLog(ConnectionHeader.class); //客户端和服务端通讯的协议名称 private String protocol; //客户真个用户组信息 private UserGroupInformation ugi = null; //验证的方式，关系到写入数据的时的格式 private AuthMethod authMethod; .....

/** A call waiting for a value. */ //客户真个1个回调 private class Call { //回调ID int id; // call id //被序列化的参数 Writable param; // parameter //返回值 Writable value; // value, null if error //出错时返回的异常 IOException error; // exception, null if value //回调是不是已被完成 boolean done; // true when call is done ....

private class Connection extends Thread { .... // currently active calls 当前活跃的回调，1个连接 可能会有很多个call回调 private Hashtable<Integer, Call> calls = new Hashtable<Integer, Call>();

/** Call implementation used for parallel calls. */ /** 继承自Call回调类，可以并行的使用，通过加了index下标做Call的辨别 */ private class ParallelCall extends Call { //每一个ParallelCall并行的回调就会有对应的结果类 private ParallelResults results; //index作为Call的辨别 private int index; ....

/** Result collector for parallel calls. */ private static class ParallelResults { //并行结果类中具有1组返回值，需要ParallelCall的index索引匹配 private Writable[] values; //结果值的数量 private int size; //values中已知的值的个数 private int count; ..... /** Collect a result. */ public synchronized void callComplete(ParallelCall call) { //将call中的值赋给result中 values[call.index] = call.value; // store the value count++; // count it //如果计数的值等到终究大小，通知caller if (count == size) // if all values are in notify(); // then notify waiting caller } }

public Writable call(Writable param, InetSocketAddress addr, Class<?> protocol, UserGroupInformation ticket, int rpcTimeout) throws InterruptedException, IOException { ConnectionId remoteId = ConnectionId.getConnectionId(addr, protocol, ticket, rpcTimeout, conf); return call(param, remoteId); }

public Writable call(Writable param, ConnectionId remoteId) throws InterruptedException, IOException { //根据参数构造1个Call回调 Call call = new Call(param); //根据远程ID获得连接 Connection connection = getConnection(remoteId, call); //发送参数 connection.sendParam(call); // send the parameter boolean interrupted = false; synchronized (call) { //如果call.done为false，就是Call还没完成 while (!call.done) { try { //等待远端程序的履行终了 call.wait(); // wait for the result } catch (InterruptedException ie) { // save the fact that we were interrupted interrupted = true; } } //如果是异常中断，则终止当前线程 if (interrupted) { // set the interrupt flag now that we are done waiting Thread.currentThread().interrupt(); } //如果call回到出错，则返回call出错信息 if (call.error != null) { if (call.error instanceof RemoteException) { call.error.fillInStackTrace(); throw call.error; } else { // local exception // use the connection because it will reflect an ip change, unlike // the remoteId throw wrapException(connection.getRemoteAddress(), call.error); } } else { //如果是正常情况下，返回回调解理后的值 return call.value; } } }

private Connection getConnection(ConnectionId remoteId, Call call) throws IOException, InterruptedException { ..... /* we could avoid this allocation for each RPC by having a * connectionsId object and with set() method. We need to manage the * refs for keys in HashMap properly. For now its ok. */ do { synchronized (connections) { //从connection连接池中获得连接，可以保证相同的连接ID可以复用 connection = connections.get(remoteId); if (connection == null) { connection = new Connection(remoteId); connections.put(remoteId, connection); } } } while (!connection.addCall(call));

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); //for serializing the //data to be written //将call回调中的参数写入到输出流中，传向服务端 d = new DataOutputBuffer(); d.writeInt(call.id); call.param.write(d); byte[] data = d.getData(); int dataLength = d.getLength(); out.writeInt(dataLength); //first put the data length out.write(data, 0, dataLength);//write the data out.flush(); } ....

public void run() { if (LOG.isDebugEnabled()) LOG.debug(getName() + ": starting, having connections " + connections.size()); //等待工作，等待要求调用 while (waitForWork()) {//wait here for work - read or close connection //调用完要求，则立即获得回复 receiveResponse(); } close(); if (LOG.isDebugEnabled()) LOG.debug(getName() + ": stopped, remaining connections " + connections.size()); }

private synchronized boolean waitForWork() { if (calls.isEmpty() && !shouldCloseConnection.get() && running.get()) { long timeout = maxIdleTime- (System.currentTimeMillis()-lastActivity.get()); if (timeout>0) { try { wait(timeout); } catch (InterruptedException e) {} } } ....

/* Receive a response. * Because only one receiver, so no synchronization on in. * 获得回复值 */ private void receiveResponse() { if (shouldCloseConnection.get()) { return; } //更新最近1次的call活动时间 touch(); try { int id = in.readInt(); // try to read an id if (LOG.isDebugEnabled()) LOG.debug(getName() + " got value #" + id); //从获得call中获得相应的call Call call = calls.get(id); //判断该结果状态 int state = in.readInt(); // read call status if (state == Status.SUCCESS.state) { Writable value = ReflectionUtils.newInstance(valueClass, conf); value.readFields(in); // read value call.setValue(value); calls.remove(id); } else if (state == Status.ERROR.state) { call.setException(new RemoteException(WritableUtils.readString(in), WritableUtils.readString(in))); calls.remove(id); } else if (state == Status.FATAL.state) { // Close the connection markClosed(new RemoteException(WritableUtils.readString(in), WritableUtils.readString(in))); } ..... } catch (IOException e) { markClosed(e); } }