hadoop IPC 源代码分析

如图所示，在hadoop中客户端需要和服务端通信。首先我们看一下需求是啥。

举一个例子，在客户端想要往hadoop集群中写数据的时候，它需要先和namenode通信，以便获得诸一个blockID。

这时，我们希望在客户端可以做到诸如调用一个方法如 getBlockID（）则就获得了服务端的发过来的ID ，如果调用本地方法一样。

需求搞定，我们看现实有的条件服务端通信我们有的能力为socket，这个是已经封装在linux内核之中， JAVA对linux内核通信又进行了封装，有了自己的

Socket ServerSocket 通信，同时在JAVA Nio中又提出了异步方式的IO。

好，我们有的资源和需要达到的目标都已经有了，下面是实现中间件来弥补两者之间的鸿沟。

首先从客户端来看。客户端调用服务端的服务，肯定需要底层通信处理，而且这些通信处理需要集中处理，不能每次远程调用，都需重新处理一遍底层连接。

有什么方法可以达到这个目的么？动态代理。

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

一般我们看到的动态代理的invoke()方法中总会有 method.invoke(ac, arg); 这句代码。而上面代码中却没有，这是为什么呢？其实使用 method.invoke(ac, arg); 是在本地JVM中调用；

在客户端这边并没有Proxy对象，我们需要到服务找到对应的对象然后调用相应的方法。在hadoop中，是将数据发送给服务端，服务端将处理的结果再返回给客户端，所以这里的invoke()方法必然需要进行网络通信。

到这里我们可以再一次从图形来表示一下我们需要达到的目标。

下面这句代码就是服务端真正的调用相应方法的语句，其中的instance对象，是运行在服务端的对象，call是客户端传递过来的参数。通过反射机制，进行方法调用。

Object value = method.invoke(instance, call.getParameters());

上面我们把大体的框架搭建起来了，下面一步步进行细节分析。

在上面所示的invok()方法中，最终调用的方法为

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

我们需求分析 Client端是如何通过这两个方法调用了远程服务器的方法，并且获取返回值得。

需要解决的三个问题是

客户端和服务端的连接是怎样建立的？、
. 客户端是怎样给服务端发送数据的？
客户端是怎样获取服务端的返回数据的？
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对象

先看看是怎么获得一个到服务端的连接吧，下面贴出ipc.Client类中的getConnection()方法。

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

下面贴出Client.Connection类中的setupIOstreams()方法：

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

再有一步我们就知道客户端的连接是怎么建立的啦，下面贴出Client.Connection类中的setupConnection()方法：

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

终于，我们知道了客户端的连接是怎样建立的了，其实就是创建一个普通的socket进行通信。

问题2：客户端是怎样给服务端发送数据的？

第一句为了完成连接的建立，我们已经分析完毕；而第二句是为了发送数据，呵呵，分析下去，看能不能解决我们的问题呢。下面贴出Client.Connection类的sendParam()方法吧：

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

问题3：客户端是怎样获取服务端的返回数据的？

，当连接建立时会启动一个线程用于处理服务端返回的数据，我们看看这个处理线程是怎么实现的吧，下面贴出Client.Connection类和Client.Call类中的相关方法吧：

方法一：
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等待线程
}

客户端的代码分析就到这里，我们可以发现，客户端使用普通的socket 连接把客户端的方法调用名称参数（形参和实参）传递到服务端了。

下面分析服务端的代码。

对于ipc.Server，我们先分析一下它的几个内部类吧：

Call ：用于存储客户端发来的请求
Listener ：监听类，用于监听客户端发来的请求，同时Listener内部还有一个静态类，Listener.Reader，当监听器监听到用户请求，便让Reader读取用户请求。
Responder ：响应RPC请求类，请求处理完毕，由Responder发送给请求客户端。
Connection ：连接类，真正的客户端请求读取逻辑在这个类中。
Handler ：请求处理类，会循环阻塞读取callQueue中的call对象，并对其进行操作。

你会发现其实ipc.Server是一个abstract修饰的抽象类。那随之而来的问题就是：hadoop是怎样初始化RPC的Server端的呢？Namenode初始化时一定初始化了RPC的Sever端，那我们去看看Namenode的初始化源码吧：

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);
}
1. this.serviceRpcServer = RPC.getServer(this, dnSocketAddr.getHostName(),
dnSocketAddr.getPort(), serviceHandlerCount,

这里面我们需要重点关注的是这个上面这个方法，可以看到这里面传递过去的第一个参数是this .我们在前面说服务端最终是需要调用在服务端的某个对象来实际运行方法的。

现在这个this对象，及namenode对象就是服务端的相应对象。我们就有疑问，那么客户端有那么多接口，namenode都实现了相应的对象么？是的都实现了。这也好理解，客户端

会调用什么方法，肯定都是服务端和客户端事先约定好的，服务端肯定把相应的对象创建好了来等待客户端的调用。我们可以看一下namenode实现的端口，就很明晰了。

public class NameNode implements ClientProtocol, DatanodeProtocol,
NamenodeProtocol, FSConstants,
RefreshAuthorizationPolicyProtocol,
RefreshUserMappingsProtocol {

下面我们来分析服务端是如何处理请求的。

分析过ipc.Client源码后，我们知道Client端的底层通信直接采用了阻塞式IO编程。但hadoop是单中心结构，所以服务端不可以这么做，而是采用了java NIO来实现Server端，那Server端采用java NIO是怎么建立连接的呢？分析源码得知，Server端采用Listener监听客户端的连接，下面先分析一下Listener的构造函数吧：

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

在启动Listener线程时，服务端会一直等待客户端的连接，下面贴出Server.Listener类的run()方法：

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) {
???
}

下面贴出Server.Listener类中doAccept ()方法中的关键源码吧：

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

当reader被唤醒，reader接着执行doRead()方法。

下面贴出Server.Listener.Reader类中的doRead()方法和Server.Connection类中的readAndProcess()方法源码：

方法一：
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;
}
}

获得call对象
下面贴出Server.Connection类中的processOneRpc()方法和processData()方法的源码

方法一：
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请求的计数
}

处理call对象
你还记得Server类中还有个Handler内部类吗？呵呵，对call对象的处理就是它干的。下面贴出Server.Handler类中run()方法中的关键代码：

while (running) {
try {
final Call call = callQueue.take(); //弹出call，可能会阻塞
???
//调用ipc.Server类中的call()方法，但该call()方法是抽象方法，具体实现在RPC.Server类中
value = call(call.connection.protocol, call.param, call.timestamp);
synchronized (call.connection.responseQueue) {
setupResponse(buf, call,
(error == null) ? Status.SUCCESS : Status.ERROR,
value, errorClass, error);
???
//给客户端响应请求
responder.doRespond(call);
}
}

终于看到了call 方法我们下面看看服务端实际的call方法是怎么执行的吧

public Writable call(Class<?> protocol, Writable param, long receivedTime)
throws IOException {
try {
Invocation call = (Invocation)param;
if (verbose) log("Call: " + call);
Method method =
protocol.getMethod(call.getMethodName(),
call.getParameterClasses());
method.setAccessible(true);
long startTime = System.currentTimeMillis();
Object value = method.invoke(instance, call.getParameters());

最后一句我们发现实际上是用了反射。反射中的那个实际对象 instance 就是在namenode起来的时候创建的namenode对象。