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

通过调用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写事件,因而形成一个循环使得数据能全部返回给客户端。

上一篇:mysql数据库存储引擎及区别


下一篇:ZeroMQ接口函数之 :zmq_tcp – 使用TCP协议的ØMQ网络单播协议