RPC的实现

1. RPC客户端 

2. RPC服务端

RPC客户端的实现

RPC客户端和RPC服务器端需要一个相同的接口类，RPC客户端通过一个代理类来调用RPC服务器端的函数

RpcConsumerImpl的实现 ...... package com.alibaba.middleware.race.rpc.api.impl; import java.lang.reflect.InvocationHandler; import java.lang.reflect.Method; import java.lang.reflect.Proxy; import java.util.ArrayList; import java.util.HashMap; import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.UUID; import java.util.concurrent.atomic.AtomicLong; import com.alibaba.middleware.race.rpc.aop.ConsumerHook; import com.alibaba.middleware.race.rpc.api.RpcConsumer; import com.alibaba.middleware.race.rpc.async.ResponseCallbackListener; import com.alibaba.middleware.race.rpc.context.RpcContext; import com.alibaba.middleware.race.rpc.model.RpcRequest; import com.alibaba.middleware.race.rpc.model.RpcResponse; import com.alibaba.middleware.race.rpc.netty.RpcConnection; import com.alibaba.middleware.race.rpc.netty.RpcNettyConnection; import com.alibaba.middleware.race.rpc.tool.Tool; public class RpcConsumerImpl extends RpcConsumer implements InvocationHandler { private static AtomicLong callTimes = new AtomicLong(0L); private RpcConnection connection; private List<RpcConnection> connection_list; private Map<String,ResponseCallbackListener> asyncMethods; private Class<?> interfaceClass; private String version; private int timeout; private ConsumerHook hook; public Class<?> getInterfaceClass() { return interfaceClass; } public String getVersion() { return version; } public int getTimeout() { this.connection.setTimeOut(timeout); return timeout; } public ConsumerHook getHook() { return hook; } RpcConnection select() { //Random rd=new Random(System.currentTimeMillis()); int d=(int) (callTimes.getAndIncrement()%(connection_list.size()+1)); if(d==0) return connection; else { return connection_list.get(d-1); } } public RpcConsumerImpl() { //String ip=System.getProperty("SIP"); String ip="127.0.0.1"; this.asyncMethods=new HashMap<String,ResponseCallbackListener>(); this.connection=new RpcNettyConnection(ip,8888); this.connection.connect(); connection_list=new ArrayList<RpcConnection>(); int num=Runtime.getRuntime().availableProcessors()/3 -2; for (int i = 0; i < num; i++) { connection_list.add(new RpcNettyConnection(ip, 8888)); } for (RpcConnection conn:connection_list) { conn.connect(); } } public void destroy() throws Throwable { if (null != connection) { connection.close(); } } @SuppressWarnings("unchecked") public <T> T proxy(Class<T> interfaceClass) throws Throwable { if (!interfaceClass.isInterface()) { throw new IllegalArgumentException(interfaceClass.getName() + " is not an interface"); } return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(), new Class<?>[] { interfaceClass }, this); } @Override public RpcConsumer interfaceClass(Class<?> interfaceClass) { // TODO Auto-generated method stub this.interfaceClass=interfaceClass; return this; } @Override public RpcConsumer version(String version) { // TODO Auto-generated method stub this.version=version; return this; } @Override public RpcConsumer clientTimeout(int clientTimeout) { // TODO Auto-generated method stub this.timeout=clientTimeout; return this; } @Override public RpcConsumer hook(ConsumerHook hook) { // TODO Auto-generated method stub this.hook=hook; return this; } @Override public Object instance() { // TODO Auto-generated method stub try { return proxy(this.interfaceClass); } catch (Throwable e) { e.printStackTrace(); } return null; } @Override public void asynCall(String methodName) { // TODO Auto-generated method stub asynCall(methodName, null); } @Override public <T extends ResponseCallbackListener> void asynCall( String methodName, T callbackListener) { this.asyncMethods.put(methodName, callbackListener); this.connection.setAsyncMethod(asyncMethods); for (RpcConnection conn:connection_list) { conn.setAsyncMethod(asyncMethods); } } @Override public void cancelAsyn(String methodName) { // TODO Auto-generated method stub this.asyncMethods.remove(methodName); this.connection.setAsyncMethod(asyncMethods); for (RpcConnection conn:connection_list) { conn.setAsyncMethod(asyncMethods); } } @Override public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { // TODO Auto-generated method stub List<String> parameterTypes = new LinkedList<String>(); for (Class<?> parameterType : method.getParameterTypes()) { parameterTypes.add(parameterType.getName()); } RpcRequest request = new RpcRequest(); request.setRequestId(UUID.randomUUID().toString()); request.setClassName(method.getDeclaringClass().getName()); request.setMethodName(method.getName()); request.setParameterTypes(method.getParameterTypes()); request.setParameters(args); if(hook!=null) hook.before(request); RpcResponse response = null; try { request.setContext(RpcContext.props); response = (RpcResponse) select().Send(request,asyncMethods.containsKey(request.getMethodName())); if(hook!=null) hook.after(request); if(!asyncMethods.containsKey(request.getMethodName())&&response.getExption()!=null) { Throwable e=(Throwable) Tool.deserialize(response.getExption(),response.getClazz()); throw e.getCause(); } } catch (Throwable t) { //t.printStackTrace(); //throw new RuntimeException(t); throw t; } finally { // if(asyncMethods.containsKey(request.getMethodName())&&asyncMethods.get(request.getMethodName())!=null) // { // cancelAsyn(request.getMethodName()); // } } if(response==null) { return null; } else if (response.getErrorMsg() != null) { throw response.getErrorMsg(); } else { return response.getAppResponse(); } } }

RpcConsumer consumer; consumer = (RpcConsumer) getConsumerImplClass().newInstance(); consumer.someMethod();

因为consumer对象是通过代理生成的，所以当consumer调用的时候，就会调用invoke函数，我们就可以把这次本地的函数调用的信息通过网络发送到RPC服务器然后等待服务器返回的信息后再返回。

服务器实现

RPC服务器主要是在收到RPC客户端之后解析出RPC调用的接口名，函数名以及参数。

package com.alibaba.middleware.race.rpc.api.impl; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; import java.lang.reflect.Method; import java.util.HashMap; import java.util.Map; import net.sf.cglib.reflect.FastClass; import net.sf.cglib.reflect.FastMethod; import com.alibaba.middleware.race.rpc.context.RpcContext; import com.alibaba.middleware.race.rpc.model.RpcRequest; import com.alibaba.middleware.race.rpc.model.RpcResponse; import com.alibaba.middleware.race.rpc.serializer.KryoSerialization; import com.alibaba.middleware.race.rpc.tool.ByteObjConverter; import com.alibaba.middleware.race.rpc.tool.ReflectionCache; import com.alibaba.middleware.race.rpc.tool.Tool; /** * 处理服务器收到的RPC请求并返回结果 * @author sei.zz * */ public class RpcRequestHandler extends ChannelInboundHandlerAdapter { //对应每个请求ID和端口好 对应一个RpcContext的Map; private static Map<String,Map<String,Object>> ThreadLocalMap=new HashMap<String, Map<String,Object>>(); //服务端接口-实现类的映射表 private final Map<String, Object> handlerMap; KryoSerialization kryo=new KryoSerialization(); public RpcRequestHandler(Map<String, Object> handlerMap) { this.handlerMap = handlerMap; } @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println("active"); } @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { // TODO Auto-generated method stub System.out.println("disconnected"); } //更新RpcContext的类容 private void UpdateRpcContext(String host,Map<String,Object> map) { if(ThreadLocalMap.containsKey(host)) { Map<String,Object> local=ThreadLocalMap.get(host); local.putAll(map);//把客户端的加进来 ThreadLocalMap.put(host, local);//放回去 for(Map.Entry<String, Object> entry:map.entrySet()){ //更新变量 RpcContext.addProp(entry.getKey(), entry.getValue()); } } else { ThreadLocalMap.put(host, map); //把对应线程的Context更新 for(Map.Entry<String, Object> entry:map.entrySet()){ RpcContext.addProp(entry.getKey(), entry.getValue()); } } } //用来缓存住需要序列化的结果 private static Object cacheName=null; private static Object cacheVaule=null; @Override public void channelRead( ChannelHandlerContext ctx, Object msg) throws Exception { RpcRequest request=(RpcRequest)msg; String host=ctx.channel().remoteAddress().toString(); //更新上下文 UpdateRpcContext(host,request.getContext()); //TODO 获取接口名 函数名 参数 找到实现类 反射实现 RpcResponse response = new RpcResponse(); response.setRequestId(request.getRequestId()); try { Object result = handle(request); if(cacheName!=null&&cacheName.equals(result)) { response.setAppResponse(cacheVaule); } else { response.setAppResponse(ByteObjConverter.ObjectToByte(result)); cacheName=result; cacheVaule=ByteObjConverter.ObjectToByte(result); } } catch (Throwable t) { //response.setErrorMsg(t); response.setExption(Tool.serialize(t)); response.setClazz(t.getClass()); } ctx.writeAndFlush(response); } /** * 运行调用的函数返回结果 * @param request * @return * @throws Throwable */ private static RpcRequest methodCacheName=null; private static Object methodCacheValue=null; private Object handle(RpcRequest request) throws Throwable { String className = request.getClassName(); Object classimpl = handlerMap.get(className);//通过类名找到实现的类 Class<?> clazz = classimpl.getClass(); String methodName = request.getMethodName(); Class<?>[] parameterTypes = request.getParameterTypes(); Object[] parameters = request.getParameters(); // Method method = ReflectionCache.getMethod(clazz.getName(),methodName, parameterTypes); // method.setAccessible(true); //System.out.println(className+":"+methodName+":"+parameters.length); if(methodCacheName!=null&&methodCacheName.equals(request)) { return methodCacheValue; } else { try { methodCacheName=request; if(methodMap.containsKey(methodName)) { methodCacheValue= methodMap.get(methodName).invoke(classimpl, parameters); return methodCacheValue; } else { FastClass serviceFastClass = FastClass.create(clazz); FastMethod serviceFastMethod = serviceFastClass.getMethod(methodName, parameterTypes); methodMap.put(methodName, serviceFastMethod); methodCacheValue= serviceFastMethod.invoke(classimpl, parameters); return methodCacheValue; } //return method.invoke(classimpl, parameters); } catch (Throwable e) { throw e.getCause(); } } } private Map<String,FastMethod> methodMap=new HashMap<String, FastMethod>(); @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ctx.flush(); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { //ctx.close(); //cause.printStackTrace(); ctx.close(); } }

handel函数通过Java的反射机制，找到要调用的接口类然后调用对应函数然后执行，然后返回结果到客户端，本次RPC调用结束。