RPC的實(shí)現(xiàn)

1. RPC客戶端?

2. RPC服務(wù)端

RPC客戶端的實(shí)現(xiàn)

RPC客戶端和RPC服務(wù)器端需要一個(gè)相同的接口類老速，RPC客戶端通過一個(gè)代理類來調(diào)用RPC服務(wù)器端的函數(shù)

RpcConsumerImpl的實(shí)現(xiàn) ...... 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();

因?yàn)閏onsumer對(duì)象是通過代理生成的诬像，所以當(dāng)consumer調(diào)用的時(shí)候咖熟，就會(huì)調(diào)用invoke函數(shù)，我們就可以把這次本地的函數(shù)調(diào)用的信息通過網(wǎng)絡(luò)發(fā)送到RPC服務(wù)器然后等待服務(wù)器返回的信息后再返回。

服務(wù)器實(shí)現(xiàn)

RPC服務(wù)器主要是在收到RPC客戶端之后解析出RPC調(diào)用的接口名除破，函數(shù)名以及參數(shù)。

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; /** * 處理服務(wù)器收到的RPC請(qǐng)求并返回結(jié)果 * @author sei.zz * */ public class RpcRequestHandler extends ChannelInboundHandlerAdapter { //對(duì)應(yīng)每個(gè)請(qǐng)求ID和端口好 對(duì)應(yīng)一個(gè)RpcContext的Map; private static Map<String,Map<String,Object>> ThreadLocalMap=new HashMap<String, Map<String,Object>>(); //服務(wù)端接口-實(shí)現(xiàn)類的映射表 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);//把客戶端的加進(jìn)來 ThreadLocalMap.put(host, local);//放回去 for(Map.Entry<String, Object> entry:map.entrySet()){ //更新變量 RpcContext.addProp(entry.getKey(), entry.getValue()); } } else { ThreadLocalMap.put(host, map); //把對(duì)應(yīng)線程的Context更新 for(Map.Entry<String, Object> entry:map.entrySet()){ RpcContext.addProp(entry.getKey(), entry.getValue()); } } } //用來緩存住需要序列化的結(jié)果 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 獲取接口名 函數(shù)名 參數(shù) 找到實(shí)現(xiàn)類 反射實(shí)現(xiàn) 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); } /** * 運(yùn)行調(diào)用的函數(shù)返回結(jié)果 * @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);//通過類名找到實(shí)現(xiàn)的類 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函數(shù)通過Java的反射機(jī)制琼腔，找到要調(diào)用的接口類然后調(diào)用對(duì)應(yīng)函數(shù)然后執(zhí)行瑰枫，然后返回結(jié)果到客戶端，本次RPC調(diào)用結(jié)束丹莲。