Hadoop RPC源码阅读-服务端Server

Hadoop版本Hadoop2.6

RPC主要分为3个部分：（1）交互协议 （2）客户端（3）服务端

（3）服务端

RPC服务端的实例代码：

public class Starter {
public static void main(String[] args) throws IOException {
RPC.Builder build = new RPC.Builder(new Configuration());
build.setBindAddress("localhost").setPort(10000).setProtocol(LoginServiceInterface.class).setInstance(new LoginServiceImpl());
RPC.Server server = build.build();
server.start();
}
}

RPC 服务端主要通过NIO来处理客户端发来的请求。

RPC服务端涉及的类主要有

org.apache.hadoop.ipc.Server（抽象类，server的最顶层类，与客户端的链接，响应，数据传输）

org.apache.hadoop.ipc.RPC.Server（RPC的内部类，是一个抽象类，主要涉及将请求交给动态代理）

org.apache.hadoop.ipc.WritableRpcEngine.Server（WritableRpcEngine的内部类，是一个静态类，动态代理的具体实现，得到请求结果）

先看ipc.Server类，它有几个主要的内部类分别是：

Call ：用于存储客户端发来的请求

Listener ： 监听类，用于监听客户端发来的请求，同时Listener内部还有一个静态类，Listener.Reader，当监听器监听到用户请求，便让Reader读取用户请求。

Responder ：响应RPC请求类，请求处理完毕，由Responder发送给请求客户端。

Connection ：连接类，真正的客户端请求读取逻辑在这个类中。

Handler ：请求处理类，会循环阻塞读取callQueue中的call对象，并对其进行操作。

（1）启动服务

如上述服务端实例代码所示，当调用函数start()，RPC服务端就启动起来了，我们先看看start()里面有什么

/** Starts the service.  Must be called before any calls will be handled. */
//该实现在ipc.Server类中
public synchronized void start() {
responder.start();
listener.start();
handlers = new Handler[handlerCount];

for (int i = 0; i < handlerCount; i++) {
handlers[i] = new Handler(i);
handlers[i].start();
}
}

可以看出，Server端通过启动Listener监听客户端发来的请求，启动responder响应客户端发来的请求，启动多个线程Handler循环阻塞读取请求交给responder响应。

先看看如何监听客户端发来的请求

（2）监听客户端的连接

Listner初始化，构造函数如下所示，通过Java NIO创建一个ServerSocketChannel监听本地地址和端口，并设置成非阻塞模式，并注册成接听客户端连接事件，并启动多个Reader线程，同时将读客户端请求数据的事件交给Reader实现。

public Listener() throws IOException {
address = new InetSocketAddress(bindAddress, port);
// Create a new server socket and set to non blocking mode
acceptChannel = ServerSocketChannel.open();
acceptChannel.configureBlocking(false);

// Bind the server socket to the local host and port
bind(acceptChannel.socket(), address, backlogLength, conf, portRangeConfig);
port = acceptChannel.socket().getLocalPort(); //Could be an ephemeral port
// create a selector;
selector= Selector.open();
readers = new Reader[readThreads];
for (int i = 0; i < readThreads; i++) {
Reader reader = new Reader(
"Socket Reader #" + (i + 1) + " for port " + port);
readers[i] = reader;
reader.start();
}

// Register accepts on the server socket with the selector.
acceptChannel.register(selector, SelectionKey.OP_ACCEPT);
this.setName("IPC Server listener on " + port);
this.setDaemon(true);
}

Listener:Listener是一个Thread类的子类，通过start()启动该线程，并运行该线程的run()方法：

public void run() {
LOG.info(Thread.currentThread().getName() + ": starting");
SERVER.set(Server.this);
connectionManager.startIdleScan();
while (running) {
SelectionKey key = null;
try {
getSelector().select();
Iterator<SelectionKey> iter = getSelector().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) {
// we can run out of memory if we have too many threads
// log the event and sleep for a minute and give
// some thread(s) a chance to finish
LOG.warn("Out of Memory in server select", e);
closeCurrentConnection(key, e);
connectionManager.closeIdle(true);
try { Thread.sleep(60000); } catch (Exception ie) {}
} catch (Exception e) {
closeCurrentConnection(key, e);
}
}
LOG.info("Stopping " + Thread.currentThread().getName());

synchronized (this) {
try {
acceptChannel.close();
selector.close();
} catch (IOException e) { }

selector= null;
acceptChannel= null;

// close all connections
connectionManager.stopIdleScan();
connectionManager.closeAll();
}
}

//对连接事件进行处理
void doAccept(SelectionKey key) throws InterruptedException, IOException,  OutOfMemoryError {
ServerSocketChannel server = (ServerSocketChannel) key.channel();
SocketChannel channel;
while ((channel = server.accept()) != null) {

channel.configureBlocking(false);
channel.socket().setTcpNoDelay(tcpNoDelay);
channel.socket().setKeepAlive(true);

Reader reader = getReader();
Connection c = connectionManager.register(channel);//把与客户端连接的管道封装成Connection
key.attach(c);  // so closeCurrentConnection can get the object
reader.addConnection(c);//将客户端请求连接的通道加入到Reader，监听用户发来的请求。
}
}

Reader:Reader也是Thread类的子类，通过start()启动该线程，并运行该线程的run()方法：

　　　　public void run() {
LOG.info("Starting " + Thread.currentThread().getName());
try {
doRunLoop();
} finally {
try {
readSelector.close();
} catch (IOException ioe) {
LOG.error("Error closing read selector in " + Thread.currentThread().getName(), ioe);
}
}
}

private synchronized void doRunLoop() {
while (running) {
SelectionKey key = null;
try {
// consume as many connections as currently queued to avoid
// unbridled acceptance of connections that starves the select
int size = pendingConnections.size();//在Listner类中通过reader.addConnection(c)加入到该阻塞队列中
for (int i=size; i>0; i--) {//遍历处理用户的连接事件，并监听用户发来的请求
Connection conn = pendingConnections.take();
conn.channel.register(readSelector, SelectionKey.OP_READ, conn);
}
readSelector.select();

Iterator<SelectionKey> iter = readSelector.selectedKeys().iterator();
while (iter.hasNext()) {
key = iter.next();
iter.remove();
if (key.isValid()) {
if (key.isReadable()) {//判断该事件是不是读事件，并处理该读事件
doRead(key);
}
}
key = null;
}
} catch (InterruptedException e) {
if (running) {                      // unexpected -- log it
LOG.info(Thread.currentThread().getName() + " unexpectedly interrupted", e);
}
} catch (IOException ex) {
LOG.error("Error in Reader", ex);
}
}
}

void doRead(SelectionKey key) throws InterruptedException {
int count = 0;
Connection c = (Connection)key.attachment();//因为Connection也保存着与客户端的连接,因此这里提取了Connection,把处理细节交给Connection
if (c == null) {
return;
}
c.setLastContact(Time.now());

try {
count = c.readAndProcess();
} catch (InterruptedException ieo) {
LOG.info(Thread.currentThread().getName() + ": readAndProcess caught InterruptedException", ieo);
throw ieo;
} catch (Exception e) {
// a WrappedRpcServerException is an exception that has been sent
// to the client, so the stacktrace is unnecessary; any other
// exceptions are unexpected internal server errors and thus the
// stacktrace should be logged
LOG.info(Thread.currentThread().getName() + ": readAndProcess from client " +
c.getHostAddress() + " threw exception [" + e + "]",
(e instanceof WrappedRpcServerException) ? null : e);
count = -1; //so that the (count < 0) block is executed
}
if (count < 0) {
closeConnection(c);
c = null;
}
else {
c.setLastContact(Time.now());
}
}

Connection类中通过channelRead(channel, data)读取客户端发送的数据，并将读取的数据通过processOneRpc(data.array())方法处理逻辑过程，

processOneRpc通过调用processRpcRequest(RpcRequestHeaderProto header,DataInputStream dis)方法，

在processRpcRequest方法中封装成Call数据对象，并加入callQueue.put(call)队列中。

（3）实现客户端的请求的服务，得到客户端请求的结果数据

Handler类实现客户端请求的服务，通过从callQueue队列获取客户端RPC请求Call对象，并调用抽象方法call处理请求。

抽象方法call的实现在WritableRpcEngine类的内部类Server类，Server类是继承RPC.Server，是RPC服务调用的具体实现并定义相关协议

下面是call方法的具体实现：

public Writable call(org.apache.hadoop.ipc.RPC.Server server,
String protocolName, Writable rpcRequest, long receivedTime)
throws IOException, RPC.VersionMismatch {

Invocation call = (Invocation)rpcRequest;
if (server.verbose) log("Call: " + call);

// Verify writable rpc version
if (call.getRpcVersion() != writableRpcVersion) {
// Client is using a different version of WritableRpc
throw new RpcServerException(
"WritableRpc version mismatch, client side version="
+ call.getRpcVersion() + ", server side version="
+ writableRpcVersion);
}

long clientVersion = call.getProtocolVersion();
final String protoName;
ProtoClassProtoImpl protocolImpl;
if (call.declaringClassProtocolName.equals(VersionedProtocol.class.getName())) {
// VersionProtocol methods are often used by client to figure out
// which version of protocol to use.
//
// Versioned protocol methods should go the protocolName protocol
// rather than the declaring class of the method since the
// the declaring class is VersionedProtocol which is not
// registered directly.
// Send the call to the highest  protocol version
VerProtocolImpl highest = server.getHighestSupportedProtocol(
RPC.RpcKind.RPC_WRITABLE, protocolName);
if (highest == null) {
throw new RpcServerException("Unknown protocol: " + protocolName);
}
protocolImpl = highest.protocolTarget;
} else {
protoName = call.declaringClassProtocolName;

// Find the right impl for the protocol based on client version.
ProtoNameVer pv =
new ProtoNameVer(call.declaringClassProtocolName, clientVersion);
protocolImpl =
server.getProtocolImplMap(RPC.RpcKind.RPC_WRITABLE).get(pv);
if (protocolImpl == null) { // no match for Protocol AND Version
VerProtocolImpl highest =
server.getHighestSupportedProtocol(RPC.RpcKind.RPC_WRITABLE,
protoName);
if (highest == null) {
throw new RpcServerException("Unknown protocol: " + protoName);
} else { // protocol supported but not the version that client wants
throw new RPC.VersionMismatch(protoName, clientVersion,
highest.version);
}
}
}

// Invoke the protocol method
long startTime = Time.now();
int qTime = (int) (startTime-receivedTime);
Exception exception = null;
try {
Method method =
protocolImpl.protocolClass.getMethod(call.getMethodName(),
call.getParameterClasses());
method.setAccessible(true);
server.rpcDetailedMetrics.init(protocolImpl.protocolClass);
Object value =
method.invoke(protocolImpl.protocolImpl, call.getParameters());
if (server.verbose) log("Return: "+value);
return new ObjectWritable(method.getReturnType(), value);

} catch (InvocationTargetException e) {
Throwable target = e.getTargetException();
if (target instanceof IOException) {
exception = (IOException)target;
throw (IOException)target;
} else {
IOException ioe = new IOException(target.toString());
ioe.setStackTrace(target.getStackTrace());
exception = ioe;
throw ioe;
}
} catch (Throwable e) {
if (!(e instanceof IOException)) {
LOG.error("Unexpected throwable object ", e);
}
IOException ioe = new IOException(e.toString());
ioe.setStackTrace(e.getStackTrace());
exception = ioe;
throw ioe;
} finally {
int processingTime = (int) (Time.now() - startTime);
if (LOG.isDebugEnabled()) {
String msg = "Served: " + call.getMethodName() +
" queueTime= " + qTime +
" procesingTime= " + processingTime;
if (exception != null) {
msg += " exception= " + exception.getClass().getSimpleName();
}
LOG.debug(msg);
}
String detailedMetricsName = (exception == null) ?
call.getMethodName() :
exception.getClass().getSimpleName();
server.rpcMetrics.addRpcQueueTime(qTime);
server.rpcMetrics.addRpcProcessingTime(processingTime);
server.rpcDetailedMetrics.addProcessingTime(detailedMetricsName,
processingTime);
}
}

View Code
通过调用RPC服务得到客户端请求的结果value，使用Responder类将结果返回给客户端。下面是Handler类run()方法中部分代码

　　　　　　CurCall.set(null);//处理完call请求，故将当前处理call表示标志为Null
synchronized (call.connection.responseQueue) {
// setupResponse() needs to be sync'ed together with
// responder.doResponse() since setupResponse may use
// SASL to encrypt response data and SASL enforces
// its own message ordering.
setupResponse(buf, call, returnStatus, detailedErr,
value, errorClass, error);//将返回值封装到输出流中

// Discard the large buf and reset it back to smaller size
// to free up heap
if (buf.size() > maxRespSize) {
LOG.warn("Large response size " + buf.size() + " for call "
+ call.toString());
buf = new ByteArrayOutputStream(INITIAL_RESP_BUF_SIZE);
}
responder.doRespond(call);//将返回请求call加入队列中，然后在Responder类中一一处理。

}

（4）返回客户端的请求结果

Responder类：负责返回客户端请求的结果，通过NIO注册OP_WRITE写事件，将结果返回给客户端。

每处理完一个请求，就会调用Responder中doRespond方法处理请求结果。

void doRespond(Call call) throws IOException {
synchronized (call.connection.responseQueue) {
call.connection.responseQueue.addLast(call);
if (call.connection.responseQueue.size() == 1) {
processResponse(call.connection.responseQueue, true);
}
}
}

可以看到上述方法又调用processResponse方法处理请求结果，这里将请求结果对象call注册OP_WRITE写事件，通过NIO返回给客户端；

channel.register(writeSelector, SelectionKey.OP_WRITE, call);

而Responder的run方法和doRunLoop方法检测OP_WRITE写事件，并通过doAsyncWrite方法处理该事件

public void run() {
LOG.info(Thread.currentThread().getName() + ": starting");
SERVER.set(Server.this);
try {
doRunLoop();
} finally {
LOG.info("Stopping " + Thread.currentThread().getName());
try {
writeSelector.close();
} catch (IOException ioe) {
LOG.error("Couldn't close write selector in " + Thread.currentThread().getName(), ioe);
}
}
}

private void doRunLoop() {
long lastPurgeTime = 0;   // last check for old calls.

while (running) {
try {
waitPending();     // If a channel is being registered, wait.
writeSelector.select(PURGE_INTERVAL);
Iterator<SelectionKey> iter = writeSelector.selectedKeys().iterator();
while (iter.hasNext()) {
SelectionKey key = iter.next();
iter.remove();
try {
if (key.isValid() && key.isWritable()) {
doAsyncWrite(key);
}
} catch (IOException e) {
LOG.info(Thread.currentThread().getName() + ": doAsyncWrite threw exception " + e);
}
}
long now = Time.now();
if (now < lastPurgeTime + PURGE_INTERVAL) {
continue;
}
lastPurgeTime = now;
//
// If there were some calls that have not been sent out for a
// long time, discard them.
//
if(LOG.isDebugEnabled()) {
LOG.debug("Checking for old call responses.");
}
ArrayList<Call> calls;

// get the list of channels from list of keys.
synchronized (writeSelector.keys()) {
calls = new ArrayList<Call>(writeSelector.keys().size());
iter = writeSelector.keys().iterator();
while (iter.hasNext()) {
SelectionKey key = iter.next();
Call call = (Call)key.attachment();
if (call != null && key.channel() == call.connection.channel) {
calls.add(call);
}
}
}

for(Call call : calls) {
doPurge(call, now);
}
} catch (OutOfMemoryError e) {
//
// we can run out of memory if we have too many threads
// log the event and sleep for a minute and give
// some thread(s) a chance to finish
//
LOG.warn("Out of Memory in server select", e);
try { Thread.sleep(60000); } catch (Exception ie) {}
} catch (Exception e) {
LOG.warn("Exception in Responder", e);
}
}
}

而doAsyncWrite方法还是通过调用processResponse方法处理结果，processResponse方法调用channelWrite将结果返回给客户端，当不能一次性返回时，在processResponse方法里将返回结果再次注册OP_WRITE写事件，因而形成一个循环使得数据能全部返回给客户端。