学步园

DFSClient是分布式文件系统客户端，它能够连接到Hadoop文件系统执行指定任务，那么它要与Namenode与Datanode基于一定的协议来进行通信。这个通信过程中，涉及到不同进程之间的通信。在org.apache.hadoop.ipc包中，定义了进程间通信的Client端与Server端的抽象，也就是基于C/S模式进行通信。这里先对org.apache.hadoop.ipc包中有关类的源代码阅读分析。

首先看该包中类的继承关系，如下所示：

。java.lang.Object<br /> 。org.apache.hadoop.ipc.Client<br /> 。org.apache.hadoop.ipc.RPC<br /> 。org.apache.hadoop.ipc.Server<br /> 。org.apache.hadoop.ipc.RPC.Server<br /> 。java.lang.Throwable (implements java.io.Serializable)<br /> 。java.lang.Exception<br /> 。java.io.IOException<br /> 。org.apache.hadoop.ipc.RemoteException<br /> 。org.apache.hadoop.ipc.RPC.VersionMismatch

我阅读该包源程序的方法是，先从C/S通讯的两端Client类与Server类来阅读分析，然后再对实现的一个RPC类进行分析。

Client类

首先从Client客户端类的实现开始，该类定义了如下属性：

private Hashtable<ConnectionId, Connection> connections = new Hashtable<ConnectionId, Connection>(); // 客户端维护到服务端的一组连接<br /> private Class<? extends Writable> valueClass; // class of call values<br /> private int counter; // counter for call ids<br /> private AtomicBoolean running = new AtomicBoolean(true); // 客户端进程是否在运行<br /> final private Configuration conf; // 配置类实例<br /> final private int maxIdleTime; // 连接的最大空闲时间<br /> final private int maxRetries; //Socket连接时，最大Retry次数<br /> private boolean tcpNoDelay; // 设置TCP连接是否延迟<br /> private int pingInterval; // ping服务端的间隔</p> <p> private SocketFactory socketFactory; // Socket工厂，用来创建Socket连接<br /> private int refCount = 1;</p> <p> final private static String PING_INTERVAL_NAME = "ipc.ping.interval"; // 通过配置文件读取ping间隔<br /> final static int DEFAULT_PING_INTERVAL = 60000; // 默认ping间隔为1分钟<br /> final static int PING_CALL_ID = -1;

从属性可以看出，一个Clinet主要处理的是与服务端进行连接的工作，包括连接的创建、监控等。为了能够了解到Client如何实现它所抽象的操作，先分别看一下该类定义的5个内部类：

• Client.Call内部类

该内部类，是客户端调用的一个抽象，主要定义了一次调用所需要的条件，以及修改Client客户端的一些全局统计变量，如下所示：

private class Call {<br /> int id; // 调用ID<br /> Writable param; // 调用参数<br /> Writable value; // 调用返回的值<br /> IOException error; // 异常信息<br /> boolean done; // 调用是否完成</p> <p> protected Call(Writable param) {<br /> this.param = param;<br /> synchronized (Client.this) {<br /> this.id = counter++; // 互斥修改法：对多个连接的调用线程进行统计<br /> }<br /> }</p> <p> /** 调用完成，设置标志，唤醒其它线程 */<br /> protected synchronized void callComplete() {<br /> this.done = true;<br /> notify(); // 唤醒其它调用者<br /> }</p> <p> /**<br /> * 调用出错，同样置调用完成标志，并设置出错信息<br /> */<br /> public synchronized void setException(IOException error) {<br /> this.error = error;<br /> callComplete();<br /> }</p> <p> /**<br /> * 调用完成，设置调用返回的值<br /> */<br /> public synchronized void setValue(Writable value) {<br /> this.value = value;<br /> callComplete();<br /> }<br /> }

上面的Call内部类主要是对一次调用的实例进行监视与管理，即使获取调用返回值，如果出错则获取出错信息，同时修改Client全局统计变量。

• Client.ConnectionId内部类

该内部类是一个连接的实体类，标识了一个连接实例的Socket地址、用户信息UserGroupInformation、连接的协议类。每个连接都是通过一个该类的实例唯一标识。如下所示：

InetSocketAddress address;<br /> UserGroupInformation ticket;<br /> Class<?> protocol;<br /> private static final int PRIME = 16777619;

该类中有一个用来判断两个连接ConnectionId实例是否相等的equals方法：

@Override<br /> public boolean equals(Object obj) {<br /> if (obj instanceof ConnectionId) {<br /> ConnectionId id = (ConnectionId) obj;<br /> return address.equals(id.address) && protocol == id.protocol && ticket == id.ticket;<br /> }<br /> return false;<br /> }

只有当Socket地址、用户信息UserGroupInformation、连接的协议类这三个属性的值相等时，才被认为是同一个ConnectionId实例。

• Client.ParallelResults内部类

该内部类是用来收集在并行调用环境中结果的实体类，如下所示：

private static class ParallelResults {<br /> private Writable[] values; // 并行调用对应多次调用，对应多个返回值<br /> private int size; // 并行调用返回值个数统计<br /> private int count; // 并行调用次数</p> <p> public ParallelResults(int size) {<br /> this.values = new Writable[size];<br /> this.size = size;<br /> }</p> <p> /** 收集并行调用返回值 */<br /> public synchronized void callComplete(ParallelCall call) {<br /> values[call.index] = call.value; // 存储返回值<br /> count++; // 统计返回值个数<br /> if (count == size) // 并行调用的多个调用完成<br /> notify(); // 唤醒下一个实例<br /> }<br /> }

• Client.ParallelCall内部类

该内部类继承自上面的内部类Call，只是返回值使用上面定义的ParallelResults实体类来封装，如下所示：

private class ParallelCall extends Call {<br /> private ParallelResults results;<br /> private int index;</p> <p> public ParallelCall(Writable param, ParallelResults results, int index) {<br /> super(param);<br /> this.results = results;<br /> this.index = index;<br /> }</p> <p> /** 收集并行调用返回结果值 */<br /> protected void callComplete() {<br /> results.callComplete(this);<br /> }<br /> }

• Client.Connection内部类

该类是一个连接管理内部线程类，该内部类是一个连接线程，继承自Thread类。它读取每一个Call调用实例执行后从服务端返回的响应信息，并通知其他调用实例。每一个连接具有一个连接到远程主机的Socket，该Socket能够实现多路复用，使得多个调用复用该Socket，客户端收到的调用得到的响应可能是无序的。

该类定义的属性如下所示：

private InetSocketAddress server; // 服务端ip:port<br /> private ConnectionHeader header; // 连接头信息，该实体类封装了连接协议与用户信息UserGroupInformation<br /> private ConnectionId remoteId; // 连接ID</p> <p> private Socket socket = null; // 客户端已连接的Socket<br /> private DataInputStream in;<br /> private DataOutputStream out;</p> <p> private Hashtable<Integer, Call> calls = new Hashtable<Integer, Call>(); // 当前活跃的调用列表<br /> private AtomicLong lastActivity = new AtomicLong();// 最后I/O活跃的时间<br /> private AtomicBoolean shouldCloseConnection = new AtomicBoolean(); // 连接是否关闭<br /> private IOException closeException; // 连接关闭原因

上面使用到了java.util.concurrent.atomic包中的一些工具，像AtomicLong、AtomicBoolean，这些类能够以原子方式更新其值，支持在单个变量上解除锁二实现线程的安全。这些类能够使用get方法读取volatile变量的内存效果，set方法可以设置对应变量的内存值。通过后面的代码可以看到该类工具类的使用。例如：

private void touch() {<br /> lastActivity.set(System.currentTimeMillis());<br /> }

上面定义的calls集合，是用来保存当前活跃的调用实例，以键值对的形式保存，键是一个Call的id，值是Call的实例，因此，该类一定提供了向该集合中添加新的调用实例、移除调用实例等等操作，分别将方法签名列表如下：

/**<br /> * 向calls集合中添加一个<Call.id, Call><br /> */<br /> private synchronized boolean addCall(Call call);</p> <p> /*<br /> * 等待某个调用线程唤醒自己，可能开始如下操作：<br /> * 1、读取RPC响应数据<br /> * 2、idle时间过长<br /> * 3、被标记为应该关闭<br /> * 4、客户端已经终止<br /> */<br /> private synchronized boolean waitForWork();</p> <p> /*<br /> * 接收到响应（因为每次从DataInputStream in中读取响应信息只有一个，无需同步）<br /> */<br /> private void receiveResponse();</p> <p> /*<br /> * 关闭连接，需要迭代calls集合，清除连接<br /> */<br /> private synchronized void close();

可以看到，当每次调用touch方法的时候，都会将lastActivity原子变量设置为系统的当前时间，更新了变量的值。该操作是对多个线程进行互斥的，也就是每次修改lastActivity的值的时候，都会对该变量加锁，从内存中读取该变量的当前值，因此可能会出现阻塞的情况。

下面看一个Connection实例的构造实现：

public Connection(ConnectionId remoteId) throws IOException {<br /> this.remoteId = remoteId; // 远程服务端连接<br /> this.server = remoteId.getAddress(); // 远程服务器地址<br /> if (server.isUnresolved()) {<br /> throw new UnknownHostException("unknown host: " + remoteId.getAddress().getHostName());<br /> }</p> <p> UserGroupInformation ticket = remoteId.getTicket(); // 用户信息<br /> Class<?> protocol = remoteId.getProtocol(); // 协议<br /> header = new ConnectionHeader(protocol == null ? null : protocol.getName(), ticket); // 连接头信息<br /> this.setName("IPC Client (" + socketFactory.hashCode() +") connection to " + remoteId.getAddress().toString() + " from " + ((ticket==null)?"an unknown user":ticket.getUserName()));<br /> this.setDaemon(true); // 并设置一个连接为后台线程<br /> }

通过Collection实例的构造，可以看到，客户端所拥有的Connection实例，通过一个远程ConnectionId实例来建立到客户端到服务端的连接。接着看一下Connection线程类线程体代码：

public void run() {<br /> if (LOG.isDebugEnabled())<br /> LOG.debug(getName() + ": starting, having connections " + connections.size());</p> <p> while (waitForWork()) {// 等待某个连接实例空闲，如果存在则唤醒它执行一些任务<br /> receiveResponse(); // 接收RPC响应<br /> } </p> <p> close(); // 关闭</p> <p> if (LOG.isDebugEnabled())<br /> LOG.debug(getName() + ": stopped, remaining connections " + connections.size());<br /> }

客户端Client类提供的最基本的功能就是执行RPC调用，其中，提供了两种调用方式，一种就是串行单个调用，另一种就是并行调用，分别介绍如下。首先是串行单个调用的实现方法call，如下所示：

/*<br /> * 执行一个调用，通过传递参数值param到运行在addr上的IPC服务器，IPC服务器基于protocol与用户的ticket来认证，并响应客户端<br /> */<br /> public Writable call(Writable param, InetSocketAddress addr, Class<?> protocol, UserGroupInformation ticket) throws InterruptedException, IOException {<br /> Call call = new Call(param); // 使用请求参数值构造一个Call实例<br /> Connection connection = getConnection(addr, protocol, ticket, call); // 从连接池connections中获取到一个连接（或可能创建一个新的连接）<br /> connection.sendParam(call); // 向IPC服务器发送参数<br /> boolean interrupted = false;<br /> synchronized (call) {<br /> while (!call.done) {<br /> try {<br /> call.wait(); // 等待IPC服务器响应<br /> } catch (InterruptedException ie) {<br /> interrupted = true;<br /> }<br /> }</p> <p> if (interrupted) {<br /> // set the interrupt flag now that we are done waiting<br /> Thread.currentThread().interrupt();<br /> }</p> <p> if (call.error != null) {<br /> if (call.error instanceof RemoteException) {<br /> call.error.fillInStackTrace();<br /> throw call.error;<br /> } else { // local exception<br /> throw wrapException(addr, call.error);<br /> }<br /> } else {<br /> return call.value; // 调用返回的响应值<br /> }<br /> }<br /> }

然后，就是并行调用的实现call方法，如下所示：

/*<br /> * 执行并行调用<br /> * 每个参数都被发送到相关的IPC服务器，然后等待服务器响应信息<br /> */<br /> public Writable[] call(Writable[] params, InetSocketAddress[] addresses, Class<?> protocol, UserGroupInformation ticket) throws IOException {<br /> if (addresses.length == 0) return new Writable[0];<br /> ParallelResults results = new ParallelResults(params.length); // 根据待调用的参数个数来构造一个用来封装并行调用返回值的ParallelResults对象<br /> synchronized (results) {<br /> for (int i = 0; i < params.length; i++) {<br /> ParallelCall call = new ParallelCall(params[i], results, i); // 构造并行调用实例<br /> try {<br /> Connection connection = getConnection(addresses[i], protocol, ticket, call); // 从连接池中获取到一个连接<br /> connection.sendParam(call); // 发送调用参数<br /> } catch (IOException e) {<br /> LOG.info("Calling "+addresses[i]+" caught: " + e.getMessage(),e);<br /> results.size--; // 更新统计数据<br /> }<br /> }<br /> while (results.count != results.size) {<br /> try {<br /> results.wait(); // 等待一组调用的返回值（如果调用失败，会返回错误信息或null值，但与计数相关）<br /> } catch (InterruptedException e) {}<br /> }</p> <p> return results.values; // 调用返回一组响应值<br /> }<br /> }

客户端可以根据服务端暴露的远程地址，来与服务器建立连接，并执行RPC调用，发送调用参数数据。

Server端有点复杂，后面单独分析其实现过程和机制。