第六章：小朱笔记hadoop之源码分析-ipc分析 第四节：RPC类分析

renshanshan

第六章：小朱笔记hadoop之源码分析-ipc分析

第四节：RPC类分析

RPC类是对Server、Client的具体化。在RPC类中规定，客户程序发出请求调用时，参数类型必须是Invocation；从服务器返回的值类型必须是ObjectWritable。
RPC类是对Server、Client的包装，简化用户的使用。如果一个类需充当服务器，只需通过RPC类的静态方法getServer获得Server实例，然后start。同时此类提供协议接口的实现。如果一个类充当客户端，可以通过getProxy或者waitForProxy获得一个实现了协议接口的proxy object，与服务器端交互。
RPC类中有5个静态内部类，分别为：

写道

Invocation ：用于封装方法名和参数，作为数据传输层，相当于VO吧。
ClientCache ：用于存储client对象，用socket factory作为hash key,存储结构为hashMap <SocketFactory, Client>。
Invoker ：是动态代理中的调用实现类，继承了InvocationHandler.
Server ： org.apache.hadoop.ipc.Server的具体实现类，实现了抽象类的call方法，获得传入参数的call实例，再获取method方法，反射调用即可。
VersionMismatch：版本不匹配异常。

 
（1）Invocation
 

    /** A method invocation, including the method name and its parameters. */
private static class Invocation implements Writable, Configurable {
private String methodName; // 方法名
private Class[] parameterClasses; // 参数类型集合
private Object[] parameters; // 参数值
private Configuration conf; // 配置类实例
......
}
（2）ClientCache
 

    /* Cache a client using its socket factory as the hash key */
static private class ClientCache {
// 该内部类定义了一个缓存Map
private Map<SocketFactory, Client> clients = new HashMap<SocketFactory, Client>();
/**
* Construct & cache an IPC client with the user-provided SocketFactory
* if no cached client exists.
* 通过客户端org.apache.hadoop.ipc.Client的SocketFactory可以快速取出对应的Client实例
*
* @param conf
*            Configuration
* @return an IPC client
*/
/**
* 从缓存Map中取出一个IPC Client实例，如果缓存够中不存在，就创建一个兵加入到缓存Map中
*/
private synchronized Client getClient(Configuration conf, SocketFactory factory) {
// Construct & cache client. The configuration is only used for
// timeout,
// and Clients have connection pools. So we can either (a) lose some
// connection pooling and leak sockets, or (b) use the same timeout
// for all
// configurations. Since the IPC is usually intended globally, not
// per-job, we choose (a).
Client client = clients.get(factory);
if (client == null) {
client = new Client(ObjectWritable.class, conf, factory); // 通过反射实例化一个ObjectWritable对象，构造Client实例
clients.put(factory, client);
} else {
client.incCount();
}
return client;
}
/**
* Construct & cache an IPC client with the default SocketFactory if no
* cached client exists.
*
* @param conf
*            Configuration
* @return an IPC client
*/
private synchronized Client getClient(Configuration conf) {
return getClient(conf, SocketFactory.getDefault());
}
/**
* Stop a RPC client connection A RPC client is closed only when its
* reference count becomes zero.
*/
private void stopClient(Client client) {
synchronized (this) {
client.decCount(); // 该client实例的引用计数减1
if (client.isZeroReference()) { // 如果client实例的引用计数此时为0
clients.remove(client.getSocketFactory()); // 从缓存中删除
}
}
if (client.isZeroReference()) { // 如果client实例引用计数为0，需要关闭
client.stop(); // 停止所有与该client实例相关的线程
}
}
}
（3）Invoker
 

private static class Invoker implements InvocationHandler {
private Client.ConnectionId remoteId;
private Client client;
private boolean isClosed = false;
public Invoker(Class<? extends VersionedProtocol> protocol, InetSocketAddress address, UserGroupInformation ticket,
Configuration conf, SocketFactory factory, int rpcTimeout) throws IOException {
this.remoteId = Client.ConnectionId.getConnectionId(address, protocol, ticket, rpcTimeout, conf);
this.client = CLIENTS.getClient(conf, factory);
}
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
final boolean logDebug = LOG.isDebugEnabled();
long startTime = 0;
if (logDebug) {
startTime = System.currentTimeMillis();
}
// 构造一个RPC.Invocation实例作为参数传递给调用程序，执行调用，返回值为value
ObjectWritable value = (ObjectWritable) client.call(new Invocation(method, args), remoteId);
if (logDebug) {
long callTime = System.currentTimeMillis() - startTime;
LOG.debug("Call: " + method.getName() + " " + callTime);
}
return value.get();
}
/* close the IPC client that's responsible for this invoker's RPCs */
synchronized private void close() {
if (!isClosed) {
isClosed = true;
CLIENTS.stopClient(client);
}
}
}
 
（4）Server

public Writable call(Class<?> protocol, Writable param, long receivedTime)throws IOException {
Invocation call = (Invocation)param;
// 通过反射，根据调用方法名和方法参数类型得到Method实例  
Method method =protocol.getMethod(call.getMethodName(),
call.getParameterClasses());
method.setAccessible(true);// 设置反射的对象在使用时取消Java语言访问检查，提高效率  
Object value = method.invoke(instance, call.getParameters());// 执行调用（instance是调用底层方法的对象，第二个参数是方法调用的参数）     
}