java ipc实例，仿照hadoop ipc写的实例
2.client端用动态代理作调用远程ipc接口方法
3.server端用反射，执行ipc接口方法，并返回给client端接口方法返回值
?
hadoop ipc的另一个特点是server端用三个角色，Listener，Handler，Responser。server聚合这三个角色
Listener：nio socket获取请求CALL对象,放入队列中
Handler：从队列中获取CALL对象，执行ipc接口方法
Responser：被Handler调用，用nio socket返回接口方法返回值
?
简单实例：（实例来自github）
1.定义协议
class="java">public interface Echo {
public String who() throws IOException;;
public void from(String name) throws IOException;;
}
?2.定义代理
/*Invocation封装方法名和参数*/
public Object invoke(Object proxy, Method method, Object[] args)
throws Throwable {
final boolean logDebug = LOG.isDebugEnabled();
long startTime = 0;
if (logDebug) {
startTime = System.currentTimeMillis();
}
//调用Client call方法
Invocation value = client.call(new Invocation(iface, method, args),
remoteId);//调用远程的当前方法，阻塞直到server返回值
if (logDebug) {
long callTime = System.currentTimeMillis() - startTime;
LOG.debug("Call: " + method.getName() + "() " + callTime);
}
return value.getResult();
}
?
public Invocation call(Invocation invoked, ConnectionId remoteId)
throws InterruptedException, IOException {
Call call = new Call(invoked); //将传入的数据封装成call对象 Serializable接口
//已经向服务器端 RPCHeader ConnectionHeader验证
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) {
// save the fact that we were interrupted
interrupted = true;
}
}
if (interrupted) {
//因中断异常而终止，设置标志interrupted为true
Thread.currentThread().interrupt();
}
if (call.error != null) {
if (call.error instanceof RemoteException) {
call.error.fillInStackTrace();
throw call.error;
} else {
/* local exception use the connection because it will
* reflect an ip change, unlike the remoteId
*/
throw wrapException(connection.getRemoteAddress(),
call.error);
}
} else {
return call.value; //返回结果数据
}
}
}
?
3.server端反射ipc接口方法
private class Handler extends Thread {
public Handler(int instanceNumber) {
this.setDaemon(true);
this.setName("IPC Server handler " + instanceNumber + " on " + port);
}
@Override
public void run() {
LOG.info(getName() + ": starting");
SERVER.set(Server.this);
//创建大小为10240个字节的响应缓冲区
ByteArrayOutputStream buf = new ByteArrayOutputStream(
INITIAL_RESP_BUF_SIZE);
while (running) {
try {
/** pop the queue; maybe blocked here */
//获取一个远程调用请求 Server.Call
final Call call = callQueue.take(); //弹出call，可能会阻塞
if (LOG.isDebugEnabled())
LOG.debug(getName() + ": has #" + call.id + " from "
+ call.connection);
String errorClass = null;
String error = null;
Serializable value = null;
/** process the current call. */
//处理当前Call
CurCall.set(call);
try {
//调用ipc.Server类中的call()方法，但该call()方法是抽象方法，具体实现在RPC.Server类中
value = call(call.connection.protocol, call.param,
call.timestamp); //Invocation对象
} catch (Throwable e) {
LOG.info(getName() + ", call " + call + ": error: " + e, e);
errorClass = e.getClass().getName();
error = e.getMessage();
}
CurCall.set(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.
*/
//将返回结果序列化到Call的成员变量response中
setupResponse(
buf,
call,
(error == null) ? Status.SUCCESS : Status.ERROR,
value, errorClass, error);
/* Discard the large buf and reset it back to smaller size to freeup heap*/
//丢弃大的buf 重设到更小的容量 释放内存
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);//Responder在Server构造器初始化
}
} catch (InterruptedException e) {
if (running) { // unexpected -- log it
LOG.info(getName() + " caught: " + e);
}
} catch (Exception e) {
LOG.info(getName() + " caught: " + e);
}
}
LOG.info(getName() + ": exiting");
}
}
?
?
public Serializable call(Class<?> iface, Serializable param,
long receivedTime) throws IOException {
try {
Invocation call = (Invocation) param; //调用参数 Invocationd对象包含方法名称 形式参数列表和实际参数列表
if (verbose)
log("Call: " + call);
//从实例缓存中按照接口寻找实例对象
Object instance = INSTANCE_CACHE.get(iface);
if (instance == null)
throw new IOException("interface `" + iface + "` not inscribe.");
//通过Class对象获取Method对象
Method method = iface.getMethod(call.getMethodName(),
call.getParameterClasses());
//取消Java语言访问权限检查
method.setAccessible(true);
long startTime = System.currentTimeMillis();
//调用Method对象的invoke方法
Object value = method.invoke(instance, call.getParameters());
int processingTime = (int) (System.currentTimeMillis() - startTime);
int qTime = (int) (startTime - receivedTime);
if (LOG.isDebugEnabled()) {
LOG.debug("Served: " + call.getMethodName()
+ " queueTime= " + qTime + " procesingTime= "
+ processingTime);
}
if (verbose)
log("Return: " + value);
call.setResult(value); //向Invocation对象设置结果
return call;
} catch (InvocationTargetException e) {
Throwable target = e.getTargetException();
if (target instanceof IOException) {
throw (IOException) target;
} else {
IOException ioe = new IOException(target.toString());
ioe.setStackTrace(target.getStackTrace());
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());
throw ioe;
}
}
}
?
?
?
?
?
?
?
?
?
?
?
?
?
?
相关文章
